feat(crud): add streaming usecase (closes #180)

packages/wyatt_crud_bloc/example/lib/advanced_cubit_view.dart

@@ -33,11 +33,11 @@ class AdvancedCubitView extends StatelessWidget {
       ),
       body: BlocProvider(
         create: (context) =>
-            UserAdvancedCubit(context.read<CrudRepository<User>>())..getAll(),
+            UserAdvancedCubit(context.read<CrudRepository<User>>())
+              ..streaming(),
         child: Builder(builder: (context) {
           return Column(
             children: [
-              const Text("Data:"),
               BlocBuilder<UserAdvancedCubit, CrudState>(
                 buildWhen: (previous, current) {
                   if (current is CrudLoading && current is! CrudReading) {
@@ -117,31 +117,6 @@ class AdvancedCubitView extends StatelessWidget {
                   },
                 ),
               ),
-              ElevatedButton(
-                onPressed: () {
-                  context.read<UserAdvancedCubit>().getAll();
-                },
-                child: BlocBuilder<UserAdvancedCubit, CrudState>(
-                  buildWhen: (previous, current) {
-                    if (current is CrudLoading && current is! CrudReading) {
-                      return false;
-                    }
-                    return true;
-                  },
-                  builder: (context, state) {
-                    return state is CrudReading
-                        ? const SizedBox(
-                            width: 20,
-                            height: 20,
-                            child: CircularProgressIndicator(
-                              color: Colors.white,
-                            ),
-                          )
-                        : const Text("GetAll");
-                  },
-                ),
-              ),
-              const SizedBox(height: 20),
             ],
           );
         }),

packages/wyatt_crud_bloc/example/lib/app.dart

@@ -16,6 +16,7 @@
 
 import 'package:crud_bloc_example/advanced_cubit_view.dart';
 import 'package:crud_bloc_example/basic_cubit_view.dart';
+import 'package:crud_bloc_example/streaming_cubit_view.dart';
 import 'package:crud_bloc_example/user_entity.dart';
 import 'package:crud_bloc_example/user_model.dart';
 import 'package:flutter/material.dart';
@@ -65,7 +66,9 @@ class MyHomePage extends StatelessWidget {
       appBar: AppBar(
         title: const Text('Flutter Demo Home Page'),
       ),
-      body: Column(
+      body: Center(
+        child: Column(
+          mainAxisAlignment: MainAxisAlignment.center,
           children: [
             ElevatedButton(
               onPressed: () {
@@ -79,6 +82,18 @@ class MyHomePage extends StatelessWidget {
               child: const Text('Basic example'),
             ),
             const SizedBox(height: 20),
+            ElevatedButton(
+              onPressed: () {
+                Navigator.push(
+                  context,
+                  MaterialPageRoute(
+                    builder: (context) => const StreamingCubitView(),
+                  ),
+                );
+              },
+              child: const Text('Streaming example'),
+            ),
+            const SizedBox(height: 20),
             ElevatedButton(
               onPressed: () {
                 Navigator.push(
@@ -92,6 +107,7 @@ class MyHomePage extends StatelessWidget {
             ),
           ],
         ),
+      ),
     );
   }
 }

packages/wyatt_crud_bloc/example/lib/streaming_cubit_view.dart

@@ -0,0 +1,125 @@
+// Copyright (C) 2023 WYATT GROUP
+// Please see the AUTHORS file for details.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+import 'dart:math';
+
+import 'package:crud_bloc_example/user_entity.dart';
+import 'package:crud_bloc_example/user_streaming_cubit.dart';
+import 'package:flutter/material.dart';
+import 'package:flutter_bloc/flutter_bloc.dart';
+import 'package:wyatt_crud_bloc/wyatt_crud_bloc.dart';
+
+class StreamingCubitView extends StatelessWidget {
+  const StreamingCubitView({super.key});
+
+  @override
+  Widget build(BuildContext context) {
+    return Scaffold(
+      appBar: AppBar(
+        title: const Text("Streaming Cubit"),
+      ),
+      body: BlocProvider(
+        create: (context) =>
+            UserStreamingCubit(context.read<CrudRepository<User>>())..read(),
+        child: Builder(builder: (context) {
+          return Column(
+            children: [
+              BlocBuilder<UserStreamingCubit, CrudState>(
+                buildWhen: (previous, current) {
+                  if (current is CrudLoading && current is! CrudReading) {
+                    return false;
+                  }
+                  return true;
+                },
+                builder: (context, state) {
+                  return Expanded(
+                    child: CrudBuilder.typed<CrudListLoaded<User?>>(
+                      state: state,
+                      builder: ((context, state) {
+                        return ListView.builder(
+                          itemCount: state.data.length,
+                          itemBuilder: (context, index) {
+                            final user = state.data.elementAt(index);
+                            return ListTile(
+                              title: Text(user?.name ?? 'Error'),
+                              subtitle: Text(user?.email ?? 'Error'),
+                              onTap: () {
+                                context
+                                    .read<UserStreamingCubit>()
+                                    .delete(id: (user?.id)!);
+                              },
+                              onLongPress: () {
+                                context.read<UserStreamingCubit>().update(
+                                      single: UpdateParameters(
+                                          id: user?.id ?? '',
+                                          raw: {
+                                            'email': '${user?.id}@updated.io',
+                                          }),
+                                    );
+                              },
+                            );
+                          },
+                        );
+                      }),
+                      initialBuilder: (context, state) =>
+                          const Center(child: CircularProgressIndicator()),
+                      loadingBuilder: (context, state) =>
+                          const Center(child: CircularProgressIndicator()),
+                      errorBuilder: (context, state) => Text("Error: $state"),
+                    ),
+                  );
+                },
+              ),
+              const SizedBox(height: 20),
+              ElevatedButton(
+                onPressed: () {
+                  final r = Random().nextInt(1000);
+                  context.read<UserStreamingCubit>().create(
+                        User(
+                          id: '$r',
+                          name: 'Wyatt $r',
+                          email: '$r@wyattapp.io',
+                          phone: '06$r',
+                        ),
+                      );
+                },
+                child: BlocBuilder<UserStreamingCubit, CrudState>(
+                  buildWhen: (previous, current) {
+                    if (current is CrudLoading && current is! CrudCreating) {
+                      return false;
+                    }
+                    return true;
+                  },
+                  builder: (context, state) {
+                    return state is CrudCreating
+                        ? const SizedBox(
+                            width: 20,
+                            height: 20,
+                            child: CircularProgressIndicator(
+                              color: Colors.white,
+                            ),
+                          )
+                        : const Text("Create");
+                  },
+                ),
+              ),
+            ],
+          );
+        }),
+      ),
+    );
+  }
+}

packages/wyatt_crud_bloc/example/lib/user_advanced_cubit.dart

@@ -46,6 +46,9 @@ class UserAdvancedCubit extends CrudAdvancedCubit<User> {
   @override
   Search<User>? get crudSearch => Search(crudRepository);
 
+  @override
+  Streaming<User>? get crudStreaming => Streaming(crudRepository);
+
   @override
   Update<User>? get crudUpdate => Update(crudRepository);
 

packages/wyatt_crud_bloc/example/lib/user_streaming_cubit.dart

@@ -0,0 +1,42 @@
+// Copyright (C) 2022 WYATT GROUP
+// Please see the AUTHORS file for details.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+import 'package:crud_bloc_example/user_entity.dart';
+import 'package:wyatt_crud_bloc/wyatt_crud_bloc.dart';
+
+/// A [CrudCubit] for [User].
+class UserStreamingCubit extends CrudCubit<User> {
+  final CrudRepository<User> crudRepository;
+
+  UserStreamingCubit(this.crudRepository);
+
+  @override
+  DefaultCreate<User>? get createOperation => Create(crudRepository);
+
+  @override
+  DefaultDelete? get deleteOperation => Delete(crudRepository);
+
+  @override
+  DefaultRead? get readOperation => Streaming(crudRepository);
+
+  @override
+  DefaultUpdate? get updateOperation => Update(crudRepository);
+
+  @override
+  ModelIdentifier<User> get modelIdentifier => ModelIdentifier(
+        getIdentifier: (user) => user.id ?? '',
+      );
+}

packages/wyatt_crud_bloc/lib/src/data/data_sources/crud_data_source_in_memory_impl.dart

@@ -16,6 +16,7 @@
 
 import 'dart:async';
 
+import 'package:rxdart/subjects.dart';
 import 'package:wyatt_crud_bloc/src/core/enums/operation_type.dart';
 import 'package:wyatt_crud_bloc/src/core/enums/where_query_type.dart';
 import 'package:wyatt_crud_bloc/src/data/data_sources/crud_data_source.dart';
@@ -58,8 +59,8 @@ class CrudDataSourceInMemoryImpl extends CrudDataSource {
   }) : _data = data ?? {};
 
   final Map<String, Map<String, dynamic>> _data;
-  final StreamController<List<Map<String, dynamic>?>> _streamData =
-      StreamController();
+  final BehaviorSubject<List<Map<String, dynamic>?>> _streamData =
+      BehaviorSubject<List<Map<String, dynamic>?>>.seeded([]);
 
   @override
   Future<void> create(Map<String, dynamic> object, {String? id}) async {
@@ -142,7 +143,7 @@ class CrudDataSourceInMemoryImpl extends CrudDataSource {
         }
 
         return res;
-      }).asBroadcastStream();
+      });
 
   @override
   Future<void> update(

packages/wyatt_crud_bloc/lib/src/domain/usecases/streaming.dart

@@ -20,14 +20,14 @@ import 'package:wyatt_crud_bloc/src/domain/repositories/crud_repository.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/usecases.dart';
 import 'package:wyatt_type_utils/wyatt_type_utils.dart';
 
-/// {@template stream}
+/// {@template streaming}
 /// A use case that streams the object models.
 /// {@endtemplate}
-class Stream<Model>
+class Streaming<Model>
     extends Usecase<StreamParameters, StreamResult<List<Model?>>>
     with ReadOperation<StreamParameters, StreamResult<List<Model?>>> {
-  /// {@macro stream}
-  const Stream(this.crudRepository);
+  /// {@macro streaming}
+  const Streaming(this.crudRepository);
 
   final CrudRepository<Model> crudRepository;
 

packages/wyatt_crud_bloc/lib/src/domain/usecases/usecases.dart

@@ -26,6 +26,7 @@ export 'get.dart';
 export 'get_all.dart';
 export 'params/params.dart';
 export 'search.dart';
+export 'streaming.dart';
 export 'update.dart';
 export 'update_all.dart';
 

packages/wyatt_crud_bloc/lib/src/features/crud/blocs/crud_advanced_cubit.dart

@@ -16,6 +16,7 @@
 
 import 'dart:async';
 
+import 'package:wyatt_architecture/wyatt_architecture.dart';
 import 'package:wyatt_crud_bloc/src/core/enums/where_query_type.dart';
 import 'package:wyatt_crud_bloc/src/core/model_identifier.dart';
 import 'package:wyatt_crud_bloc/src/domain/entities/query.dart';
@@ -24,11 +25,14 @@ import 'package:wyatt_crud_bloc/src/domain/usecases/delete.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/delete_all.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/get.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/get_all.dart';
+import 'package:wyatt_crud_bloc/src/domain/usecases/params/stream_parameters.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/params/update_parameters.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/search.dart';
+import 'package:wyatt_crud_bloc/src/domain/usecases/streaming.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/update.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/update_all.dart';
 import 'package:wyatt_crud_bloc/src/features/crud/blocs/crud_base_cubit/crud_base_cubit.dart';
+import 'package:wyatt_type_utils/wyatt_type_utils.dart';
 
 /// {@template crud_cubit_advanced}
 /// Cubit that handles CRUD operations with more granularity.
@@ -43,6 +47,7 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
   GetAll<Model>? get crudGetAll;
   Get<Model>? get crudGet;
   Search<Model>? get crudSearch;
+  Streaming<Model>? get crudStreaming;
   UpdateAll<Model>? get crudUpdateAll;
   Update<Model>? get crudUpdate;
 
@@ -50,6 +55,8 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
   /// Used to identify a model.
   ModelIdentifier<Model> get modelIdentifier;
 
+  StreamSubscription<Result<List<Model?>, AppException>>? _streamSubscription;
+
   FutureOr<void> create(Model model) async {
     final crud = crudCreate;
     if (crud == null) {
@@ -59,6 +66,10 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudCreating());
     final result = await crud.call(model);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       result.fold(
         (_) {
@@ -97,6 +108,10 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudDeleting());
     final result = await crud.call(id);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       result.fold(
         (_) {
@@ -139,6 +154,10 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudDeleting());
     final result = await crud.call(null);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       result.fold(
         (_) {
@@ -202,6 +221,31 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
     );
   }
 
+  FutureOr<void> streaming({
+    String? id,
+    List<Query>? conditions,
+  }) async {
+    final crud = crudStreaming;
+    if (crud == null) {
+      return;
+    }
+    final result = await crud.call(
+      StreamParameters(
+        id: id,
+        conditions: conditions,
+      ),
+    );
+
+    _streamSubscription = result.ok?.listen((event) {
+      emit(
+        event.fold(
+          CrudListLoaded<Model?>.new,
+          (error) => CrudError(error.toString()),
+        ),
+      );
+    });
+  }
+
   FutureOr<void> update(UpdateParameters<Model> param) async {
     final crud = crudUpdate;
     if (crud == null) {
@@ -211,6 +255,10 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudUpdating());
     final result = await crud.call(param);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       await result.foldAsync(
         (_) async {
@@ -284,6 +332,10 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudUpdating());
     final result = await crud.call(param);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       await result.foldAsync(
         (_) async {
@@ -334,4 +386,10 @@ abstract class CrudAdvancedCubit<Model> extends CrudBaseCubit {
       ),
     );
   }
+
+  @override
+  Future<void> close() async {
+    await _streamSubscription?.cancel();
+    return super.close();
+  }
 }

packages/wyatt_crud_bloc/lib/src/features/crud/blocs/crud_cubit.dart

@@ -16,6 +16,7 @@
 
 import 'dart:async';
 
+import 'package:wyatt_architecture/wyatt_architecture.dart';
 import 'package:wyatt_crud_bloc/src/core/enums/where_query_type.dart';
 import 'package:wyatt_crud_bloc/src/core/mixins/operation.dart';
 import 'package:wyatt_crud_bloc/src/core/model_identifier.dart';
@@ -25,11 +26,14 @@ import 'package:wyatt_crud_bloc/src/domain/usecases/delete.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/delete_all.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/get.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/get_all.dart';
+import 'package:wyatt_crud_bloc/src/domain/usecases/params/stream_parameters.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/params/update_parameters.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/search.dart';
+import 'package:wyatt_crud_bloc/src/domain/usecases/streaming.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/update.dart';
 import 'package:wyatt_crud_bloc/src/domain/usecases/update_all.dart';
 import 'package:wyatt_crud_bloc/src/features/crud/blocs/crud_base_cubit/crud_base_cubit.dart';
+import 'package:wyatt_type_utils/wyatt_type_utils.dart';
 
 typedef DefaultCreate<Model> = CreateOperation<Model, void>;
 typedef DefaultRead = ReadOperation<dynamic, dynamic>;
@@ -63,6 +67,8 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
   /// Used to identify a model.
   ModelIdentifier<Model> get modelIdentifier;
 
+  StreamSubscription<Result<List<Model?>, AppException>>? _streamSubscription;
+
   Expected? _checkOperation<Expected>(dynamic operation) {
     if (operation == null) {
       return null;
@@ -90,9 +96,11 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
         conditions != null) {
       return _search(conditions);
     }
-    if (_checkOperation<Stream<Model>>(readOperation) != null &&
-        conditions == null) {
-      return _getAll();
+    if (_checkOperation<Streaming<Model>>(readOperation) != null) {
+      return _streaming(
+        id: id,
+        conditions: conditions,
+      );
     }
   }
 
@@ -129,6 +137,11 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudCreating());
     final result = await crud.call(model);
+    final crudStreaming = _checkOperation<Streaming<Model>>(readOperation);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       result.fold(
         (_) {
@@ -167,6 +180,11 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudDeleting());
     final result = await crud.call(id);
+    final crudStreaming = _checkOperation<Streaming<Model>>(readOperation);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       result.fold(
         (_) {
@@ -209,6 +227,11 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudDeleting());
     final result = await crud.call(null);
+    final crudStreaming = _checkOperation<Streaming<Model>>(readOperation);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       result.fold(
         (_) {
@@ -256,6 +279,31 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
     );
   }
 
+  FutureOr<void> _streaming({
+    String? id,
+    List<Query>? conditions,
+  }) async {
+    final crud = _checkOperation<Streaming<Model>>(readOperation);
+    if (crud == null) {
+      return;
+    }
+    final result = await crud.call(
+      StreamParameters(
+        id: id,
+        conditions: conditions,
+      ),
+    );
+
+    _streamSubscription = result.ok?.listen((event) {
+      emit(
+        event.fold(
+          CrudListLoaded<Model?>.new,
+          (error) => CrudError(error.toString()),
+        ),
+      );
+    });
+  }
+
   FutureOr<void> _search(List<Query> conditions) async {
     final crud = _checkOperation<Search<Model>>(readOperation);
     if (crud == null) {
@@ -281,6 +329,11 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudUpdating());
     final result = await crud.call(param);
+    final crudStreaming = _checkOperation<Streaming<Model>>(readOperation);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       await result.foldAsync(
         (_) async {
@@ -360,6 +413,11 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
     final stateCopy = state;
     emit(const CrudUpdating());
     final result = await crud.call(param);
+    final crudStreaming = _checkOperation<Streaming<Model>>(readOperation);
+    if (crudStreaming != null && _streamSubscription != null && result.isOk) {
+      // If streaming is available, we don't need to update stateCopy.
+      return;
+    }
     emit(
       await result.foldAsync(
         (_) async {
@@ -410,4 +468,10 @@ abstract class CrudCubit<Model> extends CrudBaseCubit {
       ),
     );
   }
+
+  @override
+  Future<void> close() async {
+    await _streamSubscription?.cancel();
+    return super.close();
+  }
 }

packages/wyatt_crud_bloc/pubspec.yaml

@@ -21,6 +21,7 @@ dependencies:
   wyatt_type_utils:
     hosted: https://git.wyatt-studio.fr/api/packages/Wyatt-FOSS/pub/
     version: ^0.0.5
+  rxdart: ^0.27.7
 
 dev_dependencies:
   flutter_test: { sdk: flutter }