diff --git a/packages/powersync/lib/src/stream_utils.dart b/packages/powersync/lib/src/stream_utils.dart
index 5cf7297d..37aee207 100644
--- a/packages/powersync/lib/src/stream_utils.dart
+++ b/packages/powersync/lib/src/stream_utils.dart
@@ -68,6 +68,13 @@ Stream<Object?> ndjson(ByteStream input) {
   return jsonInput;
 }
 
+/// Given a raw ByteStream, parse each line as JSON.
+Stream<String> newlines(ByteStream input) {
+  final textInput = input.transform(convert.utf8.decoder);
+  final lineInput = textInput.transform(const convert.LineSplitter());
+  return lineInput;
+}
+
 void pauseAll(List<StreamSubscription> subscriptions) {
   for (var sub in subscriptions) {
     sub.pause();