handle socket by lifecycle

Author: jaeyunn15

app/src/main/java/com/jeremy/thunder/MainActivity.kt

@@ -39,7 +39,11 @@ class MainActivity : ComponentActivity() {
                     }
 
                     LaunchedEffect(key1 = Unit) {
-                        viewModel.observeResponse()
+                        viewModel.observeAllMarket()
+                    }
+
+                    LaunchedEffect(key1 = Unit) {
+                        viewModel.observeTicker()
                     }
 
                     val state = viewModel.response.collectAsState()

app/src/main/java/com/jeremy/thunder/MainViewModel.kt

@@ -6,34 +6,50 @@ import com.jeremy.thunder.socket.BinanceRequest
 import com.jeremy.thunder.socket.SocketService
 import com.jeremy.thunder.socket.Ticker
 import dagger.hilt.android.lifecycle.HiltViewModel
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
 import javax.inject.Inject
 
 @HiltViewModel
 class MainViewModel @Inject constructor(
     private val service: SocketService
-): ViewModel() {
+) : ViewModel() {
 
     private val _response = MutableStateFlow<Ticker?>(null)
     val response: StateFlow<Ticker?> = _response.asStateFlow()
 
     fun request() {
         viewModelScope.launch {
-            delay(4000) // thunder state observer가 생성 되기 이전 임으로 임의 딜레이
-            service.request(request = BinanceRequest())
+            // 특정 2개의 티커만 조회
+            service.request(
+                request = BinanceRequest(
+                    params = listOf(
+                        "btcusdt@markPrice",
+                        "ethusdt@markPrice"
+                    )
+                )
+            )
         }
     }
 
-    fun observeResponse() {
-        service.observeTicker().onEach { result ->
-            _response.update { result.data }
+    fun requestAllMarket() {
+        //전체 마켓 데이터 조회
+        service.request(request = BinanceRequest(params = listOf("!markPrice@arr")))
+    }
+
+    fun observeAllMarket() {
+        service.observeAllMarkets().onEach {
+
+        }.launchIn(viewModelScope)
+    }
+
+    fun observeTicker() {
+        service.observeTicker().onEach {
+
         }.launchIn(viewModelScope)
     }
 }

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

@@ -0,0 +1,17 @@
+package com.jeremy.thunder.socket
+
+import com.google.gson.annotations.SerializedName
+
+data class AllMarketResponse(
+    @SerializedName("id")
+    val id: Int,
+
+    @SerializedName("result")
+    val result: Any? = null,
+
+    @SerializedName("stream")
+    val stream: String = "",
+
+    @SerializedName("data")
+    val data: List<Ticker> = emptyList()
+)

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

@@ -3,5 +3,7 @@ package com.jeremy.thunder.socket
 data class BinanceRequest(
     val id: Int = 1,
     val method: String = "SUBSCRIBE",
-    val params: List<String> = listOf("btcusdt@markPrice", "ethusdt@markPrice")
+    val params: List<String> = listOf("btcusdt@markPrice")
 )
+
+//val params: List<String> = listOf("btcusdt@markPrice", "ethusdt@markPrice")

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

@@ -10,4 +10,7 @@ interface SocketService {
 
     @Receive
     fun observeTicker(): Flow<TickerResponse>
+
+    @Receive
+    fun observeAllMarkets(): Flow<AllMarketResponse>
 }

thunder-core/src/main/java/com/jeremy/thunder/ThunderState.kt

@@ -1,16 +1,44 @@
 package com.jeremy.thunder
 
+/**
+ * Socket Thunder State
+ * 기존 Scarlet 라이브러리에선 Event 상태를 체크해서 활성화 상태일 때만 Send를 할 수 있는 구조였기에 CONNECTED 상태 외에 모두 캐시 처리하고 저장하는 것이 필요
+* */
 sealed interface ThunderState {
+
+    /*
+    * 초기 소켓 상태
+    * */
     object IDLE : ThunderState
 
+    /*
+    * 연결 중인 상태
+    * Send 요청 시, Cache 처리
+    * */
     object CONNECTING : ThunderState
 
+    /*
+    * 연결 된 상태
+    * Send 요청 시, Cache 체크 후 데이터 전송
+    * */
     object CONNECTED : ThunderState
 
+    /*
+    * 연결을 끊고 있는 상태
+    * Send 요청 시, 무시 (Cache Clear)
+    * */
     object DISCONNECTING : ThunderState
 
+    /*
+    * 연결이 끊긴 상태
+    * Send 요청 시, 무시 (Cache Clear)
+    * */
     object DISCONNECTED : ThunderState
 
+    /*
+    * 에러가 난 상태
+    * Send 요청 시, Cache 처리
+    * */
     data class ERROR(
         val error: ThunderError
     ) : ThunderState

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

@@ -6,13 +6,22 @@ import okhttp3.OkHttpClient
 import okhttp3.Request
 
 fun OkHttpClient.makeWebSocketCore(url: String): WebSocket.Factory {
-    val socketListener = SocketListener()
-    newWebSocket(
-        request = Request.Builder().url(url).build(),
-        listener = socketListener
-    )
     return OkHttpWebSocket.Factory(
-        socketListener = socketListener,
+        provider = ConnectionProvider(this, url),
         scope = CoroutineScope(SupervisorJob())
     )
+}
+
+interface SocketListenerProvider {
+    fun provide(socketListener: SocketListener)
+}
+
+class ConnectionProvider(
+    private val okHttpClient: OkHttpClient,
+    private val url: String
+) : SocketListenerProvider {
+
+    override fun provide(socketListener: SocketListener) {
+        okHttpClient.newWebSocket(Request.Builder().url(url).build(), socketListener)
+    }
 }

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

@@ -8,31 +8,31 @@ import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.onStart
 
 
 class OkHttpWebSocket internal constructor(
+    private val provider: ConnectionProvider,
     private val socketListener: SocketListener,
     private val socketHandler: SocketHandler,
     private val scope: CoroutineScope
 ) : WebSocket {
 
     private val _event = MutableStateFlow<com.jeremy.thunder.event.WebSocketEvent?>(null)
-
-    init {
-        socketListener.collectEvent().onEach {
+    override fun open() {
+        socketListener.collectEvent().onStart {
+            provider.provide(socketListener)
+        }.onEach {
+            _event.tryEmit(it)
             when (it) {
-                is com.jeremy.thunder.event.WebSocketEvent.OnConnectionOpen -> {
-                    socketHandler.open(it.webSocket as okhttp3.WebSocket)
+                is WebSocketEvent.OnConnectionOpen -> {
+                    socketHandler.initWebSocket(it.webSocket as okhttp3.WebSocket)
                 }
-                else -> _event.tryEmit(it)
+                else -> Unit
             }
         }.launchIn(scope)
     }
 
-    override fun open(webSocket: okhttp3.WebSocket) {
-        // ?
-    }
-
     override fun events(): Flow<com.jeremy.thunder.event.WebSocketEvent> {
         return _event.asSharedFlow().filterNotNull()
     }
@@ -54,12 +54,13 @@ class OkHttpWebSocket internal constructor(
     }
 
     class Factory(
-        private val socketListener: SocketListener,
+        private val provider: ConnectionProvider,
         private val scope: CoroutineScope
     ) : WebSocket.Factory {
         override fun create(): WebSocket =
             OkHttpWebSocket(
-                socketListener = socketListener,
+                provider = provider ,
+                socketListener = SocketListener(),
                 socketHandler = SocketHandler(),
                 scope = scope
             )

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

@@ -7,8 +7,12 @@ class SocketHandler : WebSocket {
 
     private var socket: okhttp3.WebSocket? = null
 
-    override fun open(webSocket: okhttp3.WebSocket) {
-        socket = webSocket
+    fun initWebSocket(socket: okhttp3.WebSocket) {
+        this.socket = socket
+    }
+
+    override fun open() {
+
     }
 
     override fun events(): Flow<com.jeremy.thunder.event.WebSocketEvent> {
@@ -21,10 +25,12 @@ class SocketHandler : WebSocket {
 
     override fun close(code: Int, reason: String) {
         socket?.close(code, reason)
+        socket = null
     }
 
     override fun cancel() {
         socket?.cancel()
+        socket = null
     }
 
     override fun error(t: String) {

thunder/src/main/java/com/jeremy/thunder/Converter.kt

@@ -6,14 +6,15 @@ import com.google.gson.reflect.TypeToken
 import java.io.StringReader
 import java.lang.reflect.Type
 
-interface Converter <T> {
-    fun convert(data: String) : T
+interface Converter<T> {
+    fun convert(data: String): T
 }
 
 class ConvertAdapter<T> private constructor(
     private val gson: Gson,
-    private val typeAdapter: TypeAdapter<T>
-): Converter<T> {
+    private val typeAdapter: TypeAdapter<T>,
+    private val type: Type
+) : Converter<T> {
 
     override fun convert(data: String): T {
         val jsonReader = gson.newJsonReader(StringReader(data))
@@ -23,7 +24,7 @@ class ConvertAdapter<T> private constructor(
     class Factory {
         fun create(type: Type): ConvertAdapter<*> {
             val typeAdapter = Gson().getAdapter(TypeToken.get(type))
-            return ConvertAdapter(Gson(), typeAdapter)
+            return ConvertAdapter(Gson(), typeAdapter, type)
         }
     }
 }

thunder/src/main/java/com/jeremy/thunder/EventMapper.kt

@@ -0,0 +1,34 @@
+package com.jeremy.thunder
+
+import com.jeremy.thunder.event.WebSocketEvent
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.map
+
+/**
+ * mapEvent 내에서 convert가 실패하면 다른 응답이기에 실패했다고 봐야 하는지?
+* */
+
+class EventMapper<T> constructor(
+    private val converter: Converter<T>
+) {
+
+    fun mapEvent(flow: Flow<WebSocketEvent>): Flow<T?> = flow.filter {
+        it is WebSocketEvent.OnMessageReceived
+    }.map {
+        (it as WebSocketEvent.OnMessageReceived).data
+    }.map {
+        try {
+            val result = converter.convert(it)
+            result
+        } catch (e: Exception) {
+            null
+        }
+    }
+
+    class Factory {
+        fun create(converter: Converter<*>): EventMapper<*> {
+            return EventMapper(converter)
+        }
+    }
+}