Merge pull request #581 from code-payments/fix/cash-link-receiving

bmc08gt · web-flow · commit dd9f2f795fdf · 2024-10-07T10:25:19.000-04:00
fix: make gRPC observers waitForReady
diff --git a/api/src/main/java/com/getcode/network/api/AccountApi.kt b/api/src/main/java/com/getcode/network/api/AccountApi.kt
@@ -22,7 +22,7 @@ class AccountApi @Inject constructor(
     managedChannel: ManagedChannel,
     private val scheduler: Scheduler = Schedulers.io(),
 ) : GrpcApi(managedChannel) {
-    private val api = AccountGrpc.newStub(managedChannel)
+    private val api = AccountGrpc.newStub(managedChannel).withWaitForReady()
 
     fun isCodeAccount(owner: KeyPair): Flow<AccountService.IsCodeAccountResponse> {
         val request = AccountService.IsCodeAccountRequest.newBuilder()
diff --git a/api/src/main/java/com/getcode/network/api/ChatApiV1.kt b/api/src/main/java/com/getcode/network/api/ChatApiV1.kt
@@ -33,7 +33,7 @@ import com.getcode.model.chat.SetSubscriptionStateResponseV1 as SetSubscriptionS
 class ChatApiV1 @Inject constructor(
     managedChannel: ManagedChannel
 ) : GrpcApi(managedChannel) {
-    private val api = ChatGrpc.newStub(managedChannel)
+    private val api = ChatGrpc.newStub(managedChannel).withWaitForReady()
 
     fun fetchChats(owner: KeyPair): Flow<GetChatsResponse> {
         val request = GetChatsRequest.newBuilder()
diff --git a/api/src/main/java/com/getcode/network/api/ChatApiV2.kt b/api/src/main/java/com/getcode/network/api/ChatApiV2.kt
@@ -53,7 +53,7 @@ import com.getcode.model.chat.SetSubscriptionStateResponseV2 as SetSubscriptionS
 class ChatApiV2 @Inject constructor(
     managedChannel: ManagedChannel
 ) : GrpcApi(managedChannel) {
-    private val api = ChatGrpc.newStub(managedChannel)
+    private val api = ChatGrpc.newStub(managedChannel).withWaitForReady()
 
     fun startChat(
         owner: KeyPair,
diff --git a/api/src/main/java/com/getcode/network/api/CurrencyApi.kt b/api/src/main/java/com/getcode/network/api/CurrencyApi.kt
@@ -15,7 +15,7 @@ import javax.inject.Inject
 class CurrencyApi @Inject constructor(
     managedChannel: ManagedChannel,
 ) : GrpcApi(managedChannel) {
-    private val api = CurrencyGrpc.newStub(managedChannel)
+    private val api = CurrencyGrpc.newStub(managedChannel).withWaitForReady()
 
     fun getRates(request: CurrencyService.GetAllRatesRequest = CurrencyService.GetAllRatesRequest.getDefaultInstance()): Flow<CurrencyService.GetAllRatesResponse> =
         api::getAllRates
diff --git a/api/src/main/java/com/getcode/network/api/DeviceApi.kt b/api/src/main/java/com/getcode/network/api/DeviceApi.kt
@@ -17,7 +17,7 @@ class DeviceApi @Inject constructor(
     managedChannel: ManagedChannel,
 ): GrpcApi(managedChannel) {
 
-    private val api = DeviceGrpc.newStub(managedChannel)
+    private val api = DeviceGrpc.newStub(managedChannel).withWaitForReady()
 
     fun registerInstallation(owner: KeyPair, installationId: String) : Flow<DeviceService.RegisterLoggedInAccountsResponse> {
         val request = DeviceService.RegisterLoggedInAccountsRequest.newBuilder()
diff --git a/api/src/main/java/com/getcode/network/api/IdentityApi.kt b/api/src/main/java/com/getcode/network/api/IdentityApi.kt
@@ -19,7 +19,7 @@ class IdentityApi @Inject constructor(
     managedChannel: ManagedChannel,
     private val scheduler: Scheduler = Schedulers.io(),
 ) : GrpcApi(managedChannel) {
-    private val api = IdentityGrpc.newStub(managedChannel)
+    private val api = IdentityGrpc.newStub(managedChannel).withWaitForReady()
 
     fun linkAccount(request: IdentityService.LinkAccountRequest): @NonNull Single<IdentityService.LinkAccountResponse> {
         return api::linkAccount
diff --git a/api/src/main/java/com/getcode/network/api/MessagingApi.kt b/api/src/main/java/com/getcode/network/api/MessagingApi.kt
@@ -20,7 +20,7 @@ class MessagingApi @Inject constructor(
     managedChannel: ManagedChannel,
     private val scheduler: Scheduler = Schedulers.io()
 ) : GrpcApi(managedChannel) {
-    private val api = MessagingGrpc.newStub(managedChannel)
+    private val api = MessagingGrpc.newStub(managedChannel).withWaitForReady()
 
     fun openMessageStream(request: OpenMessageStreamRequest): Flowable<OpenMessageStreamResponse> =
         api::openMessageStream
diff --git a/api/src/main/java/com/getcode/network/api/PhoneApi.kt b/api/src/main/java/com/getcode/network/api/PhoneApi.kt
@@ -14,7 +14,7 @@ class PhoneApi @Inject constructor(
     managedChannel: ManagedChannel,
     private val scheduler: Scheduler = Schedulers.io(),
 ) : GrpcApi(managedChannel) {
-    private val api = PhoneVerificationGrpc.newStub(managedChannel)
+    private val api = PhoneVerificationGrpc.newStub(managedChannel).withWaitForReady()
 
     fun sendVerificationCode(request: PhoneVerificationService.SendVerificationCodeRequest): @NonNull Single<PhoneVerificationService.SendVerificationCodeResponse> {
         return api::sendVerificationCode
diff --git a/api/src/main/java/com/getcode/network/api/PushApi.kt b/api/src/main/java/com/getcode/network/api/PushApi.kt
@@ -14,7 +14,7 @@ class PushApi @Inject constructor(
     managedChannel: ManagedChannel,
     private val scheduler: Scheduler = Schedulers.io(),
 ) : GrpcApi(managedChannel) {
-    private val api = PushGrpc.newStub(managedChannel)
+    private val api = PushGrpc.newStub(managedChannel).withWaitForReady()
 
     fun addToken(request: PushService.AddTokenRequest): @NonNull Single<PushService.AddTokenResponse> {
         return api::addToken
diff --git a/api/src/main/java/com/getcode/network/api/TransactionApiV2.kt b/api/src/main/java/com/getcode/network/api/TransactionApiV2.kt
@@ -19,7 +19,7 @@ class TransactionApiV2 @Inject constructor(
     managedChannel: ManagedChannel,
     private val scheduler: Scheduler = Schedulers.io(),
 ) : GrpcApi(managedChannel) {
-    private val api = TransactionGrpc.newStub(managedChannel)
+    private val api = TransactionGrpc.newStub(managedChannel).withWaitForReady()
 
     fun submitIntent(request: StreamObserver<TransactionService.SubmitIntentResponse>): StreamObserver<TransactionService.SubmitIntentRequest> {
         return api.submitIntent(request)
diff --git a/api/src/main/java/com/getcode/network/repository/MessagingRepository.kt b/api/src/main/java/com/getcode/network/repository/MessagingRepository.kt
@@ -61,7 +61,6 @@ class MessagingRepository @Inject constructor(
             }
             .doOnNext { messagesList ->
                 if (messagesList.isEmpty()) {
-                    Timber.e("message list is empty")
                     return@doOnNext
                 }
                 ackMessages(rendezvousKeyPair, messagesList.map { it.id })
diff --git a/api/src/main/java/com/getcode/network/repository/StatusRepository.kt b/api/src/main/java/com/getcode/network/repository/StatusRepository.kt
@@ -1,10 +1,10 @@
 package com.getcode.network.repository
 
-import com.squareup.okhttp.Call
-import com.squareup.okhttp.OkHttpClient
-import com.squareup.okhttp.Request
-import com.squareup.okhttp.Response
 import io.reactivex.rxjava3.core.Single
+import okhttp3.Call
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.Response
 import org.json.JSONObject
 
 class StatusRepository {
@@ -20,7 +20,7 @@ class StatusRepository {
                 it.onError(Exception())
                 return@create
             }
-            val json = JSONObject(response.body().string())
+            val json = JSONObject(response.body?.string().orEmpty())
             val minimumVersion = json.getInt("minimumClientVersion")
             val isUpgradeRequired = currentVersionCode < minimumVersion
             it.onSuccess(isUpgradeRequired)
diff --git a/app/src/main/java/com/getcode/SessionController.kt b/app/src/main/java/com/getcode/SessionController.kt
@@ -104,6 +104,8 @@ import io.reactivex.rxjava3.core.Completable
 import io.reactivex.rxjava3.schedulers.Schedulers
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -214,8 +216,7 @@ class SessionController @Inject constructor(
     betaFlagsRepository: BetaFlagsRepository,
     features: FeatureRepository,
 ) {
-    private val scope = CoroutineScope(Dispatchers.IO)
-    
+    val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
     val state = MutableStateFlow(SessionState())
 
     private val _eventFlow: MutableSharedFlow<SessionEvent> = MutableSharedFlow()
@@ -441,17 +442,15 @@ class SessionController @Inject constructor(
                 }
             }.launchIn(scope)
 
-        CoroutineScope(Dispatchers.IO).launch {
-            SessionManager.authState
-                .distinctUntilChangedBy { it.isTimelockUnlocked }
-                .collectLatest {
-                    it.let { state ->
-                        if (state.isTimelockUnlocked) {
-                            this@SessionController.state.update { m -> m.copy(restrictionType = RestrictionType.TIMELOCK_UNLOCKED) }
-                        }
+        SessionManager.authState
+            .distinctUntilChangedBy { it.isTimelockUnlocked }
+            .onEach {
+                it.let { state ->
+                    if (state.isTimelockUnlocked) {
+                        this@SessionController.state.update { m -> m.copy(restrictionType = RestrictionType.TIMELOCK_UNLOCKED) }
                     }
                 }
-        }
+            }.launchIn(scope)
     }
 
     private fun buildScannerElements(
@@ -1750,7 +1749,7 @@ class SessionController @Inject constructor(
             )
         )
 
-        CoroutineScope(Dispatchers.IO).launch {
+        scope.launch(Dispatchers.IO) {
             withContext(Dispatchers.Main) {
                 _eventFlow.emit(SessionEvent.SendIntent(shareIntent))
             }
@@ -1938,6 +1937,7 @@ class SessionController @Inject constructor(
                 )
 
             else -> {
+                throwable.printStackTrace()
                 TopBarManager.showMessage(
                     resources.getString(R.string.error_title_failedToCollect),
                     resources.getString(R.string.error_description_failedToCollect)
