.computation

.io()

SQLiteException: database is locked

Observable.create { emitter ->
	
	val listener = db.addListener(
		...
		emitter.onNext(data)
// there's no onComplete() event here.

	)
}

db.removeListener()

/**
     * Sets a Cancellable on this emitter; any previous Disposable
     * or Cancellation will be unsubscribed/cancelled.
     * @param c the cancellable resource, null is allowed
     */
    void setCancellable(@Nullable Cancellable c);

subscriber.add(Subscriptions.create(new Action0() {
                    @Override
                    public void call() {
                        ref.removeEventListener(listener);
                    }
                }));

.setDisposable

/**
     * Sets a Disposable on this emitter; any previous Disposable
     * or Cancellation will be unsubscribed/cancelled.
     * @param d the disposable, null is allowed
     */
    void setDisposable(@Nullable Disposable d);

Observable

The emitter implementations will dispose/cancel this instance when the downstream cancels the flow or after the event generator logic calls Emitter.onError(Throwable), Emitter.onComplete() or when tryOnError(Throwable) succeeds.

Observable.create { emitter ->
    
    val listener = // . . emitter.onNext(data)
     db.addListener(listener)
    emitter.setDisposable { db.removeListener(listener) }
}