Fix batches not emitting change events.

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 3.0.4
+
+- Fix batch commits not triggering change events (#15)
+- Update dependencies
+
 ## 3.0.3
 
 - Update dependencies

lib/sqlite_crdt.dart

@@ -97,6 +97,6 @@ class SqliteCrdt extends SqlCrdt {
          WHERE pk > 0
        ''', [table])).map((e) => e['name'] as String);
 
-  BatchExecutor batch() =>
-      BatchExecutor(_db.batch(), canonicalTime.increment());
+  BatchExecutor batch() => BatchExecutor(_db.batch(), canonicalTime.increment(),
+      (affectedTables, hlc) => onDatasetChanged(affectedTables, hlc));
 }

lib/src/batch_executor.dart

@@ -9,9 +9,16 @@ import 'package:sqlite_crdt/src/sqlite_api.dart';
 /// long-lived batches is discouraged.
 class BatchExecutor extends CrdtWriteExecutor {
   final Batch _batch;
+  final Future<void> Function(Iterable<String> affectedTables, Hlc hlc)
+      onCommit;
 
-  BatchExecutor(this._batch, Hlc hlc) : super(BatchApi(_batch), hlc);
+  BatchExecutor(this._batch, Hlc hlc, this.onCommit)
+      : super(BatchApi(_batch), hlc);
 
   /// Commit this batch atomically. See Sqlite documentation for details.
-  Future<List<Object?>> commit() => _batch.commit();
+  Future<List<Object?>> commit() async {
+    final result = await _batch.commit();
+    await onCommit(affectedTables, hlc);
+    return result;
+  }
 }

pubspec.yaml

@@ -1,6 +1,6 @@
 name: sqlite_crdt
 description: Dart implementation of Conflict-free Replicated Data Types (CRDTs) using Sqlite
-version: 3.0.3
+version: 3.0.4
 homepage: https://github.com/cachapa/sqlite_crdt
 repository: https://github.com/cachapa/sqlite_crdt
 issue_tracker: https://github.com/cachapa/sqlite_crdt/issues
@@ -9,11 +9,11 @@ environment:
   sdk: ^3.0.0
 
 dependencies:
-  sqflite_common: ^2.5.5
-  sqflite_common_ffi: ^2.3.5
-  sqflite_common_ffi_web: ^1.0.0
   sql_crdt: ^3.0.3
 #    path: ../sql_credit
+  sqflite_common: ^2.5.6
+  sqflite_common_ffi: ^2.3.6
+  sqflite_common_ffi_web: ^1.0.1+2
 
 dev_dependencies:
   lints: any

test/sqlite_crdt_test.dart

@@ -19,10 +19,18 @@ Future<void> main() async {
           PRIMARY KEY (id)
         )
       ''');
+      await crdt.execute('''
+        CREATE TABLE users_2 (
+          id INTEGER NOT NULL,
+          name TEXT,
+          PRIMARY KEY (id)
+        )
+      ''');
     });
 
     tearDown(() async {
       await crdt.execute('DROP TABLE users');
+      await crdt.execute('DROP TABLE IF EXISTS users_2');
     });
 
     test('Replace', () async {
@@ -55,11 +63,33 @@ Future<void> main() async {
 
       final result = await crdt.query('SELECT * FROM users');
       expect(result.length, 2);
-      expect(result.first['name'], 'John Doe 1');
-      expect(result.last['name'], 'John Doe 2');
-      expect(result.first['hlc'], result.last['hlc']);
+      expect(result[0]['name'], 'John Doe 1');
+      expect(result[1]['name'], 'John Doe 2');
+      expect(result[0]['hlc'], result[1]['hlc']);
     });
 
+    /// This test is currently failing due to the way batches are processed:
+    /// The HLC for each batch operation is calculated at the moment of
+    /// creation rather than on commit.
+    ///
+    /// Changing this will require some work which may be unnecessary in light
+    /// of upcoming architectural changes in the project.
+    ///
+    // test('Mixed batch and standard inserts', () async {
+    //   final batch = crdt.batch();
+    //   insertUser(batch, 1, 'John Doe 1');
+    //   insertUser(crdt, 2, 'John Doe 2');
+    //   insertUser(batch, 3, 'John Doe 3');
+    //   await batch.commit();
+    //
+    //   final result = await crdt.query('SELECT * FROM users');
+    //   expect(result.length, 2);
+    //   expect(result[0]['name'], 'John Doe 1');
+    //   expect(result[1]['name'], 'John Doe 2');
+    //   expect(result[2]['name'], 'John Doe 3');
+    //   expect(result.first['hlc'], result.last['hlc']);
+    // });
+
     test('Uncommitted batch', () async {
       final batch = crdt.batch();
       insertUser(batch, 1, 'John Doe 1');
@@ -68,5 +98,39 @@ Future<void> main() async {
       final result = await crdt.query('SELECT * FROM users');
       expect(result.length, 0);
     });
+
+    test('Emit on single insert', () async {
+      final streamTest = expectLater(
+        crdt.watch('SELECT * FROM users'),
+        emitsInOrder([
+          [],
+          (List<Map<String, Object?>> e) =>
+              e.length == 1 && e[0]['name'] == 'John Doe'
+        ]),
+      );
+      final batch = crdt.batch();
+      insertUser(batch, 1, 'John Doe');
+      await batch.commit();
+      await streamTest;
+    });
+
+    test('Emit on multiple tables', () async {
+      final streamTest = expectLater(
+        crdt.watch('SELECT * FROM users UNION ALL SELECT * FROM users_2'),
+        emitsInOrder([
+          [],
+          (List<Map<String, Object?>> e) =>
+              e.length == 2 &&
+              e[0]['name'] == 'John Doe' &&
+              e[1]['name'] == 'Jane Doe',
+        ]),
+      );
+      final batch = crdt.batch();
+      insertUser(batch, 1, 'John Doe');
+      await batch.execute(
+          'INSERT INTO users_2 (id, name) VALUES (?1, ?2)', [2, 'Jane Doe']);
+      await batch.commit();
+      await streamTest;
+    });
   });
 }