From 52840a1d4528c603414638d0b698452db5bf68dc Mon Sep 17 00:00:00 2001
From: Igor Artamonov <igor@artamonov.ru>
Date: Tue, 19 Oct 2021 17:12:36 -0400
Subject: [PATCH] solution: native subscription for "syncing"

---
 .../dshackle/upstream/UpstreamAvailability.kt |  2 +-
 .../upstream/ethereum/EthereumSubscribe.kt    |  5 ++
 .../ethereum/subscribe/ConnectSyncing.kt      | 59 +++++++++++++++++++
 3 files changed, 65 insertions(+), 1 deletion(-)
 create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt
index 3717b52f..4bed644c 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt
@@ -23,7 +23,7 @@ enum class UpstreamAvailability(val grpcId: Int) {
      */
     OK(1),
     /**
-     * Good node, but is still synchronizing a latest block
+     * Good node, but is still synchronizing to a latest block
      */
     LAGGING(2),
     /**
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt
index 3c3afab9..bb00f6a1 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt
@@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectNewHeads
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectSyncing
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ProduceLogs
 import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.hex.Hex32
@@ -18,6 +19,7 @@ open class EthereumSubscribe(
 
     private val newHeads = ConnectNewHeads(upstream)
     private val logs = ConnectLogs(upstream)
+    private val syncing = ConnectSyncing(upstream)
 
     @Suppress("UNCHECKED_CAST")
     open fun subscribe(method: String, params: Any?): Flux<out Any> {
@@ -36,6 +38,9 @@ open class EthereumSubscribe(
             }
             return logs.start(paramsMap.address, paramsMap.topics)
         }
+        if (method == "syncing") {
+            return syncing.connect()
+        }
         return Flux.error(UnsupportedOperationException("Method $method is not supported"))
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt
new file mode 100644
index 00000000..bc224d56
--- /dev/null
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt
@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import java.time.Duration
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+class ConnectSyncing(
+        private val upstream: EthereumMultistream
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ConnectSyncing::class.java)
+    }
+
+    private var connected: Flux<Boolean>? = null
+    private val connectLock = ReentrantLock()
+
+    fun connect(): Flux<Boolean> {
+        val current = connected
+        if (current != null) {
+            return current
+        }
+        connectLock.withLock {
+            val currentRecheck = connected
+            if (currentRecheck != null) {
+                return currentRecheck
+            }
+            val created = upstream.observeStatus()
+                    .map { it != UpstreamAvailability.OK }
+                    .publish()
+                    .refCount(1, Duration.ofSeconds(60))
+                    .doFinally {
+                        //forget it on disconnect, so next time it's recreated
+                        connected = null
+                    }
+            connected = created
+            return created
+        }
+    }
+}
\ No newline at end of file