Fix solana subs (#449)
This commit is contained in:
@@ -38,7 +38,7 @@ class GenericSubscriptionConnect(
|
|||||||
|
|
||||||
@Suppress("UNCHECKED_CAST")
|
@Suppress("UNCHECKED_CAST")
|
||||||
override fun createConnection(): Flux<Any> {
|
override fun createConnection(): Flux<Any> {
|
||||||
return conn.subscribe(ChainRequest(topic, ListParams(getParams(params))))
|
return conn.subscribe(ChainRequest(topic, ListParams(getParams(params) as List<Any>)))
|
||||||
.data
|
.data
|
||||||
.timeout(Duration.ofSeconds(60), Mono.empty())
|
.timeout(Duration.ofSeconds(60), Mono.empty())
|
||||||
.onErrorResume { Mono.empty() } as Flux<Any>
|
.onErrorResume { Mono.empty() } as Flux<Any>
|
||||||
|
|||||||
@@ -0,0 +1,36 @@
|
|||||||
|
package io.emeraldpay.dshackle.upstream.generic
|
||||||
|
|
||||||
|
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||||
|
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
|
||||||
|
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||||
|
import org.junit.jupiter.api.Test
|
||||||
|
import org.mockito.Mockito.verify
|
||||||
|
import org.mockito.kotlin.doReturn
|
||||||
|
import org.mockito.kotlin.mock
|
||||||
|
import reactor.core.publisher.Flux
|
||||||
|
import reactor.test.StepVerifier
|
||||||
|
import java.time.Duration
|
||||||
|
import java.util.concurrent.atomic.AtomicReference
|
||||||
|
|
||||||
|
class GenericSubscriptionConnectTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `test request param is flat list`() {
|
||||||
|
val param: List<Any> = listOf("all")
|
||||||
|
val topic = "topic"
|
||||||
|
val response = "hello".toByteArray()
|
||||||
|
val ws = mock<WsSubscriptions> {
|
||||||
|
on { subscribe(ChainRequest(topic, ListParams(param))) } doReturn
|
||||||
|
WsSubscriptions.SubscribeData(Flux.just(response), "", AtomicReference(""))
|
||||||
|
}
|
||||||
|
|
||||||
|
val genericSubscriptionConnect = GenericSubscriptionConnect(ws, topic, param)
|
||||||
|
|
||||||
|
StepVerifier.create(genericSubscriptionConnect.createConnection())
|
||||||
|
.expectNext(response)
|
||||||
|
.expectComplete()
|
||||||
|
.verify(Duration.ofSeconds(1))
|
||||||
|
|
||||||
|
verify(ws).subscribe(ChainRequest(topic, ListParams(param)))
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user