Hi there. Suppose I use Observable.create { … } to...

# rx

[ubu]

Hi there. Suppose I use Observable.create { … } to wrap some async operation, some data listener that receives updates about data changes. Like this:

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

	)
}

I need to call db.removeListener() when this observable is unsubscribed. How would you do that? Thanks.

Full disclosure: it’s about wrapping Firebase Realtime Database callbacks.