Modify OnMessageReceived bytestring as Charset.defaultCharset() string

Author: jaeyunn15

app/src/main/java/com/jeremy/thunder/di/SocketModule.kt

@@ -12,6 +12,7 @@ import dagger.hilt.android.qualifiers.ApplicationContext
 import dagger.hilt.components.SingletonComponent
 import okhttp3.OkHttpClient
 import okhttp3.logging.HttpLoggingInterceptor
+import java.util.concurrent.TimeUnit
 import javax.inject.Singleton
 
 @Module
@@ -25,6 +26,10 @@ internal object SocketModule {
             .setLevel(HttpLoggingInterceptor.Level.BODY)
         return OkHttpClient.Builder()
             .addInterceptor(httpLoggingInterceptor)
+            .pingInterval(
+                10,
+                TimeUnit.SECONDS
+            ) // If there are no events for a minute, we need to put in some code for ping pong to output a socket connection error from okhttp.
             .build()
     }
 
@@ -35,7 +40,8 @@ internal object SocketModule {
         @ApplicationContext context: Context
     ): SocketService {
         return thunder {
-            webSocketCore(okHttpClient.makeWebSocketCore("wss://fstream.binance.com/stream"))
+//            webSocketCore(okHttpClient.makeWebSocketCore("wss://fstream.binance.com/stream")) // binance socket server
+            webSocketCore(okHttpClient.makeWebSocketCore("wss://api.upbit.com/websocket/v1")) // upbit socket server
             setApplicationContext(context)
             setConverterType(ConverterType.Gson)
         }.create()

app/src/main/java/com/jeremy/thunder/socket/SocketService.kt

@@ -5,6 +5,8 @@ import com.jeremy.thunder.socket.model.AllMarketResponse
 import com.jeremy.thunder.socket.model.AllMarketTickerResponse
 import com.jeremy.thunder.socket.model.BinanceRequest
 import com.jeremy.thunder.socket.model.TickerResponse
+import com.jeremy.thunder.socket.model.UpbitRequest
+import com.jeremy.thunder.socket.model.UpbitTickerResponse
 import com.jeremy.thunder.ws.Event
 import com.jeremy.thunder.ws.Receive
 import com.jeremy.thunder.ws.Send
@@ -14,6 +16,12 @@ interface SocketService {
     @Send
     fun request(request: BinanceRequest)
 
+    @Send
+    fun requestUpbit(request: List<UpbitRequest>)
+
+    @Receive
+    fun collectUpbitTicker(): Flow<UpbitTickerResponse>
+
     @Event
     fun receiveEvent(): Flow<WebSocketEvent>
 

app/src/main/java/com/jeremy/thunder/socket/model/BinanceRequest.kt

@@ -6,4 +6,20 @@ data class BinanceRequest(
     val params: List<String> = listOf("btcusdt@markPrice")
 )
 
-//val params: List<String> = listOf("btcusdt@markPrice", "ethusdt@markPrice")
+//val params: List<String> = listOf("btcusdt@markPrice", "ethusdt@markPrice")
+
+//[{"ticket":"test"},{"type":"trade","codes":["KRW-BTC","BTC-BCH"]},{"format":"SIMPLE"}]
+sealed interface UpbitRequest
+
+data class RequestTicketField(
+    val ticket: String
+): UpbitRequest
+
+data class RequestTypeField(
+    val type: String,
+    val codes: List<String>
+): UpbitRequest
+
+data class RequestFormatField(
+    val format: String = "DEFAULT"
+): UpbitRequest

app/src/main/java/com/jeremy/thunder/socket/model/UpbitResponse.kt

@@ -0,0 +1,39 @@
+package com.jeremy.thunder.socket.model
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class UpbitTickerResponse(
+    val acc_ask_volume: Double,
+    val acc_bid_volume: Double,
+    val acc_trade_price: Double,
+    val acc_trade_price_24h: Double,
+    val acc_trade_volume: Double,
+    val acc_trade_volume_24h: Double,
+    val ask_bid: String,
+    val change: String,
+    val change_price: Double,
+    val change_rate: Double,
+    val code: String,
+    val high_price: Double,
+    val highest_52_week_date: String,
+    val highest_52_week_price: Double,
+    val is_trading_suspended: Boolean,
+    val low_price: Double,
+    val lowest_52_week_date: String,
+    val lowest_52_week_price: Double,
+    val market_state: String,
+    val market_warning: String,
+    val opening_price: Double,
+    val prev_closing_price: Double,
+    val signed_change_price: Double,
+    val signed_change_rate: Double,
+    val stream_type: String,
+    val timestamp: Long,
+    val trade_date: String,
+    val trade_price: Double,
+    val trade_time: String,
+    val trade_timestamp: Long,
+    val trade_volume: Double,
+    val type: String
+)

app/src/main/java/com/jeremy/thunder/ui/home/HomeViewModel.kt

@@ -6,12 +6,16 @@ import com.jeremy.thunder.event.WebSocketEvent
 import com.jeremy.thunder.socket.SocketService
 import com.jeremy.thunder.socket.model.AllMarketTickerResponseItem
 import com.jeremy.thunder.socket.model.BinanceRequest
+import com.jeremy.thunder.socket.model.RequestFormatField
+import com.jeremy.thunder.socket.model.RequestTicketField
+import com.jeremy.thunder.socket.model.RequestTypeField
 import dagger.hilt.android.lifecycle.HiltViewModel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.update
+import java.util.UUID
 import javax.inject.Inject
 
 @HiltViewModel
@@ -46,6 +50,19 @@ class HomeViewModel @Inject constructor(
     }
 
     fun requestAllMarketTicker() {
+        // upbit socket request
+        service.requestUpbit(
+            listOf(
+                RequestTicketField(ticket = UUID.randomUUID().toString()),
+                RequestTypeField(
+                    type = "ticker",
+                    codes = listOf("KRW-BTC","KRW-ETH")
+                ),
+                RequestFormatField()
+            )
+        )
+
+        // binance socket request subscribe
         service.request(
             request = BinanceRequest(
                 method = "SUBSCRIBE",
@@ -54,6 +71,7 @@ class HomeViewModel @Inject constructor(
     }
 
     fun requestCancelAllMarketTicker() {
+        // binance socket request unsubscribe
         service.request(
             request = BinanceRequest(
                 method = "UNSUBSCRIBE",
@@ -63,6 +81,10 @@ class HomeViewModel @Inject constructor(
     }
 
     fun observeAllMarket() {
+        service.collectUpbitTicker().onEach {
+            // this is upbit flow
+        }.launchIn(viewModelScope)
+
         service.observeAllMarketTickers().onEach { response ->
             _allMarketTickerFlow.update { response.data.sortedBy { it.c } }
         }.launchIn(viewModelScope)

thunder-okhttp/src/main/java/com/jeremy/thunder/SocketListener.kt

@@ -1,5 +1,6 @@
 package com.jeremy.thunder
 
+import com.jeremy.thunder.event.WebSocketEvent
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
@@ -8,37 +9,44 @@ import okhttp3.Response
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
 import okio.ByteString
+import java.nio.charset.Charset
 
 class SocketListener : WebSocketListener(), EventCollector {
 
-    private val _eventFlow = MutableSharedFlow<com.jeremy.thunder.event.WebSocketEvent>(
+    private val _eventFlow = MutableSharedFlow<WebSocketEvent>(
         replay = 1,
         extraBufferCapacity = DEFAULT_BUFFER,
         onBufferOverflow = BufferOverflow.DROP_OLDEST
     )
 
-    override fun collectEvent(): Flow<com.jeremy.thunder.event.WebSocketEvent> {
+    override fun collectEvent(): Flow<WebSocketEvent> {
         return _eventFlow.asSharedFlow()
     }
 
     override fun onOpen(webSocket: WebSocket, response: Response) {
-        _eventFlow.tryEmit(com.jeremy.thunder.event.WebSocketEvent.OnConnectionOpen(webSocket))
+        _eventFlow.tryEmit(WebSocketEvent.OnConnectionOpen(webSocket))
     }
 
     override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
-        _eventFlow.tryEmit(com.jeremy.thunder.event.WebSocketEvent.OnMessageReceived(bytes.toString()))
+        _eventFlow.tryEmit(
+            WebSocketEvent.OnMessageReceived(
+                bytes.string(
+                    Charset.defaultCharset()
+                )
+            )
+        )
     }
 
     override fun onMessage(webSocket: WebSocket, text: String) {
-        _eventFlow.tryEmit(com.jeremy.thunder.event.WebSocketEvent.OnMessageReceived(text))
+        _eventFlow.tryEmit(WebSocketEvent.OnMessageReceived(text))
     }
 
     override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
-        _eventFlow.tryEmit(com.jeremy.thunder.event.WebSocketEvent.OnConnectionClosed)
+        _eventFlow.tryEmit(WebSocketEvent.OnConnectionClosed)
     }
 
     override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
-        _eventFlow.tryEmit(com.jeremy.thunder.event.WebSocketEvent.OnConnectionError(t.message))
+        _eventFlow.tryEmit(WebSocketEvent.OnConnectionError(t.message))
     }
 
     companion object {