Only set connected on first line

Author: simolus3

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -48,8 +48,7 @@ class StreamingSyncImplementation implements StreamingSync {
 
   final http.Client _client;
 
-  @visibleForTesting
-  final SyncStatusStateStream state = SyncStatusStateStream();
+  final SyncStatusStateStream _state = SyncStatusStateStream();
 
   AbortController? _abort;
 
@@ -81,7 +80,7 @@ class StreamingSyncImplementation implements StreamingSync {
   Duration get _retryDelay => options.retryDelay;
 
   @override
-  Stream<SyncStatus> get statusStream => state.statusStream;
+  Stream<SyncStatus> get statusStream => _state.statusStream;
 
   @override
   Future<void> abort() async {
@@ -105,7 +104,7 @@ class StreamingSyncImplementation implements StreamingSync {
       await _nonLineSyncEvents.close();
 
       _client.close();
-      state.close();
+      _state.close();
     }
   }
 
@@ -121,7 +120,7 @@ class StreamingSyncImplementation implements StreamingSync {
       _crudLoop();
       var invalidCredentials = false;
       while (!aborted) {
-        state.updateStatus((s) => s.setConnectingIfNotConnected());
+        _state.updateStatus((s) => s.setConnectingIfNotConnected());
         try {
           if (invalidCredentials) {
             // This may error. In that case it will be retried again on the next
@@ -143,7 +142,7 @@ class StreamingSyncImplementation implements StreamingSync {
           logger.warning('Sync error: $message', e, stacktrace);
           invalidCredentials = true;
 
-          state.updateStatus((s) => s.applyDownloadError(e));
+          _state.updateStatus((s) => s.applyDownloadError(e));
 
           // On error, wait a little before retrying
           // When aborting, don't wait
@@ -189,7 +188,7 @@ class StreamingSyncImplementation implements StreamingSync {
           // This is the first item in the FIFO CRUD queue.
           CrudEntry? nextCrudItem = await adapter.nextCrudItem();
           if (nextCrudItem != null) {
-            state.updateStatus((s) => s.uploading = true);
+            _state.updateStatus((s) => s.uploading = true);
             if (nextCrudItem.clientId == checkedCrudItem?.clientId) {
               // This will force a higher log level than exceptions which are caught here.
               logger.warning(
@@ -202,7 +201,7 @@ class StreamingSyncImplementation implements StreamingSync {
 
             checkedCrudItem = nextCrudItem;
             await connector.uploadCrud();
-            state.updateStatus((s) => s.uploadError = null);
+            _state.updateStatus((s) => s.uploadError = null);
           } else {
             // Uploading is completed
             await adapter.updateLocalTarget(() => getWriteCheckpoint());
@@ -211,10 +210,10 @@ class StreamingSyncImplementation implements StreamingSync {
         } catch (e, stacktrace) {
           checkedCrudItem = null;
           logger.warning('Data upload error', e, stacktrace);
-          state.updateStatus((s) => s.applyUploadError(e));
+          _state.updateStatus((s) => s.applyUploadError(e));
           await _delayRetry();
 
-          if (!state.status.connected) {
+          if (!_state.status.connected) {
             // Exit the upload loop if the sync stream is no longer connected
             break;
           }
@@ -223,7 +222,7 @@ class StreamingSyncImplementation implements StreamingSync {
               e,
               stacktrace);
         } finally {
-          state.updateStatus((s) => s.uploading = false);
+          _state.updateStatus((s) => s.uploading = false);
         }
       }
     }, timeout: _retryDelay).whenComplete(() {
@@ -264,7 +263,7 @@ class StreamingSyncImplementation implements StreamingSync {
   }
 
   void _updateStatusForPriority(SyncPriorityStatus completed) {
-    state.updateStatus((s) {
+    _state.updateStatus((s) {
       // All status entries with a higher priority can be deleted since this
       // partial sync includes them.
       s.priorityStatusEntries = [
@@ -325,7 +324,7 @@ class StreamingSyncImplementation implements StreamingSync {
           bucketMap = newBuckets;
           await adapter.removeBuckets([...bucketsToDelete]);
           final initialProgress = await adapter.getBucketOperationProgress();
-          state.updateStatus(
+          _state.updateStatus(
               (s) => s.applyCheckpointStarted(initialProgress, line));
         case StreamingSyncCheckpointComplete():
           final result = await _applyCheckpoint(targetCheckpoint!, _abort);
@@ -376,7 +375,7 @@ class StreamingSyncImplementation implements StreamingSync {
               writeCheckpoint: diff.writeCheckpoint);
           targetCheckpoint = newCheckpoint;
           final initialProgress = await adapter.getBucketOperationProgress();
-          state.updateStatus(
+          _state.updateStatus(
               (s) => s.applyCheckpointStarted(initialProgress, newCheckpoint));
 
           bucketMap = newBuckets.map((name, checksum) =>
@@ -386,7 +385,7 @@ class StreamingSyncImplementation implements StreamingSync {
         case SyncDataBatch():
           // TODO: This increments the counters before actually saving sync
           // data. Might be fine though?
-          state.updateStatus((s) => s.applyBatchReceived(line));
+          _state.updateStatus((s) => s.applyBatchReceived(line));
           await adapter.saveSyncData(line);
         case StreamingSyncKeepalive(:final tokenExpiresIn):
           if (tokenExpiresIn == 0) {
@@ -421,6 +420,7 @@ class StreamingSyncImplementation implements StreamingSync {
 
       switch (line) {
         case ReceivedLine(:final line):
+          _state.updateStatus((s) => s.setConnected());
           await handleLine(line as StreamingSyncLine);
         case UploadCompleted():
           // Only relevant for the Rust sync implementation.
@@ -433,7 +433,6 @@ class StreamingSyncImplementation implements StreamingSync {
           throw AssertionError('unreachable');
       }
 
-      state.updateStatus((s) => s.setConnected());
       if (haveInvalidated) {
         // Stop this connection, so that a new one will be started
         break;
@@ -473,7 +472,7 @@ class StreamingSyncImplementation implements StreamingSync {
     if (result.checkpointValid && result.ready) {
       logger.fine('validated checkpoint: $targetCheckpoint');
 
-      state.updateStatus((s) => s.applyCheckpointReached(targetCheckpoint));
+      _state.updateStatus((s) => s.applyCheckpointReached(targetCheckpoint));
 
       return const (abort: false, didApply: true);
     } else {