Remove Rust-specific parts

Author: simolus3

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

@@ -1,7 +1,6 @@
 import 'dart:async';
 
 import 'package:collection/collection.dart';
-import 'package:powersync_core/src/sync/instruction.dart';
 
 import 'sync_status.dart';
 import 'bucket_storage.dart';
@@ -80,20 +79,6 @@ final class MutableSyncStatus {
     }
   }
 
-  void applyFromCore(CoreSyncStatus status) {
-    connected = status.connected;
-    connecting = status.connecting;
-    downloading = status.downloading != null;
-    priorityStatusEntries = status.priorityStatus;
-    downloadProgress = switch (status.downloading) {
-      null => null,
-      final downloading => InternalSyncDownloadProgress(downloading.progress),
-    };
-    lastSyncedAt = status.priorityStatus
-        .firstWhereOrNull((s) => s.priority == BucketPriority.fullSyncPriority)
-        ?.lastSyncedAt;
-  }
-
   SyncStatus immutableSnapshot() {
     return SyncStatus(
       connected: connected,

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

@@ -24,19 +24,10 @@ final class SyncOptions {
   /// When set to null, PowerSync defaults to a delay of 5 seconds.
   final Duration? retryDelay;
 
-  /// The client implementation to use.
-  ///
-  /// This allows switching between the existing [SyncClientImplementation.dart]
-  /// implementation and a newer one ([SyncClientImplementation.rust]).
-  ///
-  /// Note that not setting this field to the default value is experimental.
-  final SyncClientImplementation syncImplementation;
-
   const SyncOptions({
     this.crudThrottleTime,
     this.retryDelay,
     this.params,
-    this.syncImplementation = SyncClientImplementation.dart,
   });
 }
 
@@ -54,37 +45,13 @@ extension type ResolvedSyncOptions(SyncOptions source) {
       crudThrottleTime: other.crudThrottleTime ?? crudThrottleTime,
       retryDelay: other.retryDelay ?? retryDelay,
       params: other.params ?? params,
-      syncImplementation: other.syncImplementation,
     );
 
     final didChange = !_mapEquality.equals(other.params, params) ||
         other.crudThrottleTime != crudThrottleTime ||
-        other.retryDelay != retryDelay ||
-        other.syncImplementation != source.syncImplementation;
+        other.retryDelay != retryDelay;
     return (ResolvedSyncOptions(newOptions), didChange);
   }
 
   static const _mapEquality = MapEquality<String, dynamic>();
 }
-
-/// Supported sync client implementations.
-///
-/// Not using the default implementation (currently [dart], but this may change
-/// in the future) is experimental.
-@experimental
-enum SyncClientImplementation {
-  /// Decode and handle data received from the sync service in Dart.
-  ///
-  /// This is the default option.
-  dart,
-
-  /// An _experimental_ implementation of the sync client that is written in
-  /// Rust and shared across the PowerSync SDKs.
-  ///
-  /// Since this client decodes sync lines in Rust instead of parsing them in
-  /// Dart, it can be more performant than the the default [dart]
-  /// implementation. Since this option has not seen as much real-world testing,
-  /// it is marked as __experimental__ at the moment!
-  @experimental
-  rust,
-}

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

@@ -4,6 +4,7 @@ import 'dart:convert' as convert;
 
 /// Inject a broadcast stream into a normal stream.
 Stream<T> addBroadcast<T>(Stream<T> a, Stream<T> broadcast) {
+  assert(broadcast.isBroadcast);
   return mergeStreams([a, broadcast]);
 }
 
@@ -63,17 +64,27 @@ Stream<T> mergeStreams<T>(List<Stream<T>> streams) {
   return controller.stream;
 }
 
-/// Given a stream of lines, parse each line as JSON.
-Stream<Object?> mapJson(Stream<String> lineInput) {
-  final jsonInput = lineInput.transform(
-      StreamTransformer.fromHandlers(handleError: (error, stackTrace, sink) {
-    /// On Web if the connection is closed, this error will throw, but
-    /// the stream is never closed. This closes the stream on error.
-    sink.close();
-  }, handleData: (String data, EventSink<dynamic> sink) {
-    sink.add(convert.jsonDecode(data));
-  }));
-  return jsonInput;
+extension ByteStreamToLines on Stream<List<int>> {
+  /// Decodes this stream using UTF8 and then splits the text stream by
+  /// newlines.
+  Stream<String> get lines {
+    final textInput = transform(convert.utf8.decoder);
+    return textInput.transform(const convert.LineSplitter());
+  }
+}
+
+extension StreamToJson on Stream<String> {
+  Stream<Object?> get parseJson {
+    final jsonInput = transform(
+        StreamTransformer.fromHandlers(handleError: (error, stackTrace, sink) {
+      /// On Web if the connection is closed, this error will throw, but
+      /// the stream is never closed. This closes the stream on error.
+      sink.close();
+    }, handleData: (String data, EventSink<dynamic> sink) {
+      sink.add(convert.jsonDecode(data));
+    }));
+    return jsonInput;
+  }
 }
 
 void pauseAll(List<StreamSubscription<void>> subscriptions) {