Asynchronous and Event-Driven Programming Questions

Answer (RxSwift / RxKotlin — Mobile Developer view)

Situation: high-rate sensor stream, must transform in background, avoid unbounded memory, deliver batches to UI on main thread with low latency.

Design summary

• Schedulers/Dispatchers:
  • RxSwift: subscribeOn(ConcurrentDispatchQueueScheduler(qos:.userInitiated)) for sensor ingestion and heavy transforms; observeOn(MainScheduler.instance) for UI.
  • RxKotlin: Flowable.subscribeOn(Schedulers.io()) for work; observeOn(AndroidSchedulers.mainThread()) for UI.
• Operators:
  • Use window/buffer to produce small batches: buffer(timeSpan: .milliseconds(200), count: 50).
  • Apply map/flatMap for transforms; use flatMapLatest for replacing stale work when only newest matters.
  • Apply sample or throttle as alternative if single latest sample suffices.
• Backpressure strategy:
  • RxKotlin: use Flowable with onBackpressureLatest() or onBackpressureDrop { log/dropPolicy } to prevent unbounded queues.
  • RxSwift: emulate backpressure by limiting buffer size and dropping oldest when capacity reached (e.g., custom operator or use buffer with count and short timespan), or use .sample/.throttle to reduce rate before heavy transforms.

Example (RxSwift):

let sensorStream: Observable<Sensor> = sensorPublisher()

sensorStream
  .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .userInitiated)) // ingest on background
  .buffer(timeSpan: .milliseconds(200), count: 50, scheduler: MainScheduler.instance) // small batches
  .map { batch in
    // transform on background: switch to background scheduler for CPU work
    Observable.just(batch)
      .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .userInitiated))
      .map { batch.map { transform($0) } }
  }
  .flatMapLatest { $0 } // drop prior processing if new batches arrive
  .observeOn(MainScheduler.instance) // UI updates
  .subscribe(onNext: { processedBatch in
    self.updateUI(with: processedBatch)
  })

Why this works

• Batching limits per-item overhead and reduces UI thread hops.
• subscribeOn/observeOn separate concerns (CPU vs UI).
• flatMapLatest or onBackpressureLatest prevents queue buildup by prioritizing newest data; bounded buffers (time/count) ensure memory stays bounded.
• Trade-offs: dropping or sampling loses some raw data; acceptable when UI only needs recent/aggregated info. If no loss allowed, persist to local storage or slow down sensor sampling.

Approach

• Use a coroutine + Job to schedule delayed search 300ms after last keystroke, cancel previous Job when new input arrives, and post results to main thread.

Kotlin (Android) implementation

// In a ViewModel
class SearchViewModel(private val repo: SearchRepository) : ViewModel() {
  private var debounceJob: Job? = null
  private val _results = MutableLiveData<List<Result>>()
  val results: LiveData<List<Result>> = _results

  fun onQueryChanged(query: String) {
    // cancel previously scheduled work
    debounceJob?.cancel()
    // schedule new work after 300ms
    debounceJob = viewModelScope.launch {
      delay(300) // wait for typing to pause
      if (query.isBlank()) {
        _results.postValue(emptyList())
        return@launch
      }
      try {
        val res = withContext(Dispatchers.IO) { repo.search(query) } // network on IO
        // update LiveData on main thread
        _results.value = res
      } catch (e: Exception) {
        // handle error (post error state or empty)
      }
    }
  }

  override fun onCleared() {
    debounceJob?.cancel()
    super.onCleared()
  }
}

Why

• viewModelScope + Dispatchers ensure lifecycle safety and thread separation.
• Cancelling previous Job prevents stale network calls and wasted work.
• delay(300) implements debounce behaviour; updating LiveData on main thread keeps UI safe.

Edge cases

• Rapid input: previous jobs canceled
• Empty query: avoid unnecessary network calls
• Error handling & cancellation of in-flight network requests at repository layer (support for call.cancel() or retrofit Coroutine cancellation).

Approach (high level)
Design an event-driven architecture where UI events, sensors, network, and DB streams are modeled as Reactive streams. Use Flowable (RxJava) / Publishers when backpressure matters; Observable/Single for single-shot ops. Schedule work off the main thread and observe results on the main thread.

Threading

• Use subscribeOn for upstream work, observeOn for downstream consumers.
• Android (RxJava) example:

api.fetchItems()                // network call
  .subscribeOn(Schedulers.io()) // perform network on IO thread
  .observeOn(AndroidSchedulers.mainThread()) // update UI on main
  .subscribe(...);

• iOS (RxSwift) use ConcurrentDispatchQueueScheduler and MainScheduler.instance.

Backpressure
When producers can outrun consumers (high-frequency sensors, file/network bursts, infinite event sources), use Flowable + backpressure strategies:

• Examples where it matters:
  • Gyroscope/accelerometer producing 200Hz while UI updates ~16Hz
  • RecyclerView infinite scroll emitting dozens of load requests
  • Server push / websocket flooding
• Strategies:
  • buffer(n) with a bound and drop oldest/newest when full
  • onBackpressureDrop() to drop excess events
  • sample/throttleLast to reduce event rate
  • conflate (keep only latest) via onBackpressureLatest or sample

Flowable.create(emitter -> emitSensor(emitter),
                BackpressureStrategy.MISSING)
        .onBackpressureLatest()   // keep latest
        .observeOn(Schedulers.computation())
        .subscribe(...);

Long-running operations

• Offload CPU-heavy tasks (image processing, ML inference) to Schedulers.computation() or a dedicated ExecutorService.
• Use .timeout() to avoid indefinite blocking; cancel via Disposable/Subscription.
• For cancellable work, honor downstream cancellation: implement emitter.setCancellable.

Error handling

• Use onErrorResumeNext / onErrorReturn for recoverable errors.
• Use retryWhen with exponential backoff for transient network issues.
• Centralize error mapping to domain errors and surface user-friendly messages on main thread.

api.call()
  .retryWhen(errors -> errors
      .zipWith(Flowable.range(1,3), (e, i) -> i)
      .flatMap(i -> Flowable.timer((long)Math.pow(2,i), TimeUnit.SECONDS)))
  .onErrorResumeNext(Flowable.empty())

Putting it together (pattern)

• Source -> map/transform -> apply backpressure operators -> subscribeOn(IO/computation) -> observeOn(main) -> apply retries/timeouts -> subscribe with lifecycle-aware disposal (CompositeDisposable / DisposeBag).

This yields responsive UI, controlled resource usage, predictable backpressure behavior, and robust error handling appropriate for mobile constraints (battery, memory, responsiveness).