Hey guys, I’m trying to migrate RxJava code deali...

# coroutines

[Yuri]

Hey guys,

I’m trying to migrate RxJava code dealing with Android SharedPreferences to Kotlin Flow.
How correct migration should look like?

private val keyChangesObservable: Observable<String>

  init {
    keyChangesObservable = io.reactivex.Observable.create(ObservableOnSubscribe<String> { emitter ->
      val listener = SharedPreferences.OnSharedPreferenceChangeListener { _, key -> emitter.onNext(key) }
      emitter.setCancellable { prefs.unregisterOnSharedPreferenceChangeListener(listener) }
      prefs.registerOnSharedPreferenceChangeListener(listener)
    }).share()
  }

[louiscad]

Here you go: https://github.com/LouisCAD/Splitties/blob/c395dcbf9d19fceae67a520215c2ab96fd138390/modules/preferences/src/androidMain/kotlin/splitties/preferences/Changes.kt#L14

[Pablichjenkov]

Quick Question at Louis, is runCatching{ offer(event)} picking up any exception thrown all the way down the Flow stream. Or just catching the operation of offering the event to some EventQueue. What would be the difference of using send(...)?

[gildor]

runCatching in this case would just swallow exception

[Pablichjenkov]

That’s right but what exception?
Any Exception all the way down the stream? It does not look like so.
It looks like is just to guard against possible ClosedChannel exceptions but nothing else, is that it?

[gildor]

yes, it’s probably against closedchannel exception

[gildor]

actually not sure that it needed here, channel is closed on flow cancellation in this example and unregister listener, so it should be safe

[Pablichjenkov]

🤔

[louiscad]

It should be safe, but I'm not sure and I've been bitten by this throwing unexpectedly in others use cases using Channels in the past, so I'm guarding myself against this bad design until kotlinx.coroutines provides an alternative.

I opened an issue for that a while ago:
https://github.com/Kotlin/kotlinx.coroutines/issues/974