Laimiux

// User can submit the comment
data class SubmitCommentEvent(val comment: String)

Laimiux

fun dataStream(
  textChangedEvents: Flowable<TextChangedEvent>,
  submitCommentEvents: Flowable<SubmitCommentEvent>
): Flowable<CommentFormData> {
  // We need to start with something
  val initialFormState = CommentFormData(comment = "", isCommentValid = false, submitRequest = null)
  // RxJava scan operator takes initial state and a reduce function
  // This operator manages the state for us, where it will keep track 
  // of the latest state and provide it for next incoming event 
  return textChangedEvents.scan(initialFormState) { event, currentState ->                                              

Laimiux

fun reduce(event: TextChangedEvent, state: CommentFormData): CommentFormData {
  // Let's check comment validity
  val isValid = event.enteredText.length > 5
  // Update the current state with the changes
  return state.copy(comment = event.enteredText, isCommentValid = isValid)
}

Laimiux

data class TodoFormState(
  val enteredText: String,
  // After user clicks create, we send a post request.
  val createTodoRequest: CreateTodoRequestState?)

Laimiux

// User changes text entry
data class TextChangedEvent(val newText: String)

// User clicks create 
data class CreateTodoEvent(val text: String)

Laimiux

class CommentFormModel(val commentService: CommentService) {
  
  fun formState(
    textChangedEvents: Observable<TextChangedEvent>,
    submitEvents: Observable<SubmitCommentEvent>
  ) : Observable<CommentFormData> {

    // Explicitly declaring the type, for clarity sake
    val requestStateEvents: Observable<Lce<Comment>> = submitEvents
      .switchMap { event -> commentService.submit(event.comment) }

Laimiux

typealias Reducer<T> = (T) -> T

Laimiux

class StoreRepository {
  Observable<Lce<List<Store>> getStoreEventStream() {
    return userBundleManager.switchMap { bundleEvent ->
      if (bundleEvent.isLoading()) {
        return Observable.just(Lce.loading())
      } else if (bundleEvent.hasError()) {
        return Observable.just(Lce.error(bundleEvent.getError()))
      } else {
       // Stores can come from cached source or network
       return storeListEventStream(bundleEvent.data.getZipCode())

Laimiux

class UserManager {
  Observable<Lce<User>> getUserEventStream() {
   return updateEventStream().startWith(cachedBundle().map(toLce())) 
  }
}

Laimiux

class UserContent {
 // requires both the user and list of published content
}

Observable<Lce<User>> getUserEventStream(String userId);
Observable<Lce<List<Content>> getContentEventStream(String userId);

Observable<Lce<UserContent>> getProfileEventStream(String id) {
  return Observable.combineLatest(
    getUserEventStream(id),