Otak rx

, Rx, , , . - , «» Rx- , — . , , Rx UI API Rx, . , .

, RxJava2. , , . , RxJava . — , , , , .

MVP. , Presenter View , ; , View , , . MVVM , .

Android- FunCorp Rx. , , , , .

, , RxBinding — View Rx. , .

implementation "com.jakewharton.rxbinding3:rxbinding-core:$rx_binding_version"

, View .

View :

• - ;
• .

class AuthFullRegistrationView @Inject constructor(): BaseArchView() {

   fun doneClick(): Observable<Unit> = viewHolder.doneBtn.clicks()
   fun loginClick(): Observable<Unit> = viewHolder.loginBtn.clicks()
   fun nicknameText(): Observable<CharSequence> = viewHolder.nickname.textChanges()
   fun passwordText(): Observable<CharSequence> = viewHolder.password.textChanges()
   fun emailText(): Observable<CharSequence> = viewHolder.email.textChanges()

   fun setNickname(nickname: String?) {
       viewHolder.nickname.setText(nickname)
   }

   fun setNicknameError(enabled: Boolean, text: String? = null) {
       viewHolder.nicknameRegistrationView.error = text
       viewHolder.nicknameRegistrationView.isErrorEnabled = enabled
   }

   fun setEmailError(enabled: Boolean, text: String? = null) {
       viewHolder.emailRegistrationView.error = text
       viewHolder.emailRegistrationView.isErrorEnabled = enabled
   }

   fun setPasswordError(enabled: Boolean, text: String? = null) {
       viewHolder.passwordRegistrationView.error = text
       viewHolder.passwordRegistrationView.isErrorEnabled = enabled
   }
}

View , ( View MVP).

- Presenter. Presenter View. onNext/OnError View . :

archView.loginClick()
   .subscribe { authNavigationController.goToLogin(LoginParams()) }

, . CompositeDisposable -.

abstract class SimpleArchPresenter<V: ArchView> {

   var args: Any? = null
   var state: Bundle? = null

   val compositeDisposable = CompositeDisposable()

   @CallSuper
   open fun bindIntents(archView: V) {
   }

   @CallSuper
   open fun unbindIntents() {
       compositeDisposable.clear()
   }
}

fun Disposable.addToDisposable(compositeDisposable: CompositeDisposable): Disposable {
   compositeDisposable.add(this)
   return this
}

:

archView.loginClick()
   .subscribe { authNavigationController.goToLogin(LoginParams()) }
   .addToDisposable(compositeDisposable)

. , :

• Sign in;
• API;
• .

private fun bindRegistrationFlow(archView: AuthFullRegistrationView) {
   val handleFields = Observable
       .zip(archView.nicknameText(), archView.emailText(), archView.passwordText(),
           Function3{ nicknameText: CharSequence, emailText: CharSequence, passwordText: CharSequence ->
               FullRegistrationInteractor.RegisterFields(nicknameText, passwordText, emailText)
           })
       .take(1)

   archView.doneClick()
       .flatMap { handleFields }
       .flatMap {
           fullRegistrationInteractor.registration(it)
                   .subscribeOn(Schedulers.io())
       }
       .safeResponseSubscribe({ authNavigationController.goToVerifyNewEmail() },
                              { handleError(archView, it) })
       .addToDisposable(compositeDisposable)
}

, , . .

:

/**
* if we hide progress with delay and after that this observable is completed, doOnDispose not
* called and we dont reset delayed progress
*/
fun <E, T> Observable<E>.withProgress(progress: ProgressDelegate, action: (data: E) -> Observable<T>): Observable<T> {
  return this.flatMap {
     action.invoke(it)
           .doOnSubscribe { progress.show() }
           .observeOn(AndroidSchedulers.mainThread())
           .doFinally {
              progress.hide()
           }
  }
}

interface ProgressDelegate {

  fun show()
  fun hide()
}

Observable, onSubscribe onFinally.

.

class ProgressDialogManager constructor(activity: Activity, defaultMessage: CharSequence) : ProgressDelegate {

  private val progressDialog = ProgressDialog(activity)

  init {
     progressDialog.setCancelable(false)
     progressDialog.setMessage(defaultMessage)
  }

  override fun show() {
     if (progressDialog.isShowing) return
     progressDialog.show()
  }

  override fun hide() {
     if (!progressDialog.isShowing) return
     progressDialog.dismiss()
  }
}

ProgressDelegate, RecycleView :

private class StatisticsServersListProgressDelegate(private val statisticsServersAdapter: StatisticsServersAdapter) : ProgressDelegate {

  override fun show() {
     with(statisticsServersAdapter) {
        transaction {
           clear()
           add(ArchAdapterItem(PROGRESS_VIEW_TYPE.toString(),
                               SimpleProgressItemData(isFooter = false),
                               PROGRESS_VIEW_TYPE))
        }
     }
  }

  override fun hide() {
     statisticsServersAdapter.removeItem(PROGRESS_VIEW_TYPE.toString())
  }
}

, , , ProgressDelegate, .

class DelayedProgressDelegate internal constructor(private val progressDelegate: ProgressDelegate,
                                                  private val delayTimeMillis: Long = SHOW_DELAY_MILLIS) : ProgressDelegate {

  companion object {
     private const val SHOW_DELAY_MILLIS = 800L
  }

  enum class ProgressState {
     SHOW,
     HIDE
  }

  private val stateChangeListener = object : StateMachine.StateChangeListener<ProgressState> {
     override fun onStateChanged(oldState: ProgressState, newState: ProgressState) {
        when (newState) {
           ProgressState.SHOW -> progressDelegate.show()
           ProgressState.HIDE -> progressDelegate.hide()
        }
     }
  }

  private val stateMachine = StateMachine(ProgressState.HIDE, stateChangeListener)

  override fun show() {
     stateMachine.gotoState(ProgressState.SHOW, delayTimeMillis, true)
  }

  override fun hide() {
     stateMachine.gotoState(ProgressState.HIDE, 0, true)
  }

  fun reset() {
     stateMachine.clear()
  }
}

.

:

private fun bindRegistrationFlow(archView: AuthFullRegistrationView) {
   val handleFields = Observable
       .zip(archView.nicknameText(), archView.emailText(), archView.passwordText(),
           Function3{ nicknameText: CharSequence, emailText: CharSequence, passwordText: CharSequence ->
               FullRegistrationInteractor.RegisterFields(nicknameText, passwordText, emailText)
           })
       .take(1)

   archView.doneClick()
       .flatMap { handleFields }
       .withProgress(progressDialogManager.delayed())  {  // flatMap -> withProgress
           fullRegistrationInteractor.registration(it)
                   .subscribeOn(Schedulers.io())
       }
       .safeResponseSubscribe({ authNavigationController.goToVerifyNewEmail(VerifyEmailParams(archView.getEmailText())) },
                              { handleError(archView, it) })
       .addToDisposable(compositeDisposable)
}

, Rx.

fun createSimpleDialog(@StringRes messageId: Int, @StringRes positiveTitle: Int, @StringRes negativeTitle: Int,
                      postCreateDialogAction: (dialog: AlertDialog) -> Unit = {}): Observable<Boolean> {
  val result = BehaviorSubject.create<Boolean>()
  val dialog = AlertDialog.Builder(activity, styleId)
        .setMessage(messageId)
        .setPositiveButton(positiveTitle) { _, _ -> result.onNext(true) }
        .setNegativeButton(negativeTitle) { _, _ -> result.onNext(false) }
        .create()
  dialog.setOnDismissListener { result.onComplete() }
  return result.doOnSubscribe {
     dialog.show()
     postCreateDialogAction.invoke(dialog)
  }
        .subscribeOn(AndroidSchedulers.mainThread())
        .doOnDispose { dialog.dismiss() }
}

:

private fun bindDeleteMenuItem(archView: CurrentServerArchView, server: ServerEntity) {
   archView.serverMenuDeleteClicks()
           .concatMap {
               alertDialogRxFactory.createSimpleDialog(R.string.delete_server_alert_message,
                                                       R.string.common_yes,
                                                       R.string.common_no)
           }
           .filter { it }
           .withProgress(progressDialogManager.delayed()) {
               deleteServerInteractor.deleteServer(server.id)
                       .subscribeOn(Schedulers.io())
           }
           .safeResponseSubscribe(onError = {
               errorDialogProvider.showModelError(it)
           })
           .addToDisposable(serverActions)
}

archView.serverMenuDeleteClicks — , .

class ActionMenuArchView @Inject constructor(private val activity: Activity) : BaseArchView() {

  val items: List<MenuItem> get() = menuView.menu.items()
  val menuClicks: Observable<MenuItem> by lazy { menuView.menuClicks().share() }

  private val menuView: ActionMenuView get() = viewHolder.view as ActionMenuView

  fun inflateMenu(@MenuRes menu: Int): List<MenuItem> {
     menuView.menu.clear()
     activity.menuInflater.inflate(menu, menuView.menu)
     return items
  }

  fun itemClicks(@IdRes itemId: Int): Observable<MenuItem> =
        menuClicks.filter { it.itemId == itemId }
}

, :

fun openCalendar(activity: Activity): Observable<Date> {
   val result = BehaviorSubject.create<Date>()
   val dateCalendar = Calendar.getInstance()
   dateCalendar.timeInMillis = System.currentTimeMillis()

   val dialog = DatePickerDialog(activity,
       DatePickerDialog.OnDateSetListener { _, year, month, dayOfMonth ->
           val calendar = Calendar.getInstance()
           calendar.set(year, month, dayOfMonth)
           result.onNext(calendar.time)
           result.onComplete()
       },dateCalendar.get(Calendar.YEAR),
       dateCalendar.get(Calendar.MONTH),
       dateCalendar.get(Calendar.DAY_OF_MONTH)
   )
   dialog.setOnDismissListener { result.onComplete() }
   dialog.show()
   return result.doOnDispose { dialog.dismiss() }
}

:

fun openSexChooser(activity: Activity): Observable<String> {
   val result = BehaviorSubject.create<String>()
   val items = arrayOf("Male", "Female")
   val builder = AlertDialog.Builder(activity)
   builder.setItems(items) { _, which ->
       result.onNext(items[which])
       result.onComplete()
   }

   val alertDialog = builder.create()
   alertDialog.setOnDismissListener { result.onComplete() }
   alertDialog.show()
   return result.doOnDispose { alertDialog.dismiss() }
}

doOnSubscribe, :)

?

• ;
• , , ;
• , «» «» ;
• — .

. ? , , LiveData, .. UI onPause onResume.

@ActivityScope
class ActivityLifecycleDispatcher @Inject constructor() {
   private val isResumedSubject = BehaviorSubject.create<Boolean>()

   fun isResumed(): Observable<Boolean> = isResumedSubject

   fun onResumed(isResumed: Boolean) {
       isResumedSubject.onNext(isResumed)
   }
}

private fun startServers() {
  updateServersDisposable?.dispose()
  updateServersDisposable = visibleScreen()
        .flatMap { serversInteractor.load().repeatWhen { it.delay(UPDATING_PERIOD_SECONDS, TimeUnit.SECONDS) } }
        .subscribeOn(Schedulers.io())
        .subscribe()
}

private fun visibleScreen() = activityLifecycleDispatcher.isResumed().filter { it }.take(1)

, .

onNewIntent.

@ActivityScope
class ActivityResultDispatcher @Inject constructor() {
   private val onActivityResultSubject = PublishSubject.create<OnActivityResultInfo>()

   fun onActivityResult(requestCode: Int): Observable<OnActivityResultInfo> = onActivityResultSubject.filter { it.requestCode == requestCode }

   fun handleOnActivityResult(requestCode: Int, resultCode: Int, data: Intent?) {
       onActivityResultSubject.onNext(OnActivityResultInfo(requestCode, resultCode, data))
   }
}

data class OnActivityResultInfo constructor(val requestCode: Int, val resultCode: Int, val data: Intent?)

, :

activityResultDispatcher.onActivityResult(ServerChannelsListPresenter.CREATE_CHANNEL_REQUEST_CODE)
     .filter { it.resultCode == Activity.RESULT_OK }
     .map { it.data!!.extras!!.revealNavigationParams<ChannelSettingsFinishParams>()!! }
     .flatMap {
        getChannelInteractor.get(it.channelId)
              .take(1)
              .subscribeOn(Schedulers.io())
     }
     .observeOn(AndroidSchedulers.mainThread())
     .subscribe {
        if (it.data != null) {
           galleryNavigator.openChannel(it.requireData)
           mainMenuVisibilityController.closeImmediately()
        }
     }
     .addToDisposable(compositeDisposable)

• onActivityResult;
• ;
• ;
• .

, onNewIntent, , , .

:

fun <T> Observable<SafeResponse<T>>.repeatOnNetwork(networkController: NetworkController): Observable<SafeResponse<T>> {
    return this
        .flatMap {
            if (it.exception is ServerException) {
                Observable.just(it)
            }
            Observable.error<SafeResponse<T>>(Exception())
        }
        .retryWhen {
            it.flatMap { networkController.networkState }
        }
}

RxBinding ExoPlayer, , :

archView.renderFirstFrame()
     .take(1)
     .subscribe { trackViewed() }
     .addToDisposable(contentDisposable)

archView.playWhenReadyState()
     .subscribe { onChangePlayWhenReadyState(it) }
     .addToDisposable(contentDisposable)

archView.videoSize()
     .take(1)
     .subscribe { archView.setSize(it.height, it.width) }
     .addToDisposable(contentDisposable)

archView.playerState()
     .distinctUntilChanged()
     .subscribe { onChangePlayerState(it, archView) }
     .addToDisposable(contentDisposable)

fun SimpleExoPlayer.renderFirstFrame(): Observable<Unit> = ExoPlayerRenderFirstFrameObservable(this)

private class ExoPlayerRenderFirstFrameObservable(private val view: SimpleExoPlayer) : Observable<Unit>() {

   override fun subscribeActual(observer: Observer<in Unit>) {
       val listener = Listener(observer)

       val disposable = object : MainThreadDisposable() {
           override fun onDispose() {
               view.removeVideoListener(listener)
           }
       }
       observer.onSubscribe(disposable)
       view.addVideoListener(listener)
   }

   private class Listener(private val observer: Observer<in Unit>) : VideoListener {
       override fun onRenderedFirstFrame() {
           observer.onNext(Unit)
       }
   }
}

ExoPlayer.

RecycleView

, :

abstract class ArchRecyclerViewAdapter constructor(diffExecutor: Executor,
                                                  private val componentsFactory: ArchRecycleComponentsFactory) : RecyclerView.Adapter<ArchRecycleViewHolder>() {

  /** this items are update after diff is applied. They may be used when you items adapter is currently showing */
  val adapterItems: List<ArchAdapterItem<out Any>> get() = differ.currentList
  /** this items are updated on [update] call. They represent the future state of this adapter and may be used as a cache for further updates*/
  var dataItems: List<ArchAdapterItem<out Any>> = emptyList()
     private set

  private val differ = AsyncListDifferFactory(this, diffExecutor).create()

  fun update(newList: List<ArchAdapterItem<out Any>>?) {
     dataItems = newList ?: emptyList()
     differ.submitList(newList)
  }

  @Suppress("unused")
  fun transaction(transaction: MutableList<ArchAdapterItem<out Any>>.() -> Unit) {
     update(dataItems.toMutableList().apply(transaction))
  }

  override fun onCreateViewHolder(parent: ViewGroup, viewType: Int): ArchRecycleViewHolder {
     val view = componentsFactory.inflateView(viewType, parent)
     return componentsFactory.createViewHolder(viewType, view)!!
  }

  override fun onBindViewHolder(holder: ArchRecycleViewHolder, position: Int) {
     getBinder(holder).bind(holder, adapterItems[position].data)
  }

  override fun onViewRecycled(holder: ArchRecycleViewHolder) {
     super.onViewRecycled(holder)
     getBinder(holder).unbind(holder)
  }

  override fun onViewDetachedFromWindow(holder: ArchRecycleViewHolder) {
     super.onViewDetachedFromWindow(holder)
     getBinder(holder).detach(holder)
  }

  override fun onViewAttachedToWindow(holder: ArchRecycleViewHolder) {
     super.onViewAttachedToWindow(holder)
     getBinder(holder).attach(holder)
  }

  override fun getItemViewType(position: Int) = adapterItems[position].viewType
  override fun getItemCount() = adapterItems.size

  private fun getBinder(holder: ArchRecycleViewHolder) = componentsFactory.createViewBinder(holder.itemViewType) as ArchRecycleViewBinder<ArchRecycleViewHolder, Any?>
}

interface ArchRecycleViewBinder<V: ArchRecycleViewHolder, D> {
   fun bind(holder: V, data: D)
   fun unbind(holder: V) {}

   fun attach(holder: V) {}
   fun detach(holder: V) {}
}

, ArchRecycleViewBinder, bind . . , . View Presenter.

class ServerMemberViewBinder @Inject constructor() : ArchRecycleViewBinder<ServerMemberViewHolder, ServerMemberItemData> {

   private val memberClickSubject = BehaviorSubject.create<ServerMember>()
   private val roleMenuClickSubject = BehaviorSubject.create<ServerMember>()
   private var roleSettingsEnable: Boolean = false

   fun memberClicks(): Observable<ServerMember> = memberClickSubject
   fun roleMenuClicks(): Observable<ServerMember> = roleMenuClickSubject

   fun setRoleSettingsEnable(enable: Boolean) {
       roleSettingsEnable = enable
   }

   override fun bind(holder: ServerMemberViewHolder, data: ServerMemberItemData) {
       val serverMember = data.serverMember

       ViewUtils.setViewVisibility(holder.moreMenu,
                                   roleSettingsEnable && serverMember.role != ServerRole.OWNER)

       holder.itemView.clicks()
               .map { serverMember }
               .subscribe(memberClickSubject)
       holder.moreMenu.clicks()
               .map { serverMember }
               .subscribe(roleMenuClickSubject)
   }
}

: , , , , .. , . , , , Room Rx, , . , , .

.