Room upgrade: add rx flux and handle failures more precisely

matrix-sdk-android-rx/src/main/java/im/vector/matrix/rx/RxRoom.kt

@@ -45,6 +45,10 @@ class RxRoom(private val room: Room) {
         room.loadRoomMembersIfNeeded(MatrixCallbackSingle(it)).toSingle(it)
     }
 
+    fun joinRoom(viaServers: List<String> = emptyList()): Single<Unit> = Single.create {
+        room.join(viaServers, MatrixCallbackSingle(it)).toSingle(it)
+    }
+
 }
 
 fun Room.rx(): RxRoom {

matrix-sdk-android-rx/src/main/java/im/vector/matrix/rx/RxSession.kt

@@ -63,6 +63,10 @@ class RxSession(private val session: Session) {
         session.searchUsersDirectory(search, limit, excludedUserIds, MatrixCallbackSingle(it)).toSingle(it)
     }
 
+    fun joinRoom(roomId: String, viaServers: List<String> = emptyList()): Single<Unit> = Single.create {
+        session.joinRoom(roomId, viaServers, MatrixCallbackSingle(it)).toSingle(it)
+    }
+
 }
 
 fun Session.rx(): RxSession {

matrix-sdk-android/src/main/java/im/vector/matrix/android/api/session/room/RoomService.kt

@@ -39,7 +39,7 @@ interface RoomService {
      */
     fun joinRoom(roomId: String,
                  viaServers: List<String> = emptyList(),
-                 callback: MatrixCallback<Unit>)
+                 callback: MatrixCallback<Unit>): Cancelable
 
     /**
      * Get a room from a roomId

matrix-sdk-android/src/main/java/im/vector/matrix/android/api/session/room/failure/CreateRoomFailure.kt

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2019 New Vector Ltd
+ *
+ * Licensed 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
+ *
+ * http://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 im.vector.matrix.android.api.session.room.failure
+
+import im.vector.matrix.android.api.failure.Failure
+
+sealed class CreateRoomFailure : Failure.FeatureFailure() {
+
+    object CreatedWithTimeout: CreateRoomFailure()
+
+}

matrix-sdk-android/src/main/java/im/vector/matrix/android/api/session/room/failure/JoinRoomFailure.kt

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2019 New Vector Ltd
+ *
+ * Licensed 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
+ *
+ * http://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 im.vector.matrix.android.api.session.room.failure
+
+import im.vector.matrix.android.api.failure.Failure
+
+sealed class JoinRoomFailure : Failure.FeatureFailure() {
+
+    object JoinedWithTimeout : JoinRoomFailure()
+
+}

matrix-sdk-android/src/main/java/im/vector/matrix/android/internal/database/RealmQueryLatch.kt

@@ -16,45 +16,45 @@
 
 package im.vector.matrix.android.internal.database
 
-import android.os.Handler
+import im.vector.matrix.android.internal.util.createBackgroundHandler
-import android.os.HandlerThread
 import io.realm.*
-import timber.log.Timber
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicReference
 
-private const val THREAD_NAME = "REALM_QUERY_LATCH"
 
 class RealmQueryLatch<E : RealmObject>(private val realmConfiguration: RealmConfiguration,
                                        private val realmQueryBuilder: (Realm) -> RealmQuery<E>) {
 
-    @Throws(InterruptedException::class)
+    private companion object {
-    fun await(timeout: Long = Long.MAX_VALUE, timeUnit: TimeUnit = TimeUnit.MILLISECONDS) {
+        val QUERY_LATCH_HANDLER = createBackgroundHandler("REALM_QUERY_LATCH")
-        val latch = CountDownLatch(1)
+    }
-        val handlerThread = HandlerThread(THREAD_NAME + hashCode())
-        handlerThread.start()
-        val handler = Handler(handlerThread.looper)
-        val runnable = Runnable {
-            val realm = Realm.getInstance(realmConfiguration)
-            val result = realmQueryBuilder(realm).findAllAsync()
 
+    @Throws(InterruptedException::class)
+    fun await(timeout: Long, timeUnit: TimeUnit) {
+        val realmRef = AtomicReference<Realm>()
+        val latch = CountDownLatch(1)
+        QUERY_LATCH_HANDLER.post {
+            val realm = Realm.getInstance(realmConfiguration)
+            realmRef.set(realm)
+            val result = realmQueryBuilder(realm).findAllAsync()
             result.addChangeListener(object : RealmChangeListener<RealmResults<E>> {
                 override fun onChange(t: RealmResults<E>) {
                     if (t.isNotEmpty()) {
                         result.removeChangeListener(this)
-                        realm.close()
                         latch.countDown()
                     }
                 }
             })
         }
-        handler.post(runnable)
         try {
             latch.await(timeout, timeUnit)
         } catch (exception: InterruptedException) {
             throw exception
         } finally {
-            handlerThread.quit()
+            QUERY_LATCH_HANDLER.post {
+                realmRef.getAndSet(null).close()
+            }
         }
     }
 

matrix-sdk-android/src/main/java/im/vector/matrix/android/internal/session/room/DefaultRoomService.kt

@@ -67,8 +67,8 @@ internal class DefaultRoomService @Inject constructor(private val monarchy: Mona
         )
     }
 
-    override fun joinRoom(roomId: String, viaServers: List<String>, callback: MatrixCallback<Unit>) {
+    override fun joinRoom(roomId: String, viaServers: List<String>, callback: MatrixCallback<Unit>): Cancelable {
-        joinRoomTask
+        return joinRoomTask
                 .configureWith(JoinRoomTask.Params(roomId, viaServers))
                 .dispatchTo(callback)
                 .executeBy(taskExecutor)

matrix-sdk-android/src/main/java/im/vector/matrix/android/internal/session/room/create/CreateRoomTask.kt

@@ -17,7 +17,9 @@
 package im.vector.matrix.android.internal.session.room.create
 
 import arrow.core.Try
+import arrow.core.recoverWith
 import com.zhuinden.monarchy.Monarchy
+import im.vector.matrix.android.api.session.room.failure.CreateRoomFailure
 import im.vector.matrix.android.api.session.room.model.create.CreateRoomParams
 import im.vector.matrix.android.api.session.room.model.create.CreateRoomResponse
 import im.vector.matrix.android.internal.database.RealmQueryLatch
@@ -57,9 +59,11 @@ internal class DefaultCreateRoomTask @Inject constructor(private val roomAPI: Ro
                 realm.where(RoomEntity::class.java)
                         .equalTo(RoomEntityFields.ROOM_ID, roomId)
             }
-            Try {
+            try {
-                rql.await(timeout = 20L, timeUnit = TimeUnit.SECONDS)
+                rql.await(timeout = 1L, timeUnit = TimeUnit.MINUTES)
-                roomId
+                Try.just(roomId)
+            } catch (exception: Exception) {
+                Try.raise<String>(CreateRoomFailure.CreatedWithTimeout)
             }
         }.flatMap { roomId ->
             if (params.isDirect()) {

matrix-sdk-android/src/main/java/im/vector/matrix/android/internal/session/room/membership/joining/JoinRoomTask.kt

@@ -17,6 +17,8 @@
 package im.vector.matrix.android.internal.session.room.membership.joining
 
 import arrow.core.Try
+import im.vector.matrix.android.api.session.room.failure.CreateRoomFailure
+import im.vector.matrix.android.api.session.room.failure.JoinRoomFailure
 import im.vector.matrix.android.internal.database.RealmQueryLatch
 import im.vector.matrix.android.internal.database.model.RoomEntity
 import im.vector.matrix.android.internal.database.model.RoomEntityFields
@@ -50,9 +52,11 @@ internal class DefaultJoinRoomTask @Inject constructor(private val roomAPI: Room
                 realm.where(RoomEntity::class.java)
                         .equalTo(RoomEntityFields.ROOM_ID, roomId)
             }
-            Try {
+            try {
-                rql.await(20L, TimeUnit.SECONDS)
+                rql.await(timeout = 1L, timeUnit = TimeUnit.MINUTES)
-                roomId
+                Try.just(roomId)
+            } catch (exception: Exception) {
+                Try.raise<String>(JoinRoomFailure.JoinedWithTimeout)
             }
         }.flatMap { roomId ->
             setReadMarkers(roomId)

vector/src/main/java/im/vector/riotx/core/error/ErrorFormatter.kt

@@ -30,12 +30,16 @@ class ErrorFormatter @Inject constructor(val stringProvider: StringProvider) {
     }
 
     fun toHumanReadable(throwable: Throwable?): String {
-
         return when (throwable) {
             null                         -> ""
             is Failure.NetworkConnection -> stringProvider.getString(R.string.error_no_network)
-            else                         -> throwable.localizedMessage
+            is Failure.ServerError       -> {
+                throwable.error.message.takeIf { it.isNotEmpty() }
+                ?: throwable.error.code.takeIf { it.isNotEmpty() }
+                ?: stringProvider.getString(R.string.unknown_error)
+            }
+            else                         -> throwable.localizedMessage
+                                            ?: stringProvider.getString(R.string.unknown_error)
         }
-
     }
 }

vector/src/main/java/im/vector/riotx/features/home/createdirect/CreateDirectRoomActivity.kt

@@ -29,6 +29,7 @@ import com.airbnb.mvrx.Fail
 import com.airbnb.mvrx.Loading
 import com.airbnb.mvrx.Success
 import com.airbnb.mvrx.viewModel
+import im.vector.matrix.android.api.session.room.failure.CreateRoomFailure
 import im.vector.riotx.R
 import im.vector.riotx.core.di.ScreenComponent
 import im.vector.riotx.core.error.ErrorFormatter
@@ -38,6 +39,7 @@ import im.vector.riotx.core.extensions.observeEvent
 import im.vector.riotx.core.platform.SimpleFragmentActivity
 import im.vector.riotx.core.platform.WaitingViewData
 import kotlinx.android.synthetic.main.activity.*
+import timber.log.Timber
 import javax.inject.Inject
 
 class CreateDirectRoomActivity : SimpleFragmentActivity() {
@@ -91,6 +93,9 @@ class CreateDirectRoomActivity : SimpleFragmentActivity() {
 
     private fun renderCreationFailure(error: Throwable) {
         hideWaitingView()
+        if (error is CreateRoomFailure.CreatedWithTimeout) {
+            finish()
+        } else
             AlertDialog.Builder(this)
                     .setMessage(errorFormatter.toHumanReadable(error))
                     .setPositiveButton(R.string.ok) { dialog, id -> dialog.cancel() }

vector/src/main/java/im/vector/riotx/features/home/room/detail/RoomDetailFragment.kt

@@ -65,6 +65,7 @@ import com.otaliastudios.autocomplete.CharPolicy
 import im.vector.matrix.android.api.permalinks.PermalinkFactory
 import im.vector.matrix.android.api.session.Session
 import im.vector.matrix.android.api.session.events.model.Event
+import im.vector.matrix.android.api.session.room.failure.JoinRoomFailure
 import im.vector.matrix.android.api.session.room.model.EditAggregatedSummary
 import im.vector.matrix.android.api.session.room.model.Membership
 import im.vector.matrix.android.api.session.room.model.message.*
@@ -608,7 +609,7 @@ class RoomDetailFragment :
                 // TODO Better handling progress
                 vectorBaseActivity.showWaitingView()
                 vectorBaseActivity.waiting_view_status_text.visibility = View.VISIBLE
-                vectorBaseActivity.waiting_view_status_text.text = getString(R.string.join)
+                vectorBaseActivity.waiting_view_status_text.text = getString(R.string.joining_room)
             }
             is Success -> {
                 navigator.openRoom(vectorBaseActivity, async())

vector/src/main/java/im/vector/riotx/features/home/room/detail/RoomDetailViewModel.kt

@@ -140,8 +140,8 @@ class RoomDetailViewModel @AssistedInject constructor(@Assisted initialState: Ro
                                ?: return
 
         val roomId = tombstoneContent.replacementRoom
-        // TODO replace with rx flux
+        val isRoomJoined = session.getRoom(roomId)?.roomSummary()?.membership == Membership.JOIN
-        if (session.getRoom(roomId)?.roomSummary()?.membership == Membership.JOIN) {
+        if (isRoomJoined) {
             setState { copy(tombstoneEventHandling = Success(roomId)) }
         } else {
             val viaServer = MatrixPatterns.extractServerNameFromId(action.event.senderId).let {
@@ -151,17 +151,12 @@ class RoomDetailViewModel @AssistedInject constructor(@Assisted initialState: Ro
                     listOf(it)
                 }
             }
-            setState { copy(tombstoneEventHandling = Loading()) }
+            session.rx()
-            session.joinRoom(roomId, viaServer, object : MatrixCallback<Unit> {
+                    .joinRoom(roomId, viaServer)
-                override fun onSuccess(data: Unit) {
+                    .map { roomId }
-                    setState { copy(tombstoneEventHandling = Success(roomId)) }
+                    .execute {
-
+                        copy(tombstoneEventHandling = it)
                     }
-
-                override fun onFailure(failure: Throwable) {
-                    setState { copy(tombstoneEventHandling = Fail(failure)) }
-                }
-            })
         }
 
     }

vector/src/main/res/values/strings_riotX.xml

@@ -7,4 +7,6 @@
     <string name="direct_room_no_known_users">"No result found, use Add by matrix ID to search on server."</string>
     <string name="direct_room_start_search">"Start typing to get results"</string>
     <string name="direct_room_filter_hint">"Filter by username or ID…"</string>
+
+    <string name="joining_room">"Joining room…"</string>
 </resources>