Add some cancelable on service methods and start branching Rx

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

@@ -0,0 +1,39 @@
+/*
+ * 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.rx
+
+import im.vector.matrix.android.api.MatrixCallback
+import im.vector.matrix.android.api.util.Cancelable
+import io.reactivex.CompletableEmitter
+import io.reactivex.SingleEmitter
+
+internal class MatrixCallbackCompletable<T>(private val completableEmitter: CompletableEmitter) : MatrixCallback<T> {
+
+    override fun onSuccess(data: T) {
+        completableEmitter.onComplete()
+    }
+
+    override fun onFailure(failure: Throwable) {
+        completableEmitter.onError(failure)
+    }
+}
+
+fun Cancelable.toCompletable(completableEmitter: CompletableEmitter) {
+    completableEmitter.setCancellable {
+        this.cancel()
+    }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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.rx
+
+import im.vector.matrix.android.api.MatrixCallback
+import im.vector.matrix.android.api.util.Cancelable
+import io.reactivex.SingleEmitter
+
+internal class MatrixCallbackSingle<T>(private val singleEmitter: SingleEmitter<T>) : MatrixCallback<T> {
+
+    override fun onSuccess(data: T) {
+        singleEmitter.onSuccess(data)
+    }
+
+    override fun onFailure(failure: Throwable) {
+        singleEmitter.onError(failure)
+    }
+}
+
+fun <T> Cancelable.toSingle(singleEmitter: SingleEmitter<T>) {
+    singleEmitter.setCancellable {
+        this.cancel()
+    }
+}

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

@@ -21,7 +21,7 @@ import im.vector.matrix.android.api.session.room.model.EventAnnotationsSummary
 import im.vector.matrix.android.api.session.room.model.RoomSummary
 import im.vector.matrix.android.api.session.room.timeline.TimelineEvent
 import io.reactivex.Observable
-import io.reactivex.schedulers.Schedulers
+import io.reactivex.Single
 
 class RxRoom(private val room: Room) {
 
@@ -41,6 +41,10 @@ class RxRoom(private val room: Room) {
         return room.liveTimeLineEvent(eventId).asObservable()
     }
 
+    fun loadRoomMembersIfNeeded(): Single<Boolean> = Single.create {
+        room.loadRoomMembersIfNeeded(MatrixCallbackSingle(it)).toSingle(it)
+    }
+
 }
 
 fun Room.rx(): RxRoom {

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

@@ -20,11 +20,12 @@ import im.vector.matrix.android.api.session.Session
 import im.vector.matrix.android.api.session.group.model.GroupSummary
 import im.vector.matrix.android.api.session.pushers.Pusher
 import im.vector.matrix.android.api.session.room.model.RoomSummary
+import im.vector.matrix.android.api.session.room.model.create.CreateRoomParams
 import im.vector.matrix.android.api.session.sync.SyncState
 import im.vector.matrix.android.api.session.user.model.User
+import io.reactivex.Completable
 import io.reactivex.Observable
-import io.reactivex.android.schedulers.AndroidSchedulers
+import io.reactivex.Single
-import io.reactivex.schedulers.Schedulers
 
 class RxSession(private val session: Session) {
 
@@ -48,6 +49,10 @@ class RxSession(private val session: Session) {
         return session.liveUsers().asObservable()
     }
 
+    fun createRoom(roomParams: CreateRoomParams): Single<String> = Single.create {
+        session.createRoom(roomParams, MatrixCallbackSingle(it)).toSingle(it)
+    }
+
 }
 
 fun Session.rx(): RxSession {

matrix-sdk-android/src/main/java/im/vector/matrix/android/api/pushrules/PushRuleService.kt

@@ -18,6 +18,7 @@ package im.vector.matrix.android.api.pushrules
 import im.vector.matrix.android.api.MatrixCallback
 import im.vector.matrix.android.api.pushrules.rest.PushRule
 import im.vector.matrix.android.api.session.events.model.Event
+import im.vector.matrix.android.api.util.Cancelable
 
 interface PushRuleService {
 
@@ -31,7 +32,7 @@ interface PushRuleService {
 
     //TODO update rule
 
-    fun updatePushRuleEnableStatus(kind: String, pushRule: PushRule, enabled: Boolean, callback: MatrixCallback<Unit>)
+    fun updatePushRuleEnableStatus(kind: String, pushRule: PushRule, enabled: Boolean, callback: MatrixCallback<Unit>): Cancelable
 
     fun addPushRuleListener(listener: PushRuleListener)
 

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

@@ -30,20 +30,17 @@ interface RoomDirectoryService {
     /**
      * Get rooms from directory
      */
-    fun getPublicRooms(server: String?,
+    fun getPublicRooms(server: String?, publicRoomsParams: PublicRoomsParams, callback: MatrixCallback<PublicRoomsResponse>): Cancelable
-                       publicRoomsParams: PublicRoomsParams,
-                       callback: MatrixCallback<PublicRoomsResponse>): Cancelable
 
     /**
      * Join a room by id
      */
-    fun joinRoom(roomId: String,
+    fun joinRoom(roomId: String, callback: MatrixCallback<Unit>): Cancelable
-                 callback: MatrixCallback<Unit>)
 
     /**
      * Fetches the overall metadata about protocols supported by the homeserver.
      * Includes both the available protocols and all fields required for queries against each protocol.
      */
-    fun getThirdPartyProtocol(callback: MatrixCallback<Map<String, ThirdPartyProtocol>>)
+    fun getThirdPartyProtocol(callback: MatrixCallback<Map<String, ThirdPartyProtocol>>): Cancelable
 
 }

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

@@ -20,6 +20,7 @@ import androidx.lifecycle.LiveData
 import im.vector.matrix.android.api.MatrixCallback
 import im.vector.matrix.android.api.session.room.model.RoomSummary
 import im.vector.matrix.android.api.session.room.model.create.CreateRoomParams
+import im.vector.matrix.android.api.util.Cancelable
 
 /**
  * This interface defines methods to get rooms. It's implemented at the session level.
@@ -27,10 +28,9 @@ import im.vector.matrix.android.api.session.room.model.create.CreateRoomParams
 interface RoomService {
 
     /**
-     * Create a room
+     * Create a room asynchronously
      */
-    fun createRoom(createRoomParams: CreateRoomParams,
+    fun createRoom(createRoomParams: CreateRoomParams, callback: MatrixCallback<String>): Cancelable
-                   callback: MatrixCallback<String>)
 
     /**
      * Get a room from a roomId

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

@@ -30,7 +30,7 @@ interface MembershipService {
      * This methods load all room members if it was done yet.
      * @return a [Cancelable]
      */
-    fun loadRoomMembersIfNeeded(): Cancelable
+    fun loadRoomMembersIfNeeded(matrixCallback: MatrixCallback<Boolean>): Cancelable
 
     /**
      * Return the roomMember with userId or null.
@@ -52,16 +52,16 @@ interface MembershipService {
     /**
      * Invite a user in the room
      */
-    fun invite(userId: String, callback: MatrixCallback<Unit>)
+    fun invite(userId: String, callback: MatrixCallback<Unit>): Cancelable
 
     /**
      * Join the room, or accept an invitation.
      */
-    fun join(callback: MatrixCallback<Unit>)
+    fun join(callback: MatrixCallback<Unit>): Cancelable
 
     /**
      * Leave the room, or reject an invitation.
      */
-    fun leave(callback: MatrixCallback<Unit>)
+    fun leave(callback: MatrixCallback<Unit>): Cancelable
 
 }

matrix-sdk-android/src/main/java/im/vector/matrix/android/internal/session/notification/DefaultPushRuleService.kt

@@ -23,6 +23,7 @@ import im.vector.matrix.android.api.pushrules.PushRuleService
 import im.vector.matrix.android.api.pushrules.rest.GetPushRulesResponse
 import im.vector.matrix.android.api.pushrules.rest.PushRule
 import im.vector.matrix.android.api.session.events.model.Event
+import im.vector.matrix.android.api.util.Cancelable
 import im.vector.matrix.android.internal.database.mapper.PushRulesMapper
 import im.vector.matrix.android.internal.database.model.PushRulesEntity
 import im.vector.matrix.android.internal.database.model.PusherEntityFields
@@ -82,8 +83,8 @@ internal class DefaultPushRuleService @Inject constructor(
         return contentRules + overrideRules + roomRules + senderRules + underrideRules
     }
 
-    override fun updatePushRuleEnableStatus(kind: String, pushRule: PushRule, enabled: Boolean, callback: MatrixCallback<Unit>) {
+    override fun updatePushRuleEnableStatus(kind: String, pushRule: PushRule, enabled: Boolean, callback: MatrixCallback<Unit>): Cancelable {
-        updatePushRuleEnableStatusTask
+        return updatePushRuleEnableStatusTask
                 .configureWith(UpdatePushRuleEnableStatusTask.Params(kind, pushRule, enabled))
                 // TODO Fetch the rules
                 .dispatchTo(callback)

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

@@ -44,15 +44,15 @@ internal class DefaultRoomDirectoryService @Inject constructor(private val getPu
                 .executeBy(taskExecutor)
     }
 
-    override fun joinRoom(roomId: String, callback: MatrixCallback<Unit>) {
+    override fun joinRoom(roomId: String, callback: MatrixCallback<Unit>): Cancelable {
-        joinRoomTask
+        return joinRoomTask
                 .configureWith(JoinRoomTask.Params(roomId))
                 .dispatchTo(callback)
                 .executeBy(taskExecutor)
     }
 
-    override fun getThirdPartyProtocol(callback: MatrixCallback<Map<String, ThirdPartyProtocol>>) {
+    override fun getThirdPartyProtocol(callback: MatrixCallback<Map<String, ThirdPartyProtocol>>): Cancelable {
-        getThirdPartyProtocolsTask
+        return getThirdPartyProtocolsTask
                 .toConfigurableTask()
                 .dispatchTo(callback)
                 .executeBy(taskExecutor)

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

@@ -23,6 +23,7 @@ import im.vector.matrix.android.api.session.room.Room
 import im.vector.matrix.android.api.session.room.RoomService
 import im.vector.matrix.android.api.session.room.model.RoomSummary
 import im.vector.matrix.android.api.session.room.model.create.CreateRoomParams
+import im.vector.matrix.android.api.util.Cancelable
 import im.vector.matrix.android.internal.database.mapper.RoomSummaryMapper
 import im.vector.matrix.android.internal.database.model.RoomEntity
 import im.vector.matrix.android.internal.database.model.RoomSummaryEntity
@@ -40,8 +41,8 @@ internal class DefaultRoomService @Inject constructor(private val monarchy: Mona
                                                       private val roomFactory: RoomFactory,
                                                       private val taskExecutor: TaskExecutor) : RoomService {
 
-    override fun createRoom(createRoomParams: CreateRoomParams, callback: MatrixCallback<String>) {
+    override fun createRoom(createRoomParams: CreateRoomParams, callback: MatrixCallback<String>): Cancelable {
-        createRoomTask
+        return createRoomTask
                 .configureWith(createRoomParams)
                 .dispatchTo(callback)
                 .executeBy(taskExecutor)

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

@@ -42,9 +42,11 @@ internal class DefaultMembershipService @Inject constructor(private val roomId:
                                                             private val leaveRoomTask: LeaveRoomTask
 ) : MembershipService {
 
-    override fun loadRoomMembersIfNeeded(): Cancelable {
+    override fun loadRoomMembersIfNeeded(matrixCallback: MatrixCallback<Boolean>): Cancelable {
         val params = LoadRoomMembersTask.Params(roomId, Membership.LEAVE)
-        return loadRoomMembersTask.configureWith(params).executeBy(taskExecutor)
+        return loadRoomMembersTask.configureWith(params)
+                .dispatchTo(matrixCallback)
+                .executeBy(taskExecutor)
     }
 
     override fun getRoomMember(userId: String): RoomMember? {
@@ -73,23 +75,23 @@ internal class DefaultMembershipService @Inject constructor(private val roomId:
         return result
     }
 
-    override fun invite(userId: String, callback: MatrixCallback<Unit>) {
+    override fun invite(userId: String, callback: MatrixCallback<Unit>): Cancelable {
         val params = InviteTask.Params(roomId, userId)
-        inviteTask.configureWith(params)
+        return inviteTask.configureWith(params)
                 .dispatchTo(callback)
                 .executeBy(taskExecutor)
     }
 
-    override fun join(callback: MatrixCallback<Unit>) {
+    override fun join(callback: MatrixCallback<Unit>): Cancelable {
         val params = JoinRoomTask.Params(roomId)
-        joinTask.configureWith(params)
+        return joinTask.configureWith(params)
                 .dispatchTo(callback)
                 .executeBy(taskExecutor)
     }
 
-    override fun leave(callback: MatrixCallback<Unit>) {
+    override fun leave(callback: MatrixCallback<Unit>): Cancelable {
         val params = LeaveRoomTask.Params(roomId)
-        leaveRoomTask.configureWith(params)
+        return leaveRoomTask.configureWith(params)
                 .dispatchTo(callback)
                 .executeBy(taskExecutor)
     }