Add support for automatic resetting incremental backoff when network status changes

CHANGELOG.md

@@ -28,6 +28,7 @@ if the content is the same. Custom implementations of these methods will be resp
 * Support for performing geospatial queries using the new classes: `GeoPoint`, `GeoCircle`, `GeoBox`, and `GeoPolygon`. See `GeoPoint` documentation on how to persist locations. (Issue [#1403](https://github.com/realm/realm-kotlin/pull/1403))
 * Support for automatic resolution of embedded object constraints during migration through `RealmConfiguration.Builder.migration(migration: AutomaticSchemaMigration, resolveEmbeddedObjectConstraints: Boolean)`. (Issue [#1464](https://github.com/realm/realm-kotlin/issues/1464)
 * [Sync] Add support for customizing authorization headers and adding additional custom headers to all Atlas App service requests with `AppConfiguration.Builder.authorizationHeaderName()` and `AppConfiguration.Builder.addCustomRequestHeader(...)`. (Issue [#1453](https://github.com/realm/realm-kotlin/pull/1453))
+* [Sync] Added support for manually triggering a reconnect attempt for Device Sync. This is done through a new `App.Sync.reconnect()` method. This method is also now called automatically when a mobile device toggles off airplane mode. (Issue [#1479](https://github.com/realm/realm-kotlin/issues/1479))
 
 ### Fixed
 * None.

packages/cinterop/src/commonMain/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -707,6 +707,11 @@ expect object RealmInterop {
         callback: AppCallback<String>
     )
 
+    // Sync Client
+    fun realm_app_sync_client_reconnect(app: RealmAppPointer)
+    fun realm_app_sync_client_has_sessions(app: RealmAppPointer): Boolean
+    fun realm_app_sync_client_wait_for_sessions_to_terminate(app: RealmAppPointer)
+
     // Sync config
     fun realm_config_set_sync_config(
         realmConfiguration: RealmConfigurationPointer,

packages/cinterop/src/jvm/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -1558,6 +1558,17 @@ actual object RealmInterop {
         )
     }
 
+    actual fun realm_app_sync_client_reconnect(app: RealmAppPointer) {
+        realmc.realm_app_sync_client_reconnect(app.cptr())
+    }
+    actual fun realm_app_sync_client_has_sessions(app: RealmAppPointer): Boolean {
+        return realmc.realm_app_sync_client_has_sessions(app.cptr())
+    }
+
+    actual fun realm_app_sync_client_wait_for_sessions_to_terminate(app: RealmAppPointer) {
+        realmc.realm_app_sync_client_wait_for_sessions_to_terminate(app.cptr())
+    }
+
     actual fun realm_sync_config_new(user: RealmUserPointer, partition: String): RealmSyncConfigurationPointer {
         return LongPointerWrapper<RealmSyncConfigT>(realmc.realm_sync_config_new(user.cptr(), partition)).also { ptr ->
             // Stop the session immediately when the Realm is closed, so the lifecycle of the

packages/cinterop/src/nativeDarwin/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -2948,6 +2948,17 @@ actual object RealmInterop {
         }
     }
 
+    actual fun realm_app_sync_client_reconnect(app: RealmAppPointer) {
+        realm_wrapper.realm_app_sync_client_reconnect(app.cptr())
+    }
+    actual fun realm_app_sync_client_has_sessions(app: RealmAppPointer): Boolean {
+        return realm_wrapper.realm_app_sync_client_has_sessions(app.cptr())
+    }
+
+    actual fun realm_app_sync_client_wait_for_sessions_to_terminate(app: RealmAppPointer) {
+        realm_wrapper.realm_app_sync_client_wait_for_sessions_to_terminate(app.cptr())
+    }
+
     actual fun realm_config_set_sync_config(realmConfiguration: RealmConfigurationPointer, syncConfiguration: RealmSyncConfigurationPointer) {
         realm_wrapper.realm_config_set_sync_config(realmConfiguration.cptr(), syncConfiguration.cptr())
     }

packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/platform/SystemUtils.kt

@@ -131,7 +131,7 @@ public expect fun epochInSeconds(): Long
 /**
  * Returns a RealmInstant representing the time that has passed since the Unix epoch.
  */
-internal expect fun currentTime(): RealmInstant
+public expect fun currentTime(): RealmInstant
 
 /**
  * Returns the type of a mutable property.

packages/library-base/src/jvm/kotlin/io/realm/kotlin/internal/platform/SystemUtils.kt

@@ -28,7 +28,7 @@ public actual fun epochInSeconds(): Long =
 * Since internalNow() should only logically return a value after the Unix epoch, it is safe to create a RealmInstant
 * without considering having to pass negative nanoseconds.
 */
-internal actual fun currentTime(): RealmInstant {
+public actual fun currentTime(): RealmInstant {
     val jtInstant = systemUTC().instant()
     return RealmInstantImpl(jtInstant.epochSecond, jtInstant.nano)
 }

packages/library-base/src/nativeDarwin/kotlin/io/realm/kotlin/internal/platform/SystemUtils.kt

@@ -68,7 +68,7 @@ public actual fun epochInSeconds(): Long =
  * without considering having to pass negative nanoseconds.
  */
 @Suppress("MagicNumber")
-internal actual fun currentTime(): RealmInstant {
+public actual fun currentTime(): RealmInstant {
     val secs: Double = NSDate().timeIntervalSince1970
     return when {
         // We can't convert the MIN value to ms - it is initialized with Long.MIN_VALUE and

packages/library-sync/src/androidMain/AndroidManifest.xml

@@ -16,8 +16,22 @@
   -->
 
 <manifest xmlns:android="http://schemas.android.com/apk/res/android"
+    xmlns:tools="http://schemas.android.com/tools"
     package="io.realm.kotlin.mongodb">
 
+    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
     <uses-permission android:name="android.permission.INTERNET" />
 
+    <application>
+        <provider
+            android:name="androidx.startup.InitializationProvider"
+            tools:node="merge"
+            android:authorities="${applicationId}.androidx-startup"
+            android:exported="false">
+            <meta-data
+                android:name="io.realm.kotlin.mongodb.internal.RealmSyncInitializer"
+                android:value="androidx.startup" />
+        </provider>
+    </application>
+
 </manifest>

packages/library-sync/src/androidMain/kotlin/io/realm/kotlin/mongodb/internal/NetworkStateObserver.kt

@@ -0,0 +1,6 @@
+package io.realm.kotlin.mongodb.internal
+
+internal actual fun registerSystemNetworkObserver() {
+    // Registering network state listeners are done in io.realm.kotlin.mongodb.RealmSyncInitializer
+    // so we do not have to store the Android Context.
+}

packages/library-sync/src/androidMain/kotlin/io/realm/kotlin/mongodb/internal/RealmSyncInitializer.kt

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2021 Realm Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.realm.kotlin.mongodb.internal
+
+import android.content.BroadcastReceiver
+import android.content.Context
+import android.content.Intent
+import android.content.IntentFilter
+import android.net.ConnectivityManager
+import android.net.ConnectivityManager.NetworkCallback
+import android.net.Network
+import android.net.NetworkCapabilities
+import android.net.NetworkInfo
+import android.net.NetworkRequest
+import android.os.Build
+import androidx.startup.Initializer
+import io.realm.kotlin.internal.RealmInitializer
+import io.realm.kotlin.log.RealmLog
+
+/**
+ * An **initializer** for Sync specific functionality that does not fit into the `RealmInitializer`
+ * in cinterop.o allow Realm to access context properties.
+ */
+class RealmSyncInitializer : Initializer<Context> {
+
+    companion object {
+        @Suppress("DEPRECATION") // Should only be called below API 21
+        fun isConnected(cm: ConnectivityManager?): Boolean {
+            return cm?.let {
+                val networkInfo: NetworkInfo? = cm.activeNetworkInfo
+                networkInfo != null && networkInfo.isConnectedOrConnecting || isEmulator()
+            } ?: true
+        }
+
+        // Credit: http://stackoverflow.com/questions/2799097/how-can-i-detect-when-an-android-application-is-running-in-the-emulator
+        fun isEmulator(): Boolean {
+            return Build.FINGERPRINT.startsWith("generic") ||
+                Build.FINGERPRINT.startsWith("unknown") ||
+                Build.MODEL.contains("google_sdk") ||
+                Build.MODEL.contains("Emulator") ||
+                Build.MODEL.contains("Android SDK built for x86") ||
+                Build.MANUFACTURER.contains("Genymotion") ||
+                (Build.BRAND.startsWith("generic") && Build.DEVICE.startsWith("generic")) ||
+                "google_sdk" == Build.PRODUCT
+        }
+    }
+
+    private var connectivityManager: ConnectivityManager? = null
+
+    override fun create(context: Context): Context {
+        connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager?
+        // There has been a fair amount of changes and deprecations with regard to how to listen
+        // to the network status. ConnectivityManager#CONNECTIVITY_ACTION was deprecated in API 28
+        // but ConnectivityManager.NetworkCallback became available a lot sooner in API 21, so
+        // we default to this as soon as possible.
+        //
+        // On later versions of Android (need reference), these callbacks will also only trigger
+        // if the app is in the foreground.
+        //
+        // The current implementation is a best-effort in detecting when the network is available
+        // again.
+        //
+        // See https://developer.android.com/training/basics/network-ops/reading-network-state
+        // See https://developer.android.com/reference/android/net/ConnectivityManager#CONNECTIVITY_ACTION
+        // See https://developer.android.com/reference/android/net/ConnectivityManager.NetworkCallback
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP /* 21 */) {
+            val request = NetworkRequest.Builder()
+                .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
+            if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.LOLLIPOP /* 23 */) {
+                request.addCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)
+            }
+            @Suppress("invisible_member", "invisible_reference")
+            RealmLog.info("Register ConnectivityManager network callbacks")
+            connectivityManager?.registerNetworkCallback(
+                request.build(),
+                object : NetworkCallback() {
+                    override fun onAvailable(network: Network) {
+                        NetworkStateObserver.notifyConnectionChange(true)
+                    }
+                    override fun onUnavailable() {
+                        NetworkStateObserver.notifyConnectionChange(false)
+                    }
+                }
+            )
+        } else {
+            @Suppress("invisible_member", "invisible_reference")
+            RealmLog.info("Register BroadcastReceiver connectivity callbacks")
+            @Suppress("DEPRECATION")
+            context.registerReceiver(
+                object : BroadcastReceiver() {
+                    override fun onReceive(context: Context?, intent: Intent?) {
+                        val isConnected: Boolean = isConnected(connectivityManager)
+                        NetworkStateObserver.notifyConnectionChange(isConnected)
+                    }
+                },
+                IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION)
+            )
+        }
+        return context
+    }
+
+    override fun dependencies(): MutableList<Class<out Initializer<*>>> {
+        return mutableListOf(RealmInitializer::class.java)
+    }
+}

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/App.kt

@@ -23,6 +23,7 @@ import io.realm.kotlin.mongodb.exceptions.AuthException
 import io.realm.kotlin.mongodb.exceptions.InvalidCredentialsException
 import io.realm.kotlin.mongodb.internal.AppConfigurationImpl
 import io.realm.kotlin.mongodb.internal.AppImpl
+import io.realm.kotlin.mongodb.sync.Sync
 import kotlinx.coroutines.flow.Flow
 
 /**
@@ -77,6 +78,12 @@ public interface App {
      */
     public val currentUser: User?
 
+    /**
+     * Returns a Device Sync manager that control functionality across all open realms associated
+     * with this app.
+     */
+    public val sync: Sync
+
     /**
      * Returns all known users that are either [User.State.LOGGED_IN] or [User.State.LOGGED_OUT].
      * Only users that at some point logged into this device will be returned.

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/internal/AppImpl.kt

@@ -23,12 +23,15 @@ import io.realm.kotlin.internal.interop.sync.NetworkTransport
 import io.realm.kotlin.internal.util.DispatcherHolder
 import io.realm.kotlin.internal.util.Validation
 import io.realm.kotlin.internal.util.use
+import io.realm.kotlin.log.RealmLog
 import io.realm.kotlin.mongodb.App
 import io.realm.kotlin.mongodb.AppConfiguration
 import io.realm.kotlin.mongodb.AuthenticationChange
 import io.realm.kotlin.mongodb.Credentials
 import io.realm.kotlin.mongodb.User
 import io.realm.kotlin.mongodb.auth.EmailPasswordAuth
+import io.realm.kotlin.mongodb.sync.Sync
+import io.realm.kotlin.types.RealmInstant
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
@@ -45,6 +48,39 @@ public class AppImpl(
     internal val appNetworkDispatcher: DispatcherHolder
     private val networkTransport: NetworkTransport
 
+    private var lastOnlineStateReportedMs: Long? = null
+    private var lastConnectedState: Boolean? = null // null = unknown, true = connected, false = disconnected
+    @Suppress("MagicNumber")
+    private val reconnectThresholdMs = 5_000 // 5 seconds
+
+    @Suppress("invisible_member", "invisible_reference", "MagicNumber")
+    private val connectionListener = NetworkStateObserver.ConnectionListener { connectionAvailable ->
+        // In an ideal world, we would be able to reliably detect the network coming and
+        // going. Unfortunately that does not seem to be case (at least on Android).
+        //
+        // So instead of assuming that we have always detect the device going offline first,
+        // we just tell Realm Core to reconnect when we detect the network has come back.
+        //
+        // Due to the way network interfaces are re-enabled on Android, we might see multiple
+        // "isOnline" messages in short order. So in order to prevent resetting the network
+        // too often we throttle messages, so a reconnect can only happen ever 5 seconds.
+        RealmLog.debug("Network state change detected. ConnectionAvailable = $connectionAvailable")
+        val nowMs: Long = RealmInstant.now().let { timestamp ->
+            timestamp.epochSeconds * 1000L + timestamp.nanosecondsOfSecond / 1_000_000L
+        }
+        if (connectionAvailable && (lastOnlineStateReportedMs == null || nowMs - lastOnlineStateReportedMs!! > reconnectThresholdMs)
+        ) {
+            RealmLog.info("Trigger network reconnect.")
+            try {
+                sync.reconnect()
+            } catch (ex: Exception) {
+                RealmLog.error(ex.toString())
+            }
+            lastOnlineStateReportedMs = nowMs
+        }
+        lastConnectedState = connectionAvailable
+    }
+
     // Allow some delay between events being reported and them being consumed.
     // When the (somewhat arbitrary) limit is hit, we will throw an exception, since we assume the
     // consumer is doing something wrong. This is also needed because we don't
@@ -61,13 +97,15 @@ public class AppImpl(
         appNetworkDispatcher = appResources.first
         networkTransport = appResources.second
         nativePointer = appResources.third
+        NetworkStateObserver.addListener(connectionListener)
     }
 
     override val emailPasswordAuth: EmailPasswordAuth by lazy { EmailPasswordAuthImpl(nativePointer) }
 
     override val currentUser: User?
         get() = RealmInterop.realm_app_get_current_user(nativePointer)
             ?.let { UserImpl(it, this) }
+    override val sync: Sync by lazy { SyncImpl(nativePointer) }
 
     override fun allUsers(): Map<String, User> {
         val nativeUsers: List<RealmUserPointer> =
@@ -130,6 +168,7 @@ public class AppImpl(
         // be beneficial in order to reason about the lifecycle of the Sync thread and dispatchers.
         networkTransport.close()
         nativePointer.release()
+        NetworkStateObserver.removeListener(connectionListener)
     }
 
     internal companion object {

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/internal/NetworkStateObserver.kt

@@ -0,0 +1,64 @@
+package io.realm.kotlin.mongodb.internal
+
+import io.realm.kotlin.internal.interop.SynchronizableObject
+
+// Register a system specific network listener (if supported)
+internal expect fun registerSystemNetworkObserver()
+
+/**
+ * This class is responsible for keeping track of system events related to the network so it can
+ * delegate them to interested parties.
+ */
+internal object NetworkStateObserver {
+
+    /**
+     * This interface is used in a thread-safe manner, i.e. implementers do not have to think
+     * about race conditions.
+     */
+    internal fun interface ConnectionListener {
+        fun onChange(connectionAvailable: Boolean)
+    }
+
+    private val mutex = SynchronizableObject()
+    private val listeners = mutableListOf<ConnectionListener>()
+
+    init {
+        registerSystemNetworkObserver()
+    }
+
+    /**
+     * Called by each custom network implementation whenever a network change is detected.
+     */
+    fun notifyConnectionChange(isOnline: Boolean) {
+        mutex.withLock {
+            listeners.forEach {
+                it.onChange(isOnline)
+            }
+        }
+    }
+
+    /**
+     * Add a listener to be notified about any network changes.
+     * This method is thread safe.
+     * IMPORTANT: Not removing it again will result in leaks.
+     * @param listener the listener to add.
+     */
+    fun addListener(listener: ConnectionListener) {
+        mutex.withLock {
+            listeners.add(listener)
+        }
+    }
+
+    /**
+     * Removes a network listener.
+     * This method is thread safe.
+     *
+     * @param listener the listener to remove.
+     * @return `true` if the listener was removed.
+     */
+    fun removeListener(listener: ConnectionListener): Boolean {
+        mutex.withLock {
+            return listeners.remove(listener)
+        }
+    }
+}