Merge pull request #19 from journeyapps/better-connectivity-updates

Update connectivity status

Author: stevensJourney

.changeset/fair-planes-walk.md

@@ -0,0 +1,6 @@
+---
+'@journeyapps/powersync-sdk-react-native': patch
+'@journeyapps/powersync-sdk-common': patch
+---
+
+Use default timeout in post streaming warning message. Update connectivity status on streaming messages.

packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -54,10 +54,13 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
 
   private isUploadingCrud: boolean;
 
+  protected _isConnected: boolean;
+
   constructor(options: AbstractStreamingSyncImplementationOptions) {
     super();
     this.options = { ...DEFAULT_STREAMING_SYNC_OPTIONS, ...options };
     this.isUploadingCrud = false;
+    this._isConnected = false;
   }
 
   get lastSyncedAt() {
@@ -68,6 +71,10 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
     return this.options.logger!;
   }
 
+  get isConnected() {
+    return this._isConnected;
+  }
+
   abstract obtainLock<T>(lockOptions: LockOptions<T>): Promise<T>;
 
   async hasCompletedSync() {
@@ -163,6 +170,9 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
           },
           signal
         )) {
+          // A connection is active and messages are being received
+          this.updateSyncStatus(true);
+
           if (isStreamingSyncCheckpoint(line)) {
             targetCheckpoint = line.checkpoint;
             const bucketsToDelete = new Set<string>(bucketSet);
@@ -289,7 +299,13 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
   }
 
   private updateSyncStatus(connected: boolean, lastSyncedAt?: Date) {
+    const takeSnapShot = () => [this._isConnected, this._lastSyncedAt?.valueOf()];
+
+    const previousValues = takeSnapShot();
     this._lastSyncedAt = lastSyncedAt ?? this.lastSyncedAt;
-    this.iterateListeners((cb) => cb.statusChanged?.(new SyncStatus(connected, this.lastSyncedAt)));
+    this._isConnected = connected;
+    if (!_.isEqual(previousValues, takeSnapShot())) {
+      this.iterateListeners((cb) => cb.statusChanged?.(new SyncStatus(this.isConnected, this.lastSyncedAt)));
+    }
   }
 }

packages/powersync-sdk-react-native/src/sync/stream/ReactNativeRemote.ts

@@ -65,7 +65,9 @@ export class ReactNativeRemote extends AbstractRemote {
       Platform.OS == 'android'
         ? setTimeout(() => {
             this.logger.warn(
-              `HTTP Streaming POST is taking longer than 30 seconds to resolve. If using a debug build, please ensure Flipper Network plugin is disabled.`
+              `HTTP Streaming POST is taking longer than ${Math.ceil(
+                STREAMING_POST_TIMEOUT_MS / 1000
+              )} seconds to resolve. If using a debug build, please ensure Flipper Network plugin is disabled.`
             );
           }, STREAMING_POST_TIMEOUT_MS)
         : null;