Example of wrapping OnSharedPreferenceChangeListener in RxJava2 observable

my-preferences.kt

// package and imports


@Singleton
class MyPreferences
@Inject constructor(val context: Context) {

    companion object {
        private const val PREF_FILE_NAME = "pg_cfg"

        private const val PREF_KEY_UNREAD_MESSAGES_COUNT = "PREF_KEY_UNREAD_MESSAGES_COUNT"

        val currentLocale: String
            get() = Locale.getDefault().language.toLowerCase()
    }

    private val gson: Gson = GsonBuilder().create()
    private val sharedPreferences: SharedPreferences = context.getSharedPreferences(PREF_FILE_NAME, MODE_PRIVATE)


    // unread messages


    var unreadMessagesCount: MutableMap<Long, Int>
        get() {
            return if (sharedPreferences.contains(PREF_KEY_UNREAD_MESSAGES_COUNT)) {
                val type = object : TypeToken<Map<Long, Int>>() {}.type
                gson.fromJson(sharedPreferences.getString(PREF_KEY_UNREAD_MESSAGES_COUNT, "{}"), type)
            } else HashMap()
        }
        set(counts) = sharedPreferences.edit().putString(PREF_KEY_UNREAD_MESSAGES_COUNT, gson.toJson(counts)).apply()

    fun unreadMessagesCountObservable(): Observable<MutableMap<Long, Int>> = Observable
            .create<MutableMap<Long, Int>> { emitter ->
                val prefListener = SharedPreferences.OnSharedPreferenceChangeListener { _, key ->
                    if (!emitter.isDisposed && key == PREF_KEY_UNREAD_MESSAGES_COUNT) emitter.onNext(unreadMessagesCount)
                }
                sharedPreferences.registerOnSharedPreferenceChangeListener(prefListener)
                emitter.setCancellable { sharedPreferences.unregisterOnSharedPreferenceChangeListener(prefListener) }
                emitter.onNext(unreadMessagesCount)
            }

}