Merge pull request #3215 from vector-im/feature/bma/crashRxSingle

Fix crash rx single

CHANGES.md

@@ -9,6 +9,8 @@ Improvements 🙌:
 
 Bugfix 🐛:
  - Message states cosmetic changes (#3007)
+ - Fix exception in rxSingle (#3180)
+ - Do not invite the current user when creating a room (#3123)
  - Fix color issues when the system theme is changed (#2738)
 
 Translations 🗣:

matrix-sdk-android-rx/src/main/java/org/matrix/android/sdk/rx/RxCallbackBuilders.kt

@@ -1,56 +0,0 @@
-/*
- * Copyright 2020 The Matrix.org Foundation C.I.C.
- *
- * 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 org.matrix.android.sdk.rx
-
-import org.matrix.android.sdk.api.MatrixCallback
-import org.matrix.android.sdk.api.util.Cancelable
-import io.reactivex.Completable
-import io.reactivex.Single
-
-fun <T> singleBuilder(builder: (MatrixCallback<T>) -> Cancelable): Single<T> = Single.create { emitter ->
-    val callback = object : MatrixCallback<T> {
-        override fun onSuccess(data: T) {
-            // Add `!!` to fix the warning:
-            // "Type mismatch: type parameter with nullable bounds is used T is used where T was expected. This warning will become an error soon"
-            emitter.onSuccess(data!!)
-        }
-
-        override fun onFailure(failure: Throwable) {
-            emitter.tryOnError(failure)
-        }
-    }
-    val cancelable = builder(callback)
-    emitter.setCancellable {
-        cancelable.cancel()
-    }
-}
-
-fun <T> completableBuilder(builder: (MatrixCallback<T>) -> Cancelable): Completable = Completable.create { emitter ->
-    val callback = object : MatrixCallback<T> {
-        override fun onSuccess(data: T) {
-            emitter.onComplete()
-        }
-
-        override fun onFailure(failure: Throwable) {
-            emitter.tryOnError(failure)
-        }
-    }
-    val cancelable = builder(callback)
-    emitter.setCancellable {
-        cancelable.cancel()
-    }
-}

matrix-sdk-android/src/main/java/org/matrix/android/sdk/api/failure/Failure.kt

@@ -32,7 +32,6 @@ import java.io.IOException
  */
 sealed class Failure(cause: Throwable? = null) : Throwable(cause = cause) {
     data class Unknown(val throwable: Throwable? = null) : Failure(throwable)
-    data class Cancelled(val throwable: Throwable? = null) : Failure(throwable)
     data class UnrecognizedCertificateFailure(val url: String, val fingerprint: Fingerprint) : Failure()
     data class NetworkConnection(val ioException: IOException? = null) : Failure(ioException)
     data class ServerError(val error: MatrixError, val httpCode: Int) : Failure(RuntimeException(error.toString()))

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/network/Request.kt

@@ -88,8 +88,8 @@ internal suspend inline fun <DATA> executeRequest(globalErrorReceiver: GlobalErr
                 throw when (exception) {
                     is IOException              -> Failure.NetworkConnection(exception)
                     is Failure.ServerError,
-                    is Failure.OtherServerError -> exception
-                    is CancellationException    -> Failure.Cancelled(exception)
+                    is Failure.OtherServerError,
+                    is CancellationException    -> exception
                     else                        -> Failure.Unknown(exception)
                 }
             }

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/create/CreateRoomBodyBuilder.kt

@@ -27,6 +27,7 @@ import org.matrix.android.sdk.api.util.MimeTypes
 import org.matrix.android.sdk.internal.crypto.DeviceListManager
 import org.matrix.android.sdk.internal.crypto.MXCRYPTO_ALGORITHM_MEGOLM
 import org.matrix.android.sdk.internal.di.AuthenticatedIdentity
+import org.matrix.android.sdk.internal.di.UserId
 import org.matrix.android.sdk.internal.network.token.AccessTokenProvider
 import org.matrix.android.sdk.internal.session.content.FileUploader
 import org.matrix.android.sdk.internal.session.identity.EnsureIdentityTokenTask
@@ -43,6 +44,8 @@ internal class CreateRoomBodyBuilder @Inject constructor(
         private val deviceListManager: DeviceListManager,
         private val identityStore: IdentityStore,
         private val fileUploader: FileUploader,
+        @UserId
+        private val userId: String,
         @AuthenticatedIdentity
         private val accessTokenProvider: AccessTokenProvider
 ) {
@@ -80,7 +83,7 @@ internal class CreateRoomBodyBuilder @Inject constructor(
                 roomAliasName = params.roomAliasName,
                 name = params.name,
                 topic = params.topic,
-                invitedUserIds = params.invitedUserIds,
+                invitedUserIds = params.invitedUserIds.filter { it != userId },
                 invite3pids = invite3pids,
                 creationContent = params.creationContent.takeIf { it.isNotEmpty() },
                 initialStates = initialStates,

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/job/SyncThread.kt

@@ -20,6 +20,7 @@ import androidx.lifecycle.LiveData
 import androidx.lifecycle.MutableLiveData
 import androidx.lifecycle.Observer
 import com.squareup.moshi.JsonEncodingException
+import kotlinx.coroutines.CancellationException
 import org.matrix.android.sdk.api.failure.Failure
 import org.matrix.android.sdk.api.failure.isTokenError
 import org.matrix.android.sdk.api.session.sync.SyncState
@@ -199,7 +200,7 @@ internal class SyncThread @Inject constructor(private val syncTask: SyncTask,
             if (failure is Failure.NetworkConnection && failure.cause is SocketTimeoutException) {
                 // Timeout are not critical
                 Timber.v("Timeout")
-            } else if (failure is Failure.Cancelled) {
+            } else if (failure is CancellationException) {
                 Timber.v("Cancelled")
             } else if (failure.isTokenError()) {
                 // No token or invalid token, stop the thread

vector/src/main/java/im/vector/app/core/mvrx/ResultExtension.kt

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2021 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.app.core.mvrx
+
+import com.airbnb.mvrx.Async
+import com.airbnb.mvrx.Fail
+import com.airbnb.mvrx.Success
+
+/**
+ * Note: this will be removed when upgrading to mvrx2
+ */
+suspend fun <A> runCatchingToAsync(block: suspend () -> A): Async<A> {
+    return runCatching {
+        block.invoke()
+    }.fold(
+            { Success(it) },
+            { Fail(it) }
+    )
+}

vector/src/main/java/im/vector/app/features/createdirect/CreateDirectRoomViewModel.kt

@@ -19,6 +19,7 @@ package im.vector.app.features.createdirect
 import androidx.lifecycle.viewModelScope
 import com.airbnb.mvrx.ActivityViewModelContext
 import com.airbnb.mvrx.FragmentViewModelContext
+import com.airbnb.mvrx.Loading
 import com.airbnb.mvrx.MvRxViewModelFactory
 import com.airbnb.mvrx.Success
 import com.airbnb.mvrx.ViewModelContext
@@ -26,6 +27,7 @@ import dagger.assisted.Assisted
 import dagger.assisted.AssistedInject
 import dagger.assisted.AssistedFactory
 import im.vector.app.core.extensions.exhaustive
+import im.vector.app.core.mvrx.runCatchingToAsync
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.raw.wellknown.getElementWellknown
 import im.vector.app.features.raw.wellknown.isE2EByDefault
@@ -35,7 +37,6 @@ import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.raw.RawService
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.room.model.create.CreateRoomParams
-import org.matrix.android.sdk.rx.rx
 
 class CreateDirectRoomViewModel @AssistedInject constructor(@Assisted
                                                             initialState: CreateDirectRoomViewState,
@@ -82,6 +83,8 @@ class CreateDirectRoomViewModel @AssistedInject constructor(@Assisted
     }
 
     private fun createRoomAndInviteSelectedUsers(selections: Set<PendingSelection>) {
+        setState { copy(createAndInviteState = Loading()) }
+
         viewModelScope.launch(Dispatchers.IO) {
             val adminE2EByDefault = rawService.getElementWellknown(session.myUserId)
                     ?.isE2EByDefault()
@@ -99,11 +102,15 @@ class CreateDirectRoomViewModel @AssistedInject constructor(@Assisted
                         enableEncryptionIfInvitedUsersSupportIt = adminE2EByDefault
                     }
 
-            session.rx()
-                    .createRoom(roomParams)
-                    .execute {
-                        copy(createAndInviteState = it)
-                    }
+            val result = runCatchingToAsync {
+                session.createRoom(roomParams)
+            }
+
+            setState {
+                copy(
+                        createAndInviteState = result
+                )
+            }
         }
     }
 }

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

@@ -35,6 +35,7 @@ import dagger.assisted.AssistedInject
 import im.vector.app.BuildConfig
 import im.vector.app.R
 import im.vector.app.core.extensions.exhaustive
+import im.vector.app.core.mvrx.runCatchingToAsync
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
 import im.vector.app.features.call.dialpad.DialPadLookup
@@ -649,12 +650,15 @@ class RoomDetailViewModel @AssistedInject constructor(
             val viaServers = MatrixPatterns.extractServerNameFromId(action.event.senderId)
                     ?.let { listOf(it) }
                     .orEmpty()
-            session.rx()
-                    .joinRoom(roomId, viaServers = viaServers)
-                    .map { roomId }
-                    .execute {
-                        copy(tombstoneEventHandling = it)
-                    }
+            viewModelScope.launch {
+                val result = runCatchingToAsync {
+                    session.joinRoom(roomId, viaServers = viaServers)
+                    roomId
+                }
+                setState {
+                    copy(tombstoneEventHandling = result)
+                }
+            }
         }
     }
 

vector/src/main/java/im/vector/app/features/home/room/detail/search/SearchViewModel.kt

@@ -28,9 +28,9 @@ import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import im.vector.app.core.extensions.exhaustive
 import im.vector.app.core.platform.VectorViewModel
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
-import org.matrix.android.sdk.api.failure.Failure
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.search.SearchResult
 
@@ -120,7 +120,7 @@ class SearchViewModel @AssistedInject constructor(
                 )
                 onSearchResultSuccess(result)
             } catch (failure: Throwable) {
-                if (failure is Failure.Cancelled) return@launch
+                if (failure is CancellationException) return@launch
 
                 _viewEvents.post(SearchViewEvents.Failure(failure))
                 setState {

vector/src/main/java/im/vector/app/features/invite/InviteUsersToRoomViewModel.kt

@@ -92,6 +92,6 @@ class InviteUsersToRoomViewModel @AssistedInject constructor(@Assisted
     }
 
     fun getUserIdsOfRoomMembers(): Set<String> {
-        return room.roomSummary()?.otherMemberIds?.toSet() ?: emptySet()
+        return room.roomSummary()?.otherMemberIds?.toSet().orEmpty()
     }
 }

vector/src/main/java/im/vector/app/features/login/AbstractLoginFragment.kt

@@ -29,6 +29,7 @@ import im.vector.app.core.dialogs.UnrecognizedCertificateDialog
 import im.vector.app.core.extensions.exhaustive
 import im.vector.app.core.platform.OnBackPressed
 import im.vector.app.core.platform.VectorBaseFragment
+import kotlinx.coroutines.CancellationException
 import org.matrix.android.sdk.api.failure.Failure
 import org.matrix.android.sdk.api.failure.MatrixError
 import javax.net.ssl.HttpsURLConnection
@@ -76,7 +77,7 @@ abstract class AbstractLoginFragment<VB: ViewBinding> : VectorBaseFragment<VB>()
         }
 
         when (throwable) {
-            is Failure.Cancelled                      ->
+            is CancellationException                  ->
                 /* Ignore this error, user has cancelled the action */
                 Unit
             is Failure.ServerError                    ->

vector/src/main/java/im/vector/app/features/permalink/PermalinkHandler.kt

@@ -112,7 +112,10 @@ class PermalinkHandler @Inject constructor(private val activeSessionHolder: Acti
     private fun PermalinkData.RoomLink.getRoomId(): Single<Optional<String>> {
         val session = activeSessionHolder.getSafeActiveSession()
         return if (isRoomAlias && session != null) {
-            session.rx().getRoomIdByAlias(roomIdOrAlias, true).map { it.getOrNull()?.roomId.toOptional() }.subscribeOn(Schedulers.io())
+            session.rx()
+                    .getRoomIdByAlias(roomIdOrAlias, true)
+                    .map { it.getOrNull()?.roomId.toOptional() }
+                    .subscribeOn(Schedulers.io())
         } else {
             Single.just(Optional.from(roomIdOrAlias))
         }

vector/src/main/java/im/vector/app/features/roomdirectory/RoomDirectoryViewModel.kt

@@ -29,10 +29,10 @@ import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.features.settings.VectorPreferences
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import org.matrix.android.sdk.api.extensions.orFalse
-import org.matrix.android.sdk.api.failure.Failure
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.room.model.Membership
 import org.matrix.android.sdk.api.session.room.model.roomdirectory.PublicRoomsFilter
@@ -87,7 +87,7 @@ class RoomDirectoryViewModel @AssistedInject constructor(
                     val joinedRoomIds = list
                             ?.map { it.roomId }
                             ?.toSet()
-                            ?: emptySet()
+                            .orEmpty()
 
                     setState {
                         copy(joinedRoomsIds = joinedRoomIds)
@@ -183,7 +183,7 @@ class RoomDirectoryViewModel @AssistedInject constructor(
                         )
                 )
             } catch (failure: Throwable) {
-                if (failure is Failure.Cancelled) {
+                if (failure is CancellationException) {
                     // Ignore, another request should be already started
                     return@launch
                 }

vector/src/main/java/im/vector/app/features/roommemberprofile/RoomMemberProfileViewModel.kt

@@ -29,6 +29,7 @@ import dagger.assisted.Assisted
 import dagger.assisted.AssistedInject
 import dagger.assisted.AssistedFactory
 import im.vector.app.R
+import im.vector.app.core.mvrx.runCatchingToAsync
 import im.vector.app.core.platform.VectorViewModel
 import im.vector.app.core.resources.StringProvider
 import im.vector.app.features.powerlevel.PowerLevelsObservableFactory
@@ -139,7 +140,7 @@ class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private v
 
     override fun handle(action: RoomMemberProfileAction) {
         when (action) {
-            is RoomMemberProfileAction.RetryFetchingInfo      -> fetchProfileInfo()
+            is RoomMemberProfileAction.RetryFetchingInfo      -> handleRetryFetchProfileInfo()
             is RoomMemberProfileAction.IgnoreUser             -> handleIgnoreAction()
             is RoomMemberProfileAction.VerifyUser             -> prepareVerification()
             is RoomMemberProfileAction.ShareRoomMemberProfile -> handleShareRoomMemberProfile()
@@ -259,18 +260,27 @@ class RoomMemberProfileViewModel @AssistedInject constructor(@Assisted private v
                 }
     }
 
-    private fun fetchProfileInfo() {
-        session.rx().getProfileInfo(initialState.userId)
-                .map {
-                    MatrixItem.UserItem(
-                            id = initialState.userId,
-                            displayName = it[ProfileService.DISPLAY_NAME_KEY] as? String,
-                            avatarUrl = it[ProfileService.AVATAR_URL_KEY] as? String
-                    )
-                }
-                .execute {
-                    copy(userMatrixItem = it)
-                }
+    private fun handleRetryFetchProfileInfo() {
+        viewModelScope.launch {
+            fetchProfileInfo()
+        }
+    }
+
+    private suspend fun fetchProfileInfo() {
+        val result = runCatchingToAsync {
+            session.getProfile(initialState.userId)
+                    .let {
+                        MatrixItem.UserItem(
+                                id = initialState.userId,
+                                displayName = it[ProfileService.DISPLAY_NAME_KEY] as? String,
+                                avatarUrl = it[ProfileService.AVATAR_URL_KEY] as? String
+                        )
+                    }
+        }
+
+        setState {
+            copy(userMatrixItem = result)
+        }
     }
 
     private fun observeRoomSummaryAndPowerLevels(room: Room) {

vector/src/main/java/im/vector/app/features/userdirectory/UserListController.kt

@@ -109,33 +109,37 @@ class UserListController @Inject constructor(private val session: Session,
     }
 
     private fun buildKnownUsers(currentState: UserListViewState, selectedUsers: List<String>) {
-        currentState.knownUsers()?.let { userList ->
-            userListHeaderItem {
-                id("known_header")
-                header(stringProvider.getString(R.string.direct_room_user_list_known_title))
-            }
+        currentState.knownUsers()
+                ?.filter { it.userId != session.myUserId }
+                ?.let { userList ->
+                    userListHeaderItem {
+                        id("known_header")
+                        header(stringProvider.getString(R.string.direct_room_user_list_known_title))
+                    }
 
-            if (userList.isEmpty()) {
-                renderEmptyState()
-                return
-            }
-            userList.forEach { item ->
-                val isSelected = selectedUsers.contains(item.userId)
-                userDirectoryUserItem {
-                    id(item.userId)
-                    selected(isSelected)
-                    matrixItem(item.toMatrixItem())
-                    avatarRenderer(avatarRenderer)
-                    clickListener { _ ->
-                        callback?.onItemClick(item)
+                    if (userList.isEmpty()) {
+                        renderEmptyState()
+                        return
+                    }
+                    userList.forEach { item ->
+                        val isSelected = selectedUsers.contains(item.userId)
+                        userDirectoryUserItem {
+                            id(item.userId)
+                            selected(isSelected)
+                            matrixItem(item.toMatrixItem())
+                            avatarRenderer(avatarRenderer)
+                            clickListener { _ ->
+                                callback?.onItemClick(item)
+                            }
+                        }
                     }
                 }
-            }
-        }
     }
 
     private fun buildDirectoryUsers(directoryUsers: List<User>, selectedUsers: List<String>, searchTerms: String, ignoreIds: List<String>) {
-        val toDisplay = directoryUsers.filter { !ignoreIds.contains(it.userId) }
+        val toDisplay = directoryUsers
+                .filter { !ignoreIds.contains(it.userId) && it.userId != session.myUserId }
+
         if (toDisplay.isEmpty() && searchTerms.isBlank()) {
             return
         }
@@ -147,16 +151,14 @@ class UserListController @Inject constructor(private val session: Session,
             renderEmptyState()
         } else {
             toDisplay.forEach { user ->
-                if (user.userId != session.myUserId) {
-                    val isSelected = selectedUsers.contains(user.userId)
-                    userDirectoryUserItem {
-                        id(user.userId)
-                        selected(isSelected)
-                        matrixItem(user.toMatrixItem())
-                        avatarRenderer(avatarRenderer)
-                        clickListener { _ ->
-                            callback?.onItemClick(user)
-                        }
+                val isSelected = selectedUsers.contains(user.userId)
+                userDirectoryUserItem {
+                    id(user.userId)
+                    selected(isSelected)
+                    matrixItem(user.toMatrixItem())
+                    avatarRenderer(avatarRenderer)
+                    clickListener { _ ->
+                        callback?.onItemClick(user)
                     }
                 }
             }

vector/src/main/java/im/vector/app/features/userdirectory/UserListFragment.kt

@@ -89,7 +89,7 @@ class UserListFragment @Inject constructor(
 
         viewModel.observeViewEvents {
             when (it) {
-                is UserListViewEvents.OpenShareMatrixToLing -> {
+                is UserListViewEvents.OpenShareMatrixToLink -> {
                     val text = getString(R.string.invite_friends_text, it.link)
                     startSharePlainTextIntent(
                             fragment = this,

vector/src/main/java/im/vector/app/features/userdirectory/UserListViewEvents.kt

@@ -22,5 +22,5 @@ import im.vector.app.core.platform.VectorViewEvents
  * Transient events for invite users to room screen
  */
 sealed class UserListViewEvents : VectorViewEvents {
-    data class OpenShareMatrixToLing(val link: String) : UserListViewEvents()
+    data class OpenShareMatrixToLink(val link: String) : UserListViewEvents()
 }

vector/src/main/java/im/vector/app/features/userdirectory/UserListViewModel.kt

@@ -29,7 +29,6 @@ import im.vector.app.core.extensions.toggle
 import im.vector.app.core.platform.VectorViewModel
 import io.reactivex.Single
 import io.reactivex.android.schedulers.AndroidSchedulers
-import io.reactivex.disposables.Disposable
 import org.matrix.android.sdk.api.MatrixPatterns
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.profile.ProfileService
@@ -50,8 +49,6 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
     private val knownUsersSearch = BehaviorRelay.create<KnownUsersSearch>()
     private val directoryUsersSearch = BehaviorRelay.create<DirectoryUsersSearch>()
 
-    private var currentUserSearchDisposable: Disposable? = null
-
     @AssistedFactory
     interface Factory {
         fun create(initialState: UserListViewState): UserListViewModel
@@ -92,7 +89,7 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
 
     private fun handleShareMyMatrixToLink() {
         session.permalinkService().createPermalink(session.myUserId)?.let {
-            _viewEvents.post(UserListViewEvents.OpenShareMatrixToLing(it))
+            _viewEvents.post(UserListViewEvents.OpenShareMatrixToLink(it))
         }
     }
 
@@ -115,8 +112,6 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
                     copy(knownUsers = async)
                 }
 
-        currentUserSearchDisposable?.dispose()
-
         directoryUsersSearch
                 .debounce(300, TimeUnit.MILLISECONDS)
                 .switchMapSingle { search ->
@@ -124,7 +119,7 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
                         Single.just(emptyList<User>())
                     } else {
                         val searchObservable = session.rx()
-                                .searchUsersDirectory(search, 50, state.excludedUserIds ?: emptySet())
+                                .searchUsersDirectory(search, 50, state.excludedUserIds.orEmpty())
                                 .map { users ->
                                     users.sortedBy { it.toMatrixItem().firstLetterOfDisplayName() }
                                 }
@@ -144,15 +139,19 @@ class UserListViewModel @AssistedInject constructor(@Assisted initialState: User
                                     }
                                     .onErrorReturn { Optional.empty() }
 
-                            Single.zip(searchObservable, profileObservable, { searchResults, optionalProfile ->
-                                val profile = optionalProfile.getOrNull() ?: return@zip searchResults
-                                val searchContainsProfile = searchResults.indexOfFirst { it.userId == profile.userId } != -1
-                                if (searchContainsProfile) {
-                                    searchResults
-                                } else {
-                                    listOf(profile) + searchResults
-                                }
-                            })
+                            Single.zip(
+                                    searchObservable,
+                                    profileObservable,
+                                    { searchResults, optionalProfile ->
+                                        val profile = optionalProfile.getOrNull() ?: return@zip searchResults
+                                        val searchContainsProfile = searchResults.any { it.userId == profile.userId }
+                                        if (searchContainsProfile) {
+                                            searchResults
+                                        } else {
+                                            listOf(profile) + searchResults
+                                        }
+                                    }
+                            )
                         }
                     }
                     stream.toAsync {