|
1 | 1 | import 'dart:async'; |
2 | 2 | import 'dart:ffi'; |
| 3 | +import 'dart:isolate'; |
3 | 4 |
|
4 | 5 | import 'bindings/bindings.dart'; |
| 6 | +import 'bindings/helpers.dart'; |
| 7 | +import 'modelinfo/entity_definition.dart'; |
5 | 8 | import 'query/query.dart'; |
6 | 9 | import 'store.dart'; |
7 | | -import 'util.dart'; |
8 | | - |
9 | | -// ignore_for_file: non_constant_identifier_names |
10 | | - |
11 | | -// dart callback signature |
12 | | -typedef Any = void Function(Pointer<Void>, Pointer<Uint32>, int); |
13 | | - |
14 | | -class _Observable { |
15 | | - static final _anyObserver = <int, Pointer<OBX_observer>>{}; |
16 | | - static final _any = <int, Map<int, Any>>{}; |
17 | | - |
18 | | - // sync:true -> ObjectBoxException: 10001 TX is not active anymore: #101 |
19 | | - static final controller = StreamController<int>.broadcast(); |
20 | | - |
21 | | - // The user_data is used to pass the store ptr address |
22 | | - // in case there is no consensus on the entity id between stores |
23 | | - static void _anyCallback( |
24 | | - Pointer<Void> user_data, Pointer<Uint32> mutated_ids, int mutated_count) { |
25 | | - final storeAddress = user_data.address; |
26 | | - // call schema's callback |
27 | | - final storeCallbacks = _any[storeAddress]; |
28 | | - if (storeCallbacks != null) { |
29 | | - for (var i = 0; i < mutated_count; i++) { |
30 | | - storeCallbacks[mutated_ids[i]] |
31 | | - ?.call(user_data, mutated_ids, mutated_count); |
32 | | - } |
33 | | - } |
| 10 | + |
| 11 | +/// Simple wrapper used below in ObservableStore to reduce code duplication. |
| 12 | +/// Contains shared code for single-entity observer and the generic/global one. |
| 13 | +class _Observer<StreamValueType> { |
| 14 | + StreamController<StreamValueType> /*?*/ controller; |
| 15 | + Pointer<OBX_observer> /*?*/ _cObserver; |
| 16 | + ReceivePort /*?*/ receivePort; |
| 17 | + |
| 18 | + int get nativePort => receivePort /*!*/ .sendPort.nativePort; |
| 19 | + |
| 20 | + set cObserver(Pointer<OBX_observer> value) { |
| 21 | + _cObserver = checkObxPtr(value, 'observer initialization failed'); |
| 22 | + _debugLog('started'); |
34 | 23 | } |
35 | 24 |
|
36 | | - static void subscribe(Store store) { |
37 | | - syncOrObserversExclusive.mark(store); |
| 25 | + Stream<StreamValueType> get stream => controller /*!*/ .stream; |
38 | 26 |
|
39 | | - final callback = Pointer.fromFunction<obx_observer>(_anyCallback); |
40 | | - final storePtr = store.ptr; |
41 | | - _anyObserver[storePtr.address] = |
42 | | - bindings.obx_observe(storePtr, callback, storePtr.cast<Void>()); |
43 | | - StoreCloseObserver.addListener(store, _anyObserver[storePtr.address], () { |
44 | | - unsubscribe(store); |
45 | | - }); |
| 27 | + _Observer() { |
| 28 | + initializeDartAPI(); |
| 29 | + } |
| 30 | + |
| 31 | + // start() is called whenever user starts listen()-ing to the stream |
| 32 | + void init(void Function() start) { |
| 33 | + controller = StreamController<StreamValueType>( |
| 34 | + onListen: start, onPause: stop, onResume: start, onCancel: stop); |
46 | 35 | } |
47 | 36 |
|
48 | | - // #53 ffi:Pointer finalizer |
49 | | - static void unsubscribe(Store store) { |
50 | | - final storeAddress = store.ptr.address; |
51 | | - if (!_anyObserver.containsKey(storeAddress)) { |
52 | | - return; |
53 | | - } |
54 | | - StoreCloseObserver.removeListener(store, _anyObserver[storeAddress]); |
55 | | - bindings.obx_observer_close(_anyObserver[storeAddress]); |
56 | | - _anyObserver.remove(storeAddress); |
57 | | - syncOrObserversExclusive.unmark(store); |
| 37 | + // stop() is called when the stream subscription is paused or canceled |
| 38 | + void stop() { |
| 39 | + _debugLog('stopped'); |
| 40 | + if (_cObserver != null) checkObx(bindings.obx_observer_close(_cObserver)); |
58 | 41 | } |
59 | 42 |
|
60 | | - static bool isSubscribed(Store store) => |
61 | | - _Observable._anyObserver.containsKey(store.ptr.address); |
| 43 | + void _debugLog(String message) { |
| 44 | + // print('Observer=${_cObserver?.address} ' + message); |
| 45 | + } |
62 | 46 | } |
63 | 47 |
|
64 | | -extension Streamable<T> on Query<T> { |
65 | | - void _setup() { |
66 | | - if (!_Observable.isSubscribed(store)) { |
67 | | - _Observable.subscribe(store); |
68 | | - } |
69 | | - final storeAddress = store.ptr.address; |
70 | | - |
71 | | - _Observable._any[storeAddress] ??= <int, Any>{}; |
72 | | - _Observable._any[storeAddress] /*!*/ [entityId] ??= (u, _, __) { |
73 | | - // dummy value to trigger an event |
74 | | - _Observable.controller.add(entityId); |
75 | | - }; |
| 48 | +/// StreamController implementation inspired by the sample controller sample at: |
| 49 | +/// https://dart.dev/articles/libraries/creating-streams#honoring-the-pause-state |
| 50 | +/// https://dart.dev/articles/libraries/code/stream_controller.dart |
| 51 | +extension ObservableStore on Store { |
| 52 | + /// Create a stream to data changes on EntityT (stored Entity class). |
| 53 | + /// |
| 54 | + /// The stream receives an event whenever an object of EntityT is created or |
| 55 | + /// changed or deleted. Make sure to close() the subscription after you're |
| 56 | + /// done with it to avoid hanging change listeners. |
| 57 | + Stream<void> subscribe<EntityT>() { |
| 58 | + final observer = _Observer<void>(); |
| 59 | + final entityId = entityDef<EntityT>().model.id.id; |
| 60 | + |
| 61 | + observer.init(() { |
| 62 | + // We're listening to events on single entity so there's no argument. |
| 63 | + // Ideally, controller.add() would work but it doesn't, even though we're |
| 64 | + // using StreamController<Void> so the argument type is `void`. |
| 65 | + observer.receivePort = ReceivePort() |
| 66 | + ..listen((_) => observer.controller.add(null)); |
| 67 | + observer.cObserver = bindings.obx_dart_observe_single_type( |
| 68 | + ptr, entityId, observer.nativePort); |
| 69 | + }); |
| 70 | + |
| 71 | + return observer.stream; |
76 | 72 | } |
77 | 73 |
|
| 74 | + /// Create a stream to data changes on all Entity types. |
| 75 | + /// |
| 76 | + /// The stream receives an even whenever any data changes in the database. |
| 77 | + /// Make sure to close() the subscription after you're done with it to avoid |
| 78 | + /// hanging change listeners. |
| 79 | + Stream<Type> subscribeAll() { |
| 80 | + initializeDartAPI(); |
| 81 | + final observer = _Observer<Type>(); |
| 82 | + |
| 83 | + // create a map from Entity ID to Entity type (dart class) |
| 84 | + final entityTypesById = <int, Type>{}; |
| 85 | + defs.bindings.forEach((Type entity, EntityDefinition entityDef) => |
| 86 | + entityTypesById[entityDef.model.id.id] = entity); |
| 87 | + |
| 88 | + observer.init(() { |
| 89 | + // We're listening to a events for all entity types. C-API sends entity ID |
| 90 | + // and we must map it to a dart type (class) corresponding to that entity. |
| 91 | + observer.receivePort = ReceivePort() |
| 92 | + ..listen((entityIds) { |
| 93 | + if (entityIds is! List) { |
| 94 | + observer.controller.addError(Exception( |
| 95 | + 'Received invalid data format from the core notification: (${entityIds.runtimeType}) $entityIds')); |
| 96 | + return; |
| 97 | + } |
| 98 | + |
| 99 | + entityIds.forEach((entityId) { |
| 100 | + if (entityId is! int) { |
| 101 | + observer.controller.addError(Exception( |
| 102 | + 'Received invalid item data format from the core notification: (${entityId.runtimeType}) $entityId')); |
| 103 | + return; |
| 104 | + } |
| 105 | + |
| 106 | + final entityType = entityTypesById[entityId]; |
| 107 | + if (entityType == null) { |
| 108 | + observer.controller.addError(Exception( |
| 109 | + 'Received data change notification for an unknown entity ID $entityId')); |
| 110 | + } else { |
| 111 | + observer.controller.add(entityType); |
| 112 | + } |
| 113 | + }); |
| 114 | + }); |
| 115 | + observer.cObserver = bindings.obx_dart_observe(ptr, observer.nativePort); |
| 116 | + }); |
| 117 | + |
| 118 | + return observer.stream; |
| 119 | + } |
| 120 | +} |
| 121 | + |
| 122 | +extension Streamable<T> on Query<T> { |
78 | 123 | Stream<List<T>> findStream({int offset = 0, int limit = 0}) { |
79 | | - _setup(); |
80 | | - return _Observable.controller.stream |
81 | | - .where((e) => e == entityId) |
82 | | - .map((_) => find(offset: offset, limit: limit)); |
| 124 | + return store.subscribe<T>().map((_) => find(offset: offset, limit: limit)); |
83 | 125 | } |
84 | 126 |
|
85 | 127 | /// Use this for Query Property |
86 | 128 | Stream<Query<T>> get stream { |
87 | | - _setup(); |
88 | | - return _Observable.controller.stream |
89 | | - .where((e) => e == entityId) |
90 | | - .map((_) => this); |
| 129 | + return store.subscribe<T>().map((_) => this); |
91 | 130 | } |
92 | 131 | } |
0 commit comments