Merge pull request #41 from namjug-kim/develop

namjug-kim · web-flow · commit 3f446de6166e · 2019-05-23T03:02:52.000+09:00
Develop
diff --git a/reactive-crypto-coineal/src/main/kotlin/com/njkim/reactivecrypto/coineal/CoinealRawWebsocketClient.kt b/reactive-crypto-coineal/src/main/kotlin/com/njkim/reactivecrypto/coineal/CoinealRawWebsocketClient.kt
@@ -22,6 +22,8 @@ import com.njkim.reactivecrypto.core.common.model.currency.CurrencyPair
 import com.njkim.reactivecrypto.coineal.model.CoinealMessageFrame
 import com.njkim.reactivecrypto.coineal.model.CoinealOrderBook
 import com.njkim.reactivecrypto.coineal.model.CoinealTickDataWrapper
+import com.njkim.reactivecrypto.core.common.util.toEpochMilli
+import com.njkim.reactivecrypto.core.netty.HeartBeatHandler
 import io.netty.buffer.ByteBuf
 import io.netty.buffer.ByteBufInputStream
 import io.netty.channel.ChannelHandlerContext
@@ -34,6 +36,8 @@ import reactor.core.publisher.Flux
 import reactor.core.publisher.toFlux
 import reactor.netty.http.client.HttpClient
 import java.nio.charset.Charset
+import java.time.ZonedDateTime
+import java.util.concurrent.TimeUnit
 import java.util.zip.GZIPInputStream
 
 class CoinealRawWebsocketClient(
@@ -51,7 +55,20 @@ class CoinealRawWebsocketClient(
             .toFlux()
 
         return HttpClient.create()
-            .tcpConfiguration { tcp -> tcp.doOnConnected { connection -> connection.addHandler(GzipDecoder()) } }
+            .tcpConfiguration { tcp ->
+                tcp.doOnConnected { connection ->
+                    connection.addHandler(GzipDecoder())
+                    connection.addHandler(
+                        "heartBeat",
+                        HeartBeatHandler(
+                            false,
+                            10,
+                            TimeUnit.SECONDS,
+                            5
+                        ) { "{\"ping\": ${ZonedDateTime.now().toEpochMilli()}}" }
+                    )
+                }
+            }
             .websocket()
             .uri(baseUri)
             .handle { inbound, outbound ->
@@ -73,7 +90,20 @@ class CoinealRawWebsocketClient(
             .toFlux()
 
         return HttpClient.create()
-            .tcpConfiguration { tcp -> tcp.doOnConnected { connection -> connection.addHandler(GzipDecoder()) } }
+            .tcpConfiguration { tcp ->
+                tcp.doOnConnected { connection ->
+                    connection.addHandler(GzipDecoder())
+                    connection.addHandler(
+                        "heartBeat",
+                        HeartBeatHandler(
+                            false,
+                            10,
+                            TimeUnit.SECONDS,
+                            5
+                        ) { "{\"ping\": ${ZonedDateTime.now().toEpochMilli()}}" }
+                    )
+                }
+            }
             .websocket()
             .uri(baseUri)
             .handle { inbound, outbound ->
diff --git a/reactive-crypto-core/src/main/kotlin/com/njkim/reactivecrypto/core/common/exception/HeartBeatFailException.kt b/reactive-crypto-core/src/main/kotlin/com/njkim/reactivecrypto/core/common/exception/HeartBeatFailException.kt
@@ -0,0 +1,19 @@
+/*
+ * Copyright 2019 namjug-kim
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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 com.njkim.reactivecrypto.core.common.exception
+
+class HeartBeatFailException(message: String) : RuntimeException(message)
diff --git a/reactive-crypto-core/src/main/kotlin/com/njkim/reactivecrypto/core/common/model/currency/Currency.kt b/reactive-crypto-core/src/main/kotlin/com/njkim/reactivecrypto/core/common/model/currency/Currency.kt
@@ -35,7 +35,9 @@ enum class Currency {
     SWC, LINDA, FML, GUNTHY, VIN, GOB, EVY, MGO, FUNDZ, UQC, CTC, MRPH, PCCM,
     PXL, IOTW, NRC, ZAT, POE, APL, VEX, VITAE, FTX, ABTC, TSL, SCRL, NIX, TMC, MCAN,
     LGD, ZEST, ALP, CXP, ZING, UCN, RET, WIRE, GUBI, LAD, EVNS, CAPP, ZBB, KZE, DBX, NTY,
-    GFC, GMB, FLC, BTD,
+    GFC, GMB, FLC, BTD, SKM, MALL, RFOX, FN, IZI, NVL, YTA, IPWT, AT, HVCC, CL,
+    OA, VSS, STCC, CTA, CPX, KDH, LAMB, IFX, GCS, BRC, O2P, BHD, IVO, BTMC, BKBT,
+    TCLB, NEAL,
 
     M19,
 
diff --git a/reactive-crypto-core/src/main/kotlin/com/njkim/reactivecrypto/core/netty/HeartBeatHandler.kt b/reactive-crypto-core/src/main/kotlin/com/njkim/reactivecrypto/core/netty/HeartBeatHandler.kt
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2019 namjug-kim
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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 com.njkim.reactivecrypto.core.netty
+
+import com.njkim.reactivecrypto.core.common.exception.HeartBeatFailException
+import io.netty.channel.ChannelHandlerContext
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
+import io.netty.handler.timeout.IdleState
+import io.netty.handler.timeout.IdleStateEvent
+import io.netty.handler.timeout.IdleStateHandler
+import java.util.concurrent.TimeUnit
+
+/**
+ * @property pingTime ping message will be triggered when no write was performed for the specified
+ * period of time.  Specify {@code 0} to disable.
+ */
+class HeartBeatHandler(
+    observeOutput: Boolean,
+    private val pingTime: Long,
+    unit: TimeUnit,
+    pongTimeout: Long,
+    private val pingMessage: () -> String
+) : IdleStateHandler(observeOutput, pingTime + pongTimeout, pingTime, 0, unit) {
+
+    @Throws(Exception::class)
+    override fun channelIdle(ctx: ChannelHandlerContext, evt: IdleStateEvent) {
+        if (evt.state() == IdleState.READER_IDLE) {
+            ctx.close()
+            throw HeartBeatFailException("")
+        } else if (evt.state() == IdleState.WRITER_IDLE || evt.state() == IdleState.ALL_IDLE) {
+            ctx.channel().writeAndFlush(TextWebSocketFrame(pingMessage()))
+        }
+    }
+}
diff --git a/reactive-crypto-huobikorea/src/main/kotlin/com/njkim/reactivecrypto/huobikorea/HuobiKoreaWebsocketClient.kt b/reactive-crypto-huobikorea/src/main/kotlin/com/njkim/reactivecrypto/huobikorea/HuobiKoreaWebsocketClient.kt
@@ -25,6 +25,7 @@ import com.njkim.reactivecrypto.core.common.model.currency.CurrencyPair
 import com.njkim.reactivecrypto.core.common.model.order.OrderBook
 import com.njkim.reactivecrypto.core.common.model.order.TickData
 import com.njkim.reactivecrypto.core.common.util.toEpochMilli
+import com.njkim.reactivecrypto.core.netty.HeartBeatHandler
 import com.njkim.reactivecrypto.huobikorea.model.HuobiKoreaTickDataWrapper
 import com.njkim.reactivecrypto.huobikorea.model.HuobiOrderBook
 import com.njkim.reactivecrypto.huobikorea.model.HuobiSubscribeResponse
@@ -40,6 +41,7 @@ import reactor.core.publisher.Flux
 import reactor.netty.http.client.HttpClient
 import java.nio.charset.Charset
 import java.time.ZonedDateTime
+import java.util.concurrent.TimeUnit
 import java.util.zip.GZIPInputStream
 import kotlin.streams.toList
 
@@ -62,7 +64,20 @@ class HuobiKoreaWebsocketClient : AbstractExchangeWebsocketClient() {
 
         return HttpClient.create()
             .wiretap(log.isDebugEnabled)
-            .tcpConfiguration { tcp -> tcp.doOnConnected { connection -> connection.addHandler(GzipDecoder()) } }
+            .tcpConfiguration { tcp ->
+                tcp.doOnConnected { connection ->
+                    connection.addHandler(GzipDecoder())
+                    connection.addHandler(
+                        "heartBeat",
+                        HeartBeatHandler(
+                            false,
+                            2000,
+                            TimeUnit.MILLISECONDS,
+                            1000
+                        ) { "{\"ping\": ${ZonedDateTime.now().toEpochMilli()}}" }
+                    )
+                }
+            }
             .websocket()
             .uri(baseUri)
             .handle { inbound, outbound ->
@@ -93,7 +108,20 @@ class HuobiKoreaWebsocketClient : AbstractExchangeWebsocketClient() {
 
         return HttpClient.create()
             .wiretap(log.isDebugEnabled)
-            .tcpConfiguration { tcp -> tcp.doOnConnected { connection -> connection.addHandler(GzipDecoder()) } }
+            .tcpConfiguration { tcp ->
+                tcp.doOnConnected { connection ->
+                    connection.addHandler(GzipDecoder())
+                    connection.addHandler(
+                        "heartBeat",
+                        HeartBeatHandler(
+                            false,
+                            2000,
+                            TimeUnit.MILLISECONDS,
+                            1000
+                        ) { "{\"ping\": ${ZonedDateTime.now().toEpochMilli()}}" }
+                    )
+                }
+            }
             .websocket()
             .uri(baseUri)
             .handle { inbound, outbound ->
diff --git a/reactive-crypto-idax/src/main/kotlin/com/njkim/reactivecrypto/idax/IdaxRawWebsocketClient.kt b/reactive-crypto-idax/src/main/kotlin/com/njkim/reactivecrypto/idax/IdaxRawWebsocketClient.kt
@@ -19,14 +19,10 @@ package com.njkim.reactivecrypto.idax
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.kotlin.readValue
 import com.njkim.reactivecrypto.core.common.model.currency.CurrencyPair
+import com.njkim.reactivecrypto.core.netty.HeartBeatHandler
 import com.njkim.reactivecrypto.idax.model.IdaxMessageFrame
 import com.njkim.reactivecrypto.idax.model.IdaxOrderBook
 import com.njkim.reactivecrypto.idax.model.IdaxTickData
-import io.netty.channel.ChannelHandlerContext
-import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
-import io.netty.handler.timeout.IdleState
-import io.netty.handler.timeout.IdleStateEvent
-import io.netty.handler.timeout.IdleStateHandler
 import mu.KotlinLogging
 import reactor.core.publisher.Flux
 import reactor.core.publisher.toFlux
@@ -62,8 +58,10 @@ class IdaxRawWebsocketClient(
         return HttpClient.create()
             .tcpConfiguration { tcp ->
                 tcp.doOnConnected {
-                    // okex websocket 은 30초동안 응답이 없는경우 해당 연결을 끊는다. 이를 막기 위해서 20초마다 ping 요청을 보낸다.
-                    it.addHandlerFirst("heartBeat", HeartBeatHandler(false, 2, 0, 0, TimeUnit.SECONDS))
+                    it.addHandlerFirst(
+                        "heartBeat",
+                        HeartBeatHandler(false, 2, TimeUnit.SECONDS, 1) { "{\"event\":\"ping\"}" }
+                    )
                 }
             }
             .websocket()
@@ -96,8 +94,10 @@ class IdaxRawWebsocketClient(
         return HttpClient.create()
             .tcpConfiguration { tcp ->
                 tcp.doOnConnected {
-                    // okex websocket 은 30초동안 응답이 없는경우 해당 연결을 끊는다. 이를 막기 위해서 20초마다 ping 요청을 보낸다.
-                    it.addHandlerFirst("heartBeat", HeartBeatHandler(false, 2, 0, 0, TimeUnit.SECONDS))
+                    it.addHandlerFirst(
+                        "heartBeat",
+                        HeartBeatHandler(false, 2, TimeUnit.SECONDS, 1) { "{\"event\":\"ping\"}" }
+                    )
                 }
             }
             .websocket()
@@ -138,20 +138,4 @@ class IdaxRawWebsocketClient(
             }
             .map { objectMapper.readValue<IdaxMessageFrame<List<IdaxOrderBook>>>(it) }
     }
-
-    private inner class HeartBeatHandler(
-        observeOutput: Boolean,
-        readerIdleTime: Long,
-        writerIdleTime: Long,
-        allIdleTime: Long,
-        unit: TimeUnit
-    ) : IdleStateHandler(observeOutput, readerIdleTime, writerIdleTime, allIdleTime, unit) {
-
-        @Throws(Exception::class)
-        override fun channelIdle(ctx: ChannelHandlerContext, evt: IdleStateEvent) {
-            if (evt.state() == IdleState.READER_IDLE) {
-                ctx.channel().writeAndFlush(TextWebSocketFrame("{\"event\":\"ping\"}"))
-            }
-        }
-    }
 }
