return proper result from Worker

2021-03-19 22:41:19 +01:00 · 2021-03-19 22:41:19 +01:00 · 8be3ccc989
commit 8be3ccc989
parent a2fbab02f9
10 changed files with 158 additions and 157 deletions
--- a/app/src/main/java/info/nightscout/androidaps/plugins/general/food/FoodPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/general/food/FoodPlugin.kt
@ -19,8 +19,6 @@ import info.nightscout.androidaps.utils.JsonHelper
 import info.nightscout.androidaps.utils.extensions.foodFromJson
 import info.nightscout.androidaps.utils.resources.ResourceHelper
 import info.nightscout.androidaps.utils.sharedPreferences.SP
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import org.json.JSONArray
 import org.json.JSONObject
 import javax.inject.Inject
@ -41,8 +39,6 @@ class FoodPlugin @Inject constructor(
    aapsLogger, resourceHelper, injector
 ) {

-    private val disposable = CompositeDisposable()
-
    // cannot be inner class because of needed injection
    class FoodWorker(
        context: Context,
@ -54,7 +50,6 @@ class FoodPlugin @Inject constructor(
        @Inject lateinit var repository: AppRepository
        @Inject lateinit var sp: SP
        @Inject lateinit var bundleStore: BundleStore
-        @Inject lateinit var foodPlugin: FoodPlugin

        init {
            (context.applicationContext as HasAndroidInjector).androidInjector().inject(this)
@ -83,25 +78,31 @@ class FoodPlugin @Inject constructor(
                            isValid = false
                        ).also { it.interfaceIDs.nightscoutId = JsonHelper.safeGetString(jsonFood, "_id") }

-                        foodPlugin.disposable += repository.runTransactionForResult(SyncFoodTransaction(delFood)).subscribe({ result ->
-                            result.invalidated.forEach { aapsLogger.debug(LTag.DATAFOOD, "Invalidated food ${it.interfaceIDs.nightscoutId}") }
-                        }, {
-                            aapsLogger.error(LTag.DATAFOOD, "Error while removing food", it)
-                            ret = Result.failure()
-                        })
+                        repository.runTransactionForResult(SyncFoodTransaction(delFood))
+                            .doOnError {
+                                aapsLogger.error(LTag.DATAFOOD, "Error while removing food", it)
+                                ret = Result.failure()
+                            }
+                            .blockingGet()
+                            .also {
+                                it.invalidated.forEach { f -> aapsLogger.debug(LTag.DATAFOOD, "Invalidated food ${f.interfaceIDs.nightscoutId}") }
+                            }
                    }

                    else     -> {
                        val food = foodFromJson(jsonFood)
                        if (food != null) {
-                            foodPlugin.disposable += repository.runTransactionForResult(SyncFoodTransaction(food)).subscribe({ result ->
-                                result.inserted.forEach { aapsLogger.debug(LTag.DATAFOOD, "Inserted food $it") }
-                                result.updated.forEach { aapsLogger.debug(LTag.DATAFOOD, "Updated food $it") }
-                                result.invalidated.forEach { aapsLogger.debug(LTag.DATAFOOD, "Invalidated food $it") }
-                            }, {
-                                aapsLogger.error(LTag.DATAFOOD, "Error while adding/updating food", it)
-                                ret = Result.failure()
-                            })
+                            repository.runTransactionForResult(SyncFoodTransaction(food))
+                                .doOnError {
+                                    aapsLogger.error(LTag.DATAFOOD, "Error while adding/updating food", it)
+                                    ret = Result.failure()
+                                }
+                                .blockingGet()
+                                .also { result ->
+                                    result.inserted.forEach { aapsLogger.debug(LTag.DATAFOOD, "Inserted food $it") }
+                                    result.updated.forEach { aapsLogger.debug(LTag.DATAFOOD, "Updated food $it") }
+                                    result.invalidated.forEach { aapsLogger.debug(LTag.DATAFOOD, "Invalidated food $it") }
+                                }
                        } else {
                            aapsLogger.error(LTag.DATAFOOD, "Error parsing food", jsonFood.toString())
                            ret = Result.failure()
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/DexcomPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/DexcomPlugin.kt
@ -93,6 +93,8 @@ class DexcomPlugin @Inject constructor(
        }

        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!dexcomPlugin.isEnabled(PluginType.BGSOURCE)) return Result.failure()
            val bundle = bundleStore.pickup(inputData.getLong(DataReceiver.STORE_KEY, -1))
                ?: return Result.failure()
@ -142,23 +144,25 @@ class DexcomPlugin @Inject constructor(
                        broadcastToXDrip(it)
                        if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false)) {
                            nsUpload.uploadBg(it, sourceSensor.text)
-                            //aapsLogger.debug("XXXXX: dbAdd $it")
                        }
+                        aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
                    }
                    result.updated.forEach {
                        broadcastToXDrip(it)
                        if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false)) {
                            nsUpload.updateBg(it, sourceSensor.text)
-                            //aapsLogger.debug("XXXXX: dpUpdate $it")
                        }
+                        aapsLogger.debug(LTag.BGSOURCE, "Updated bg $it")
                    }
                }, {
                    aapsLogger.error("Error while saving values from Dexcom App", it)
+                    ret = Result.failure()
                })
            } catch (e: Exception) {
                aapsLogger.error("Error while processing intent from Dexcom App", e)
+                ret = Result.failure()
            }
-            return Result.success()
+            return ret
        }
    }

--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/EversensePlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/EversensePlugin.kt
@ -4,7 +4,6 @@ import android.content.Context
 import androidx.work.Worker
 import androidx.work.WorkerParameters
 import dagger.android.HasAndroidInjector
-import info.nightscout.androidaps.Constants
 import info.nightscout.androidaps.R
 import info.nightscout.androidaps.database.AppRepository
 import info.nightscout.androidaps.database.entities.GlucoseValue
@ -24,8 +23,6 @@ import info.nightscout.androidaps.utils.DateUtil
 import info.nightscout.androidaps.utils.XDripBroadcast
 import info.nightscout.androidaps.utils.resources.ResourceHelper
 import info.nightscout.androidaps.utils.sharedPreferences.SP
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import java.util.*
 import javax.inject.Inject
 import javax.inject.Singleton
@ -48,13 +45,6 @@ class EversensePlugin @Inject constructor(

    override var sensorBatteryLevel = -1

-    private val disposable = CompositeDisposable()
-
-    override fun onStop() {
-        disposable.clear()
-        super.onStop()
-    }
-
    // cannot be inner class because of needed injection
    class EversenseWorker(
        context: Context,
@ -76,6 +66,8 @@ class EversensePlugin @Inject constructor(
        }

        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!eversensePlugin.isEnabled(PluginType.BGSOURCE)) return Result.failure()
            val bundle = bundleStore.pickup(inputData.getLong(DataReceiver.STORE_KEY, -1))
                ?: return Result.failure()
@ -116,15 +108,20 @@ class EversensePlugin @Inject constructor(
                            trendArrow = GlucoseValue.TrendArrow.NONE,
                            sourceSensor = GlucoseValue.SourceSensor.EVERSENSE
                        )
-                    eversensePlugin.disposable += repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null)).subscribe({ savedValues ->
-                        savedValues.inserted.forEach {
-                            broadcastToXDrip(it)
-                            if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
-                                nsUpload.uploadBg(it, GlucoseValue.SourceSensor.EVERSENSE.text)
+                    repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null))
+                        .doOnError {
+                            aapsLogger.error("Error while saving values from Eversense App", it)
+                            ret = Result.failure()
+                        }
+                        .blockingGet()
+                        .also { savedValues ->
+                            savedValues.inserted.forEach {
+                                broadcastToXDrip(it)
+                                if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
+                                    nsUpload.uploadBg(it, GlucoseValue.SourceSensor.EVERSENSE.text)
+                                aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
+                            }
                        }
-                    }, {
-                        aapsLogger.error("Error while saving values from Eversense App", it)
-                    })
                }
            }
            if (bundle.containsKey("calibrationGlucoseLevels")) {
@ -136,22 +133,29 @@ class EversensePlugin @Inject constructor(
                    aapsLogger.debug(LTag.BGSOURCE, "calibrationTimestamps" + Arrays.toString(calibrationTimestamps))
                    aapsLogger.debug(LTag.BGSOURCE, "calibrationRecordNumbers" + Arrays.toString(calibrationRecordNumbers))
                    for (i in calibrationGlucoseLevels.indices) {
-                        eversensePlugin.disposable += repository.runTransactionForResult(InsertTherapyEventIfNewTransaction(
+                        repository.runTransactionForResult(InsertTherapyEventIfNewTransaction(
                            timestamp = calibrationTimestamps[i],
                            type = TherapyEvent.Type.FINGER_STICK_BG_VALUE,
                            glucose = calibrationGlucoseLevels[i].toDouble(),
                            glucoseType = TherapyEvent.MeterType.FINGER,
                            glucoseUnit = TherapyEvent.GlucoseUnit.MGDL,
                            enteredBy = "AndroidAPS-Eversense"
-                        )).subscribe({ result ->
-                            result.inserted.forEach { nsUpload.uploadEvent(it) }
-                        }, {
-                            aapsLogger.error(LTag.BGSOURCE, "Error while saving therapy event", it)
-                        })
+                        ))
+                            .doOnError {
+                                aapsLogger.error(LTag.BGSOURCE, "Error while saving therapy event", it)
+                                ret = Result.failure()
+                            }
+                            .blockingGet()
+                            .also { result ->
+                                result.inserted.forEach {
+                                    nsUpload.uploadEvent(it)
+                                    aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
+                                }
+                            }
                    }
                }
            }
-            return Result.success()
+            return ret
        }
    }
 }
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/GlimpPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/GlimpPlugin.kt
@ -18,8 +18,6 @@ import info.nightscout.androidaps.plugins.general.nsclient.NSUpload
 import info.nightscout.androidaps.utils.XDripBroadcast
 import info.nightscout.androidaps.utils.resources.ResourceHelper
 import info.nightscout.androidaps.utils.sharedPreferences.SP
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import javax.inject.Inject
 import javax.inject.Singleton

@ -38,13 +36,6 @@ class GlimpPlugin @Inject constructor(
    aapsLogger, resourceHelper, injector
 ), BgSourceInterface {

-    private val disposable = CompositeDisposable()
-
-    override fun onStop() {
-        disposable.clear()
-        super.onStop()
-    }
-
    // cannot be inner class because of needed injection
    class GlimpWorker(
        context: Context,
@ -64,6 +55,8 @@ class GlimpPlugin @Inject constructor(
        }

        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!glimpPlugin.isEnabled(PluginType.BGSOURCE)) return Result.failure()
            aapsLogger.debug(LTag.BGSOURCE, "Received Glimp Data: $inputData}")
            val glucoseValues = mutableListOf<CgmSourceTransaction.TransactionGlucoseValue>()
@ -75,16 +68,21 @@ class GlimpPlugin @Inject constructor(
                trendArrow = GlucoseValue.TrendArrow.fromString(inputData.getString("myTrend")),
                sourceSensor = GlucoseValue.SourceSensor.GLIMP
            )
-            glimpPlugin.disposable += repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null)).subscribe({ savedValues ->
-                savedValues.inserted.forEach {
-                    broadcastToXDrip(it)
-                    if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
-                        nsUpload.uploadBg(it, GlucoseValue.SourceSensor.GLIMP.text)
+            repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null))
+                .doOnError {
+                    aapsLogger.error("Error while saving values from Glimp App", it)
+                    ret = Result.failure()
                }
-            }, {
-                aapsLogger.error("Error while saving values from Glimp App", it)
-            })
-            return Result.success()
+                .blockingGet()
+                .also { savedValues ->
+                    savedValues.inserted.forEach {
+                        broadcastToXDrip(it)
+                        if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
+                            nsUpload.uploadBg(it, GlucoseValue.SourceSensor.GLIMP.text)
+                        aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
+                    }
+                }
+            return ret
        }
    }
 }
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/MM640gPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/MM640gPlugin.kt
@ -20,8 +20,6 @@ import info.nightscout.androidaps.utils.DateUtil
 import info.nightscout.androidaps.utils.XDripBroadcast
 import info.nightscout.androidaps.utils.resources.ResourceHelper
 import info.nightscout.androidaps.utils.sharedPreferences.SP
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import org.json.JSONArray
 import org.json.JSONException
 import javax.inject.Inject
@ -41,13 +39,6 @@ class MM640gPlugin @Inject constructor(
    aapsLogger, resourceHelper, injector
 ), BgSourceInterface {

-    private val disposable = CompositeDisposable()
-
-    override fun onStop() {
-        disposable.clear()
-        super.onStop()
-    }
-
    // cannot be inner class because of needed injection
    class MM640gWorker(
        context: Context,
@ -69,6 +60,8 @@ class MM640gPlugin @Inject constructor(
        }

        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!mM640gPlugin.isEnabled(PluginType.BGSOURCE)) return Result.failure()
            val collection = inputData.getString("collection") ?: return Result.failure()
            if (collection == "entries") {
@ -93,21 +86,27 @@ class MM640gPlugin @Inject constructor(
                                else  -> aapsLogger.debug(LTag.BGSOURCE, "Unknown entries type: $type")
                            }
                        }
-                        mM640gPlugin.disposable += repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null)).subscribe({ savedValues ->
+                        repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null))
+                            .doOnError {
+                                aapsLogger.error("Error while saving values from Eversense App", it)
+                                ret = Result.failure()
+                            }
+                            .blockingGet()
+                            .also { savedValues ->
                            savedValues.all().forEach {
                                broadcastToXDrip(it)
                                if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
                                    nsUpload.uploadBg(it, GlucoseValue.SourceSensor.MM_600_SERIES.text)
+                                aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
                            }
-                        }, {
-                            aapsLogger.error("Error while saving values from Eversense App", it)
-                        })
+                        }
                    } catch (e: JSONException) {
                        aapsLogger.error("Exception: ", e)
+                        ret = Result.failure()
                    }
                }
            }
-            return Result.success()
+            return ret
        }
    }
 }
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/NSClientSourcePlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/NSClientSourcePlugin.kt
@ -22,8 +22,6 @@ import info.nightscout.androidaps.utils.DateUtil
 import info.nightscout.androidaps.utils.XDripBroadcast
 import info.nightscout.androidaps.utils.resources.ResourceHelper
 import info.nightscout.androidaps.utils.sharedPreferences.SP
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import org.json.JSONArray
 import org.json.JSONObject
 import javax.inject.Inject
@ -55,13 +53,6 @@ class NSClientSourcePlugin @Inject constructor(
        }
    }

-    private val disposable = CompositeDisposable()
-
-    override fun onStop() {
-        disposable.clear()
-        super.onStop()
-    }
-
    override fun advancedFilteringSupported(): Boolean {
        return isAdvancedFilteringEnabled
    }
@ -113,7 +104,10 @@ class NSClientSourcePlugin @Inject constructor(
            )
        }

+        @Suppress("SpellCheckingInspection")
        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!nsClientSourcePlugin.isEnabled() && !sp.getBoolean(R.string.key_ns_autobackfill, true) && !dexcomPlugin.isEnabled()) return Result.failure()
            try {
                val glucoseValues = mutableListOf<CgmSourceTransaction.TransactionGlucoseValue>()
@ -127,27 +121,31 @@ class NSClientSourcePlugin @Inject constructor(
                    for (i in 0 until jsonArray.length())
                        glucoseValues += toGv(jsonArray.getJSONObject(i))
                }
-                nsClientSourcePlugin.disposable += repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null, !nsClientSourcePlugin.isEnabled())).subscribe({ result ->
-                    result.updated.forEach {
-                        //aapsLogger.debug("XXXXX: Updated $it")
-                        broadcastToXDrip(it)
-                        nsClientSourcePlugin.detectSource(it)
+                repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null, !nsClientSourcePlugin.isEnabled()))
+                    .doOnError {
+                        aapsLogger.error("Error while saving values from NSClient App", it)
+                        ret = Result.failure()
                    }
-                    result.inserted.forEach {
-                        //aapsLogger.debug("XXXXX: Inserted $it")
-                        broadcastToXDrip(it)
-                        nsClientSourcePlugin.detectSource(it)
+                    .blockingGet()
+                    .also { result ->
+                        result.updated.forEach {
+                            broadcastToXDrip(it)
+                            nsClientSourcePlugin.detectSource(it)
+                            aapsLogger.debug(LTag.BGSOURCE, "Updated bg $it")
+                        }
+                        result.inserted.forEach {
+                            broadcastToXDrip(it)
+                            nsClientSourcePlugin.detectSource(it)
+                            aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
+                        }
                    }
-                }, {
-                    aapsLogger.error("Error while saving values from NSClient App", it)
-                })
            } catch (e: Exception) {
                aapsLogger.error("Unhandled exception", e)
-                return Result.failure()
+                ret = Result.failure()
            }
            // Objectives 0
            sp.putBoolean(R.string.key_ObjectivesbgIsAvailableInNS, true)
-            return Result.success()
+            return ret
        }
    }
 }
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/PoctechPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/PoctechPlugin.kt
@ -20,8 +20,6 @@ import info.nightscout.androidaps.utils.JsonHelper.safeGetString
 import info.nightscout.androidaps.utils.XDripBroadcast
 import info.nightscout.androidaps.utils.resources.ResourceHelper
 import info.nightscout.androidaps.utils.sharedPreferences.SP
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import org.json.JSONArray
 import org.json.JSONException
 import javax.inject.Inject
@ -42,13 +40,6 @@ class PoctechPlugin @Inject constructor(
    aapsLogger, resourceHelper, injector
 ), BgSourceInterface {

-    private val disposable = CompositeDisposable()
-
-    override fun onStop() {
-        disposable.clear()
-        super.onStop()
-    }
-
    // cannot be inner class because of needed injection
    class PoctechWorker(
        context: Context,
@ -68,6 +59,8 @@ class PoctechPlugin @Inject constructor(
        }

        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!poctechPlugin.isEnabled(PluginType.BGSOURCE)) return Result.failure()
            aapsLogger.debug(LTag.BGSOURCE, "Received Poctech Data $inputData")
            try {
@ -86,20 +79,25 @@ class PoctechPlugin @Inject constructor(
                        sourceSensor = GlucoseValue.SourceSensor.POCTECH_NATIVE
                    )
                }
-                poctechPlugin.disposable += repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null)).subscribe({ savedValues ->
-                    savedValues.inserted.forEach {
-                        broadcastToXDrip(it)
-                        if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
-                            nsUpload.uploadBg(it, GlucoseValue.SourceSensor.POCTECH_NATIVE.text)
+                repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null))
+                    .doOnError {
+                        aapsLogger.error("Error while saving values from Poctech App", it)
+                        ret = Result.failure()
+                    }
+                    .blockingGet()
+                    .also { savedValues ->
+                        savedValues.inserted.forEach {
+                            broadcastToXDrip(it)
+                            if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
+                                nsUpload.uploadBg(it, GlucoseValue.SourceSensor.POCTECH_NATIVE.text)
+                            aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
+                        }
                    }
-                }, {
-                    aapsLogger.error("Error while saving values from Poctech App", it)
-                })
            } catch (e: JSONException) {
                aapsLogger.error("Exception: ", e)
-                return Result.failure()
+                ret = Result.failure()
            }
-            return Result.success()
+            return ret
        }
    }
 }
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/RandomBgPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/RandomBgPlugin.kt
@ -112,6 +112,7 @@ class RandomBgPlugin @Inject constructor(
                xDripBroadcast(it)
                if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
                    nsUpload.uploadBg(it, GlucoseValue.SourceSensor.RANDOM.text)
+                aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
            }
        }, {
            aapsLogger.error(LTag.BGSOURCE, "Error while saving values from Random plugin", it)
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/TomatoPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/TomatoPlugin.kt
@ -18,8 +18,6 @@ import info.nightscout.androidaps.plugins.general.nsclient.NSUpload
 import info.nightscout.androidaps.utils.XDripBroadcast
 import info.nightscout.androidaps.utils.resources.ResourceHelper
 import info.nightscout.androidaps.utils.sharedPreferences.SP
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import javax.inject.Inject
 import javax.inject.Singleton

@ -39,13 +37,6 @@ class TomatoPlugin @Inject constructor(
    aapsLogger, resourceHelper, injector
 ), BgSourceInterface {

-    private val disposable = CompositeDisposable()
-
-    override fun onStop() {
-        disposable.clear()
-        super.onStop()
-    }
-
    // cannot be inner class because of needed injection
    class TomatoWorker(
        context: Context,
@ -64,7 +55,10 @@ class TomatoPlugin @Inject constructor(
            (context.applicationContext as HasAndroidInjector).androidInjector().inject(this)
        }

+        @Suppress("SpellCheckingInspection")
        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!tomatoPlugin.isEnabled(PluginType.BGSOURCE)) return Result.failure()
            val glucoseValues = mutableListOf<CgmSourceTransaction.TransactionGlucoseValue>()
            glucoseValues += CgmSourceTransaction.TransactionGlucoseValue(
@ -75,16 +69,21 @@ class TomatoPlugin @Inject constructor(
                trendArrow = GlucoseValue.TrendArrow.NONE,
                sourceSensor = GlucoseValue.SourceSensor.LIBRE_1_TOMATO
            )
-            tomatoPlugin.disposable += repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null)).subscribe({ savedValues ->
-                savedValues.inserted.forEach {
-                    broadcastToXDrip(it)
-                    if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
-                        nsUpload.uploadBg(it, GlucoseValue.SourceSensor.LIBRE_1_TOMATO.text)
+            repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null))
+                .doOnError {
+                    aapsLogger.error("Error while saving values from Tomato App", it)
+                    ret = Result.failure()
                }
-            }, {
-                aapsLogger.error("Error while saving values from Tomato App", it)
-            })
-            return Result.success()
+                .blockingGet()
+                .also { savedValues ->
+                    savedValues.inserted.forEach {
+                        broadcastToXDrip(it)
+                        if (sp.getBoolean(R.string.key_dexcomg5_nsupload, false))
+                            nsUpload.uploadBg(it, GlucoseValue.SourceSensor.LIBRE_1_TOMATO.text)
+                        aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
+                    }
+                }
+            return ret
        }
    }
 }
--- a/app/src/main/java/info/nightscout/androidaps/plugins/source/XdripPlugin.kt
+++ b/app/src/main/java/info/nightscout/androidaps/plugins/source/XdripPlugin.kt
@ -18,8 +18,6 @@ import info.nightscout.androidaps.receivers.BundleStore
 import info.nightscout.androidaps.receivers.DataReceiver
 import info.nightscout.androidaps.services.Intents
 import info.nightscout.androidaps.utils.resources.ResourceHelper
-import io.reactivex.disposables.CompositeDisposable
-import io.reactivex.rxkotlin.plusAssign
 import javax.inject.Inject
 import javax.inject.Singleton

@ -55,19 +53,13 @@ class XdripPlugin @Inject constructor(
        ).any { it == glucoseValue.sourceSensor }
    }

-    private val disposable = CompositeDisposable()
-
-    override fun onStop() {
-        disposable.clear()
-        super.onStop()
-    }
-
    // cannot be inner class because of needed injection
    class XdripWorker(
        context: Context,
        params: WorkerParameters
    ) : Worker(context, params) {

+        @Inject lateinit var aapsLogger: AAPSLogger
        @Inject lateinit var xdripPlugin: XdripPlugin
        @Inject lateinit var repository: AppRepository
        @Inject lateinit var bundleStore: BundleStore
@ -77,11 +69,13 @@ class XdripPlugin @Inject constructor(
        }

        override fun doWork(): Result {
+            var ret = Result.success()
+
            if (!xdripPlugin.isEnabled(PluginType.BGSOURCE)) return Result.failure()
            val bundle = bundleStore.pickup(inputData.getLong(DataReceiver.STORE_KEY, -1))
                ?: return Result.failure()
-            
-            xdripPlugin.aapsLogger.debug(LTag.BGSOURCE, "Received xDrip data: $bundle")
+
+            aapsLogger.debug(LTag.BGSOURCE, "Received xDrip data: $bundle")
            val glucoseValues = mutableListOf<CgmSourceTransaction.TransactionGlucoseValue>()
            glucoseValues += CgmSourceTransaction.TransactionGlucoseValue(
                timestamp = bundle.getLong(Intents.EXTRA_TIMESTAMP, 0),
@ -92,15 +86,20 @@ class XdripPlugin @Inject constructor(
                sourceSensor = GlucoseValue.SourceSensor.fromString(bundle.getString(Intents.XDRIP_DATA_SOURCE_DESCRIPTION)
                    ?: "")
            )
-            xdripPlugin.disposable += repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null)).subscribe({ savedValues ->
-                savedValues.all().forEach {
-                    xdripPlugin.detectSource(it)
+            repository.runTransactionForResult(CgmSourceTransaction(glucoseValues, emptyList(), null))
+                .doOnError {
+                    aapsLogger.error(LTag.BGSOURCE, "Error while saving values from Eversense App", it)
+                    ret = Result.failure()
+                }
+                .blockingGet()
+                .also { savedValues ->
+                    savedValues.all().forEach {
+                        xdripPlugin.detectSource(it)
+                        aapsLogger.debug(LTag.BGSOURCE, "Inserted bg $it")
+                    }
                }
-            }, {
-                xdripPlugin.aapsLogger.error(LTag.BGSOURCE, "Error while saving values from Eversense App", it)
-            })
            xdripPlugin.sensorBatteryLevel = bundle.getInt(Intents.EXTRA_SENSOR_BATTERY, -1)
-            return Result.success()
+            return ret
        }
    }
 }