AssistantSession.kt

val assistantResponses: Observable<Either<AssistantError, AssistantResponse>>
        get() {
            Logger.i(TAG, "getAssistantResponses")
            val activityObservable = readyObservable
                .observeOn(Schedulers.io())
                .flatMap<Either<Throwable, Conversation>> { startConversation() }
                .flatMap<Either<SocketEvent, ActivitySet>> { connectConversation(it) }
                .filter { socketNotOpened(it) }
                .map<Either<AssistantError, ActivitySet>> { mapSocketEventToDomainModel(it) }
                .flatMapIterable<Either<AssistantError, Pair<Activity, String>>> { flatMapIterable(it) }
                .map<Either<AssistantError, Pair<Activity, String>>> { updateLastRequestId(it) }
                .filter { theirActivities(it) }
                .filter { notCanceledRequests(it) }
                .map<Either<AssistantError, Activity>> { updateWatermark(it) }
                .filter { onlyMessages(it) }
                .map<Either<AssistantError, Activity>> { updateDialogContext(it) }
                .map<Either<AssistantError, Activity>> { extractErrorFromChannelDataOrReturn(it) }

            return RxCustomOperators
                .bufferUntil(activityObservable) { getErrorOrHasNotMoreMessages(it) }
                .map { setResponseType(it) }
                .doOnNext { resetLastRequestId(it.second) }
                .map<Either<AssistantError, AssistantResponse>> { firstErrorOrAssistantResponse(it) }
                .flatMap<Either<AssistantError, AssistantResponse>> { prefetchResourcesOrError(it) }
                .doOnEach { clearTimeout() }
                .doOnNext { setLastResponse(it) }
                .mergeWith(responseTimeoutObservable())
        }