- refactor alpaca news ws service

- use case to get News

app/src/main/java/dev/pinkroom/marketsight/common/Resource.kt

@@ -3,7 +3,7 @@ package dev.pinkroom.marketsight.common
 import dev.pinkroom.marketsight.domain.model.ErrorMessage
 
 sealed class Resource<T>{
-    data class Success<T>(val data: T?): Resource<T>()
+    data class Success<T>(val data: T): Resource<T>()
     data class Error<T>(
         val message: String? = null,
         val errorInfo: ErrorMessage? = null,

app/src/main/java/dev/pinkroom/marketsight/data/data_source/AlpacaRemoteDataSource.kt

@@ -16,9 +16,7 @@ import dev.pinkroom.marketsight.data.mapper.toSubscriptionMessage
 import dev.pinkroom.marketsight.data.remote.AlpacaService
 import dev.pinkroom.marketsight.data.remote.model.dto.NewsMessageDto
 import dev.pinkroom.marketsight.data.remote.model.request.MessageAlpacaService
-import dev.pinkroom.marketsight.domain.model.NewsInfo
 import dev.pinkroom.marketsight.domain.model.SubscriptionMessage
-import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.take
@@ -29,11 +27,11 @@ class AlpacaRemoteDataSource @Inject constructor(
     private val alpacaService: AlpacaService,
     private val dispatchers: DispatcherProvider,
 ) {
-    fun subscribeNews(symbols: List<String> = listOf("*")): Flow<Resource<WebSocket.Event>> = flow<Resource<WebSocket.Event>> {
+    fun subscribeNews(symbols: List<String> = listOf("*")) = flow<Resource<WebSocket.Event>> {
         alpacaService.observeOnConnectionEvent().collect{
             when(it){
                 is WebSocket.Event.OnConnectionOpened<*> -> {
-                    alpacaService.sendSubscribe(message = MessageAlpacaService(action = ActionAlpaca.Subscribe.action, news = symbols))
+                    alpacaService.sendMessage(message = MessageAlpacaService(action = ActionAlpaca.Subscribe.action, news = symbols))
                 }
                 is WebSocket.Event.OnMessageReceived -> {
                     Log.d(TAG,"Received: ${it.message}")
@@ -53,21 +51,20 @@ class AlpacaRemoteDataSource @Inject constructor(
         }
     }.flowOn(dispatchers.IO)
 
-    fun getRealTimeNews(): Flow<Resource<List<NewsInfo>>> = flow {
+    fun getRealTimeNews() = flow {
         alpacaService.observeResponse().collect{ data ->
-            Log.d("TESTE","AQUI SEMPRE")
             val listNews = mutableListOf<NewsMessageDto>()
             data.forEach {
                 gson.toObject(value = it, helperIdentifier = HelperIdentifierMessagesAlpacaService.News)?.let { news ->
                     listNews.add(news)
                 }
             }
-            emit(Resource.Success(listNews.map { it.toNewsInfo() }))
+            if (listNews.isNotEmpty()) emit(listNews.map { it.toNewsInfo() })
         }
     }.flowOn(dispatchers.IO)
 
-    fun sendMessageToAlpacaService(message: MessageAlpacaService): Flow<Resource<SubscriptionMessage>> = flow<Resource<SubscriptionMessage>> {
-        alpacaService.sendSubscribe(message = message)
+    fun sendMessageToAlpacaService(message: MessageAlpacaService) = flow<Resource<SubscriptionMessage>> {
+        alpacaService.sendMessage(message = message)
         alpacaService.observeResponse().collect { data ->
             data.forEach {
                 gson.toObject(value = it, helperIdentifier = HelperIdentifierMessagesAlpacaService.Subscription)?.let { sub ->

app/src/main/java/dev/pinkroom/marketsight/data/remote/AlpacaService.kt

@@ -11,7 +11,7 @@ interface AlpacaService {
     fun observeOnConnectionEvent(): Flow<WebSocket.Event>
 
     @Send
-    fun sendSubscribe(message: MessageAlpacaService)
+    fun sendMessage(message: MessageAlpacaService)
 
     @Receive
     fun observeResponse(): Flow<List<Any>>

app/src/main/java/dev/pinkroom/marketsight/data/repository/NewsRepositoryImp.kt

@@ -1,44 +1,43 @@
 package dev.pinkroom.marketsight.data.repository
 
-import android.util.Log
-import com.tinder.scarlet.WebSocket
 import dev.pinkroom.marketsight.common.ActionAlpaca
 import dev.pinkroom.marketsight.common.DispatcherProvider
 import dev.pinkroom.marketsight.common.Resource
 import dev.pinkroom.marketsight.data.data_source.AlpacaRemoteDataSource
 import dev.pinkroom.marketsight.data.remote.model.request.MessageAlpacaService
-import dev.pinkroom.marketsight.domain.model.NewsInfo
-import dev.pinkroom.marketsight.domain.model.SubscriptionMessage
 import dev.pinkroom.marketsight.domain.repository.NewsRepository
-import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.single
 import javax.inject.Inject
 
 class NewsRepositoryImp @Inject constructor(
     private val remoteDataSource: AlpacaRemoteDataSource,
     private val dispatchers: DispatcherProvider,
 ): NewsRepository {
-    override fun subscribeNews(
-        symbols: List<String>,
-    ): Flow<Resource<WebSocket.Event>> = flow {
+    override fun subscribeNews(symbols: List<String>) = flow {
         remoteDataSource.subscribeNews(symbols = symbols).collect{
             emit(it)
         }
     }.flowOn(dispatchers.IO)
 
-    override fun getRealTimeNews(): Flow<Resource<List<NewsInfo>>> = flow {
+    override fun getRealTimeNews() = flow {
         remoteDataSource.getRealTimeNews().collect{
             emit(it)
         }
     }.flowOn(dispatchers.IO)
 
-    override fun changeFilterNews(
-        symbolsToSubscribe: List<String>?,
-        symbolsToUnsubscribe: List<String>?
-    ): Flow<Resource<SubscriptionMessage>> = flow<Resource<SubscriptionMessage>> {
-        remoteDataSource.sendMessageToAlpacaService(message = MessageAlpacaService(action = ActionAlpaca.Subscribe.action, news = listOf("*"))).collect{
-
+    override suspend fun changeFilterNews(
+        symbols: List<String>,
+        actionAlpaca: ActionAlpaca,
+    ): Resource<List<String>> = flow {
+        val message = MessageAlpacaService(action = actionAlpaca.action, news = symbols)
+        remoteDataSource.sendMessageToAlpacaService(message = message).collect{ response ->
+            when(response){
+                is Resource.Error -> emit(Resource.Error(data = symbols))
+                is Resource.Success -> emit(Resource.Success(data = response.data.news ?: symbols))
+                else -> Unit
+            }
         }
-    }.flowOn(dispatchers.IO)
+    }.flowOn(dispatchers.IO).single()
 }

app/src/main/java/dev/pinkroom/marketsight/domain/repository/NewsRepository.kt

@@ -1,18 +1,18 @@
 package dev.pinkroom.marketsight.domain.repository
 
 import com.tinder.scarlet.WebSocket
+import dev.pinkroom.marketsight.common.ActionAlpaca
 import dev.pinkroom.marketsight.common.Resource
 import dev.pinkroom.marketsight.domain.model.NewsInfo
-import dev.pinkroom.marketsight.domain.model.SubscriptionMessage
 import kotlinx.coroutines.flow.Flow
 
 interface NewsRepository {
     fun subscribeNews(
         symbols: List<String> = listOf("*"),
     ): Flow<Resource<WebSocket.Event>>
-    fun getRealTimeNews(): Flow<Resource<List<NewsInfo>>>
-    fun changeFilterNews(
-        symbolsToSubscribe: List<String>? = null,
-        symbolsToUnsubscribe: List<String>? = null,
-    ): Flow<Resource<SubscriptionMessage>>
+    fun getRealTimeNews(): Flow<List<NewsInfo>>
+    suspend fun changeFilterNews(
+        symbols: List<String>,
+        actionAlpaca: ActionAlpaca,
+    ): Resource<List<String>>
 }

app/src/main/java/dev/pinkroom/marketsight/domain/use_case/news/ChangeFilterNews.kt

@@ -0,0 +1,43 @@
+package dev.pinkroom.marketsight.domain.use_case.news
+
+import dev.pinkroom.marketsight.common.ActionAlpaca
+import dev.pinkroom.marketsight.common.DispatcherProvider
+import dev.pinkroom.marketsight.common.Resource
+import dev.pinkroom.marketsight.domain.repository.NewsRepository
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
+import javax.inject.Inject
+
+class ChangeFilterNews @Inject constructor(
+    private val newsRepository: NewsRepository,
+    private val dispatchers: DispatcherProvider,
+){
+    operator fun invoke(
+        subscribeSymbols: List<String>? = null,
+        unsubscribeSymbols: List<String>? = null,
+    ) = flow<Resource<List<String>>> {
+        val symbolsToRevert = mutableListOf<String>()
+        unsubscribeSymbols?.let { symbols ->
+            val response = newsRepository.changeFilterNews(symbols = symbols, actionAlpaca = ActionAlpaca.Unsubscribe)
+            when(response){
+                is Resource.Error -> {
+                    symbolsToRevert.addAll(unsubscribeSymbols)
+                }
+                else -> Unit
+            }
+        }
+
+        subscribeSymbols?.let { symbols ->
+            val response = newsRepository.changeFilterNews(symbols = symbols, actionAlpaca = ActionAlpaca.Subscribe)
+            when(response){
+                is Resource.Success -> emit(Resource.Success(data = response.data))
+                is Resource.Error -> {
+                    symbolsToRevert.addAll(subscribeSymbols)
+                }
+                else -> Unit
+            }
+        }
+
+        if (symbolsToRevert.isNotEmpty()) emit(Resource.Error(data = symbolsToRevert))
+    }.flowOn(dispatchers.IO)
+}

app/src/main/java/dev/pinkroom/marketsight/domain/use_case/news/GetRealTimeNews.kt

@@ -0,0 +1,16 @@
+package dev.pinkroom.marketsight.domain.use_case.news
+
+import dev.pinkroom.marketsight.common.DispatcherProvider
+import dev.pinkroom.marketsight.domain.repository.NewsRepository
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
+import javax.inject.Inject
+
+class GetRealTimeNews @Inject constructor(
+    private val newsRepository: NewsRepository,
+    private val dispatchers: DispatcherProvider,
+){
+    operator fun invoke() = flow {
+        emit(newsRepository.getRealTimeNews())
+    }.flowOn(dispatchers.IO)
+}

app/src/main/java/dev/pinkroom/marketsight/domain/use_case/news/SubscribeNews.kt

@@ -0,0 +1,16 @@
+package dev.pinkroom.marketsight.domain.use_case.news
+
+import dev.pinkroom.marketsight.common.DispatcherProvider
+import dev.pinkroom.marketsight.domain.repository.NewsRepository
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
+import javax.inject.Inject
+
+class SubscribeNews @Inject constructor(
+    private val newsRepository: NewsRepository,
+    private val dispatchers: DispatcherProvider,
+){
+    operator fun invoke() = flow {
+        emit(newsRepository.subscribeNews())
+    }.flowOn(dispatchers.IO)
+}

app/src/main/java/dev/pinkroom/marketsight/ui/home_screen/HomeViewModel.kt

@@ -1,33 +1,11 @@
 package dev.pinkroom.marketsight.ui.home_screen
 
 import androidx.lifecycle.ViewModel
-import androidx.lifecycle.viewModelScope
 import dagger.hilt.android.lifecycle.HiltViewModel
-import dev.pinkroom.marketsight.domain.repository.NewsRepository
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
 import javax.inject.Inject
 
 @HiltViewModel
 class HomeViewModel @Inject constructor(
-    newsRepository: NewsRepository
 ): ViewModel(){
-    init {
-        viewModelScope.launch {
-            newsRepository.subscribeNews().collect{
 
-            }
-        }
-        viewModelScope.launch {
-            newsRepository.getRealTimeNews().collect{
-
-            }
-        }
-        viewModelScope.launch {
-            delay(10000)
-            newsRepository.changeFilterNews().collect{
-
-            }
-        }
-    }
 }