From b125f75a38cf380a95c91d4caee2516f18d86dfb Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Tue, 31 Oct 2023 11:21:24 +0200
Subject: [PATCH 1/2] fix changes

---
 .changeset/fair-planes-walk.md                   |  6 ++++++
 .../AbstractStreamingSyncImplementation.ts       | 16 +++++++++++++++-
 .../src/sync/stream/ReactNativeRemote.ts         |  4 +++-
 3 files changed, 24 insertions(+), 2 deletions(-)
 create mode 100644 .changeset/fair-planes-walk.md

diff --git a/.changeset/fair-planes-walk.md b/.changeset/fair-planes-walk.md
new file mode 100644
index 00000000..bcdd6acb
--- /dev/null
+++ b/.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.
diff --git a/packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts
index f2e8df6c..520c0945 100644
--- a/packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts
+++ b/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,11 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
   }
 
   private updateSyncStatus(connected: boolean, lastSyncedAt?: Date) {
+    const previousValues = [this._isConnected, this._lastSyncedAt?.valueOf()];
     this._lastSyncedAt = lastSyncedAt ?? this.lastSyncedAt;
-    this.iterateListeners((cb) => cb.statusChanged?.(new SyncStatus(connected, this.lastSyncedAt)));
+    this._isConnected = connected;
+    if (!_.isEqual(previousValues, [this.isConnected, this._lastSyncedAt?.valueOf()])) {
+      this.iterateListeners((cb) => cb.statusChanged?.(new SyncStatus(this.isConnected, this.lastSyncedAt)));
+    }
   }
 }
diff --git a/packages/powersync-sdk-react-native/src/sync/stream/ReactNativeRemote.ts b/packages/powersync-sdk-react-native/src/sync/stream/ReactNativeRemote.ts
index b7b73a3f..b57c110e 100644
--- a/packages/powersync-sdk-react-native/src/sync/stream/ReactNativeRemote.ts
+++ b/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;

From cd3aa1bc6f33c7355c031a5a171110812a21d61c Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Tue, 31 Oct 2023 11:30:00 +0200
Subject: [PATCH 2/2] snapshot

---
 .../sync/stream/AbstractStreamingSyncImplementation.ts      | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts
index 520c0945..cba9eff4 100644
--- a/packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts
+++ b/packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts
@@ -299,10 +299,12 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
   }
 
   private updateSyncStatus(connected: boolean, lastSyncedAt?: Date) {
-    const previousValues = [this._isConnected, this._lastSyncedAt?.valueOf()];
+    const takeSnapShot = () => [this._isConnected, this._lastSyncedAt?.valueOf()];
+
+    const previousValues = takeSnapShot();
     this._lastSyncedAt = lastSyncedAt ?? this.lastSyncedAt;
     this._isConnected = connected;
-    if (!_.isEqual(previousValues, [this.isConnected, this._lastSyncedAt?.valueOf()])) {
+    if (!_.isEqual(previousValues, takeSnapShot())) {
       this.iterateListeners((cb) => cb.statusChanged?.(new SyncStatus(this.isConnected, this.lastSyncedAt)));
     }
   }