problem: doesn't execute calls after reconnect

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt

@@ -92,7 +92,7 @@ class EthereumWsFactory(
                 .many()
                 .multicast()
                 .directBestEffort<BlockContainer>()
-        private val rpcSend = Sinks
+        private var rpcSend = Sinks
                 .many()
                 .unicast()
                 .onBackpressureBuffer<JsonRpcRequest>()
@@ -120,6 +120,13 @@ class EthereumWsFactory(
             if (alreadyReconnecting) {
                 return
             }
+            // rpcSend is already CANCELLED, since the subscription owned by the previous connection is gone
+            // so we need to create a new Sink. Emit Complete is probably useless, and just in case
+            rpcSend.tryEmitComplete()
+            rpcSend = Sinks
+                    .many()
+                    .unicast()
+                    .onBackpressureBuffer<JsonRpcRequest>()
             log.info("Reconnect to $uri in $retryInterval seconds...")
             Global.control.schedule(
                     {

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy

@@ -105,4 +105,28 @@ class EthereumWsFactoryRealSpec extends Specification {
         act[0].value.contains("\"params\":[\"newHeads\"]")
     }
 
+    def "Call after reconnect"() {
+        when:
+        conn.connect()
+        conn.retryInterval = 2
+        Thread.sleep(SLEEP)
+        server.stop()
+        Thread.sleep(SLEEP)
+        server = new MockWSServer(port)
+        server.start()
+        // reconnects in 2 seconds, give 1 extra
+        Thread.sleep(3_000)
+
+        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+        then:
+        StepVerifier.create(resp)
+                .then {
+                    server.reply('{"jsonrpc":"2.0", "id":100, "result": "baz"}')
+                }
+                .expectNextMatches {
+                    it.hasResult() && it.resultAsProcessedString == "baz"
+                }
+                .expectComplete()
+                .verify(Duration.ofSeconds(3))
+    }
 }