stop on CountDownLatch

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/OmnipodDashPumpPlugin.kt

@@ -71,7 +71,7 @@ class OmnipodDashPumpPlugin @Inject constructor(
 ) : PumpPluginBase(pluginDescription, injector, aapsLogger, resourceHelper, commandQueue), Pump {
     @Volatile var bolusCanceled = false
     private val handler: Handler = Handler(Looper.getMainLooper())
-    lateinit private var statusChecker: Runnable
+    private lateinit var statusChecker: Runnable
     var nextPodWarningCheck : Long = 0
 
     companion object {

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/comm/OmnipodDashBleManager.kt

@@ -1,10 +1,12 @@
 package info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm
 
+import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.session.Connection
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.session.ConnectionState
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.event.PodEvent
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.command.base.Command
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.Response
 import io.reactivex.Observable
+import java.util.concurrent.CountDownLatch
 import kotlin.reflect.KClass
 
 interface OmnipodDashBleManager {
@@ -13,7 +15,11 @@ interface OmnipodDashBleManager {
 
     fun getStatus(): ConnectionState
 
-    fun connect(): Observable<PodEvent>
+    // used for sync connections
+    fun connect(timeoutMs: Long = Connection.BASE_CONNECT_TIMEOUT_MS*3): Observable<PodEvent>
+
+    // used for async connections
+    fun connect(stopConnectionLatch: CountDownLatch): Observable<PodEvent>
 
     fun pairNewPod(): Observable<PodEvent>
 

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/comm/OmnipodDashBleManagerImpl.kt

@@ -16,6 +16,7 @@ import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.command.b
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.response.Response
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.state.OmnipodDashPodStateManager
 import io.reactivex.Observable
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.atomic.AtomicBoolean
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -100,8 +101,19 @@ class OmnipodDashBleManagerImpl @Inject constructor(
         return connection?.let { it.connectionState() }
             ?: NotConnected
     }
+    // used for sync connections
+    override fun connect(timeoutMs: Long): Observable<PodEvent> {
+        return connect(ConnectionWaitCondition(timeoutMs = timeoutMs))
+    }
 
-    override fun connect(): Observable<PodEvent> = Observable.create { emitter ->
+    // used for async connections
+    override fun connect(stopConnectionLatch: CountDownLatch): Observable<PodEvent> {
+        return connect(ConnectionWaitCondition(stopConnection = stopConnectionLatch))
+    }
+
+    private fun connect(connectionWaitCond: ConnectionWaitCondition): Observable<PodEvent> = Observable
+        .create {
+            emitter ->
         if (!busy.compareAndSet(false, true)) {
             throw BusyException()
         }
@@ -121,20 +133,7 @@ class OmnipodDashBleManagerImpl @Inject constructor(
                 return@create
             }
 
-            // two retries
+            conn.connect(connectionWaitCond)
-            for (i in 1..MAX_NUMBER_OF_CONNECTION_ATTEMPTS) {
-                try {
-                    // wait i * CONNECTION_TIMEOUT
-                    conn.connect(4)
-                    break
-                } catch (e: Exception) {
-                    aapsLogger.warn(LTag.PUMPBTCOMM, "connect error=$e")
-                    if (i == MAX_NUMBER_OF_CONNECTION_ATTEMPTS) {
-                        emitter.onError(e)
-                        return@create
-                    }
-                }
-            }
 
             emitter.onNext(PodEvent.BluetoothConnected(podAddress))
             emitter.onNext(PodEvent.EstablishingSession)
@@ -207,6 +206,7 @@ class OmnipodDashBleManagerImpl @Inject constructor(
             val podDevice = bluetoothAdapter.getRemoteDevice(podAddress)
             val conn = Connection(podDevice, aapsLogger, context, podState)
             connection = conn
+            conn.connect(ConnectionWaitCondition(timeoutMs = 3 * Connection.BASE_CONNECT_TIMEOUT_MS))
             emitter.onNext(PodEvent.BluetoothConnected(podAddress))
 
             emitter.onNext(PodEvent.Pairing)

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/comm/callbacks/BleCommCallbacks.kt

@@ -53,12 +53,13 @@ class BleCommCallbacks(
         }
     }
 
-    fun waitForConnection(timeoutMs: Int) {
+    fun waitForConnection(timeoutMs: Long): Boolean {
         try {
-            connected.await(timeoutMs.toLong(), TimeUnit.MILLISECONDS)
+            connected.await(timeoutMs, TimeUnit.MILLISECONDS)
         } catch (e: InterruptedException) {
             aapsLogger.warn(LTag.PUMPBTCOMM, "Interrupted while waiting for Connection")
         }
+        return connected.count == 0L
     }
 
     fun startServiceDiscovery() {

omnipod-dash/src/main/java/info/nightscout/androidaps/plugins/pump/omnipod/dash/driver/comm/session/Connection.kt

@@ -15,13 +15,14 @@ import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.callback
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.endecrypt.EnDecrypt
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.exceptions.ConnectException
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.exceptions.FailedToConnectException
-import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.io.BleSendSuccess
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.io.CharacteristicType
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.io.CmdBleIO
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.io.DataBleIO
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.io.IncomingPackets
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.comm.message.MessageIO
 import info.nightscout.androidaps.plugins.pump.omnipod.dash.driver.pod.state.OmnipodDashPodStateManager
+import java.lang.IllegalArgumentException
+import java.util.concurrent.CountDownLatch
 
 sealed class ConnectionState
 
@@ -30,6 +31,17 @@ object Handshaking: ConnectionState()
 object Connected : ConnectionState()
 object NotConnected : ConnectionState()
 
+data class ConnectionWaitCondition(val timeoutMs: Long?=null, val stopConnection: CountDownLatch?=null) {
+    init {
+        if (timeoutMs == null && stopConnection == null) {
+            throw IllegalArgumentException("One of timeoutMs or stopConnection has to be non null")
+        }
+        if (timeoutMs != null && stopConnection != null) {
+            throw IllegalArgumentException("One of timeoutMs or stopConnection has to be null")
+        }
+    }
+}
+
 class Connection(
     private val podDevice: BluetoothDevice,
     private val aapsLogger: AAPSLogger,
@@ -50,8 +62,9 @@ class Connection(
     @Volatile
     var msgIO: MessageIO? = null
 
-    fun connect(timeoutMultiplier: Int) {
+    fun connect(connectionWaitCond: ConnectionWaitCondition) {
         aapsLogger.debug("Connecting")
+
         podState.bluetoothConnectionState = OmnipodDashPodStateManager.BluetoothConnectionState.CONNECTING
         val autoConnect = false
         val gatt = gattConnection ?:
@@ -60,8 +73,7 @@ class Connection(
         if (!gatt.connect()) {
             throw FailedToConnectException("connect() returned false")
         }
-        // TODO: loop
+        if (waitForConnection(connectionWaitCond) !is Connected) {
-        if (waitForConnection(timeoutMultiplier) !is Connected) {
             podState.bluetoothConnectionState = OmnipodDashPodStateManager.BluetoothConnectionState.DISCONNECTED
             throw FailedToConnectException(podDevice.address)
         }
@@ -103,9 +115,18 @@ class Connection(
         msgIO = null
     }
 
-    private fun waitForConnection(timeoutMultiplier: Int): ConnectionState {
+    private fun waitForConnection(connectionWaitCond: ConnectionWaitCondition): ConnectionState {
         try {
-            bleCommCallbacks.waitForConnection(BASE_CONNECT_TIMEOUT_MS * timeoutMultiplier)
+            connectionWaitCond.timeoutMs?.let {
+                bleCommCallbacks.waitForConnection(it)
+            }
+            connectionWaitCond.stopConnection?.let {
+                while (!bleCommCallbacks.waitForConnection(300)) {
+                    if (it.count == 0L) {
+                        return NotConnected
+                    }
+                }
+            }
         } catch (e: InterruptedException) {
             // We are still going to check if connection was successful
             aapsLogger.info(LTag.PUMPBTCOMM, "Interrupted while waiting for connection")
@@ -158,6 +179,6 @@ class Connection(
     }
 
     companion object {
-        private const val BASE_CONNECT_TIMEOUT_MS = 10000
+        const val BASE_CONNECT_TIMEOUT_MS = 10000L
     }
 }