From e44b2c129da1374029b2fefc954e4ba243662f19 Mon Sep 17 00:00:00 2001
From: Igor Artamonov <igor@artamonov.ru>
Date: Tue, 21 Sep 2021 20:33:53 -0400
Subject: [PATCH] problem: doesn't execute calls after reconnect

---
 .../upstream/ethereum/EthereumWsFactory.kt    |  9 ++++++-
 .../ethereum/EthereumWsFactoryRealSpec.groovy | 24 +++++++++++++++++++
 2 files changed, 32 insertions(+), 1 deletion(-)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt
index 0321155f..28785e8a 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt
+++ b/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(
                     {
diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy
index ca2a3b59..bf66d4e9 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy
+++ b/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))
+    }
 }