Sequential Cubit

main.dart

/*
 * Sequential Cubit
 * https://gist.github.com/PlugFox/a183c3c804a3369efe8ad3584f0550ac
 * https://dartpad.dev?id=a183c3c804a3369efe8ad3584f0550ac
 * Matiunin Mikhail <plugfox@gmail.com>, 15 February 2024
 */

import 'dart:async';
import 'dart:collection';

import 'package:flutter/foundation.dart';
import 'package:flutter/material.dart';
import 'package:flutter_bloc/flutter_bloc.dart';

void main() => runZonedGuarded<void>(
      () => runApp(const CubitExampleApp()),
      (error, stackTrace) => print('Top level exception: $error\n$stackTrace'), // ignore: avoid_print
    );

class CubitExampleApp extends StatelessWidget {
  const CubitExampleApp({super.key});

  @override
  Widget build(BuildContext context) => BlocProvider<CounterSequentialCubit>(
        create: (context) => CounterSequentialCubit(),
        child: MaterialApp(
          title: 'Cubit example',
          theme: ThemeData.dark(),
          builder: (context, _) => Scaffold(
            appBar: AppBar(
              title: const Text('Cubit example'),
              centerTitle: true,
            ),
            floatingActionButton: RepaintBoundary(
              child: Column(
                mainAxisSize: MainAxisSize.min,
                mainAxisAlignment: MainAxisAlignment.end,
                crossAxisAlignment: CrossAxisAlignment.end,
                children: <Widget>[
                  FloatingActionButton.large(
                    onPressed: () => context.read<CounterSequentialCubit>().increment(),
                    child: const Icon(Icons.add),
                  ),
                  const SizedBox(height: 8),
                  FloatingActionButton.large(
                    onPressed: () => context.read<CounterSequentialCubit>().decrement(),
                    child: const Icon(Icons.remove),
                  ),
                ],
              ),
            ),
            body: SafeArea(
              child: Center(
                child: BlocBuilder<CounterSequentialCubit, CounterState>(
                  builder: (context, state) => Column(
                    mainAxisSize: MainAxisSize.min,
                    mainAxisAlignment: MainAxisAlignment.center,
                    crossAxisAlignment: CrossAxisAlignment.center,
                    children: <Widget>[
                      Text(
                        state.value.toString(),
                        style: const TextStyle(fontSize: 72),
                        maxLines: 1,
                        overflow: TextOverflow.ellipsis,
                      ),
                      Text(
                        state.message,
                        style: const TextStyle(fontSize: 48),
                        maxLines: 1,
                        overflow: TextOverflow.ellipsis,
                      ),
                    ],
                  ),
                ),
              ),
            ),
          ),
        ),
      );
}

class CounterState {
  const CounterState(this.value, this.message);

  final int value;
  final String message;
}

class CounterCubit extends Cubit<CounterState> {
  CounterCubit() : super(const CounterState(0, 'Idle'));

  Future<void> increment() async {
    try {
      final currentValue = state.value;
      emit(CounterState(currentValue, 'Loading...'));
      final newValue = await Future<int>.delayed(const Duration(seconds: 5), () => currentValue + 1);
      emit(CounterState(newValue, 'Success'));
    } on Object catch (error, stackTrace) {
      onError(error, stackTrace);
      emit(CounterState(state.value, 'An error occurred'));
    } finally {
      emit(CounterState(state.value, 'Idle'));
    }
  }

  Future<void> decrement() async {
    try {
      final currentValue = state.value;
      emit(CounterState(currentValue, 'Loading...'));
      final newValue = await Future<int>.delayed(const Duration(seconds: 5), () => currentValue - 1);
      emit(CounterState(newValue, 'Success'));
    } on Object catch (error, stackTrace) {
      onError(error, stackTrace);
      emit(CounterState(state.value, 'An error occurred'));
    } finally {
      emit(CounterState(state.value, 'Idle'));
    }
  }
}

// --- Sequential Cubit --- //

class CounterSequentialCubit extends SequentialCubit<CounterState> {
  CounterSequentialCubit() : super(const CounterState(0, 'Idle'));

  Future<void> increment() => handle<void>(
        (emit) async {
          final currentValue = state.value;
          emit(CounterState(currentValue, 'Loading...'));
          final newValue = await Future<int>.delayed(const Duration(seconds: 5), () => currentValue + 1);
          emit(CounterState(newValue, 'Success'));
        },
        (emit, error, stackTrace) async {
          emit(CounterState(state.value, 'An error occurred'));
        },
        (emit) async {
          emit(CounterState(state.value, 'Idle'));
        },
      );

  Future<void> decrement() => handle<void>(
        (emit) async {
          final currentValue = state.value;
          emit(CounterState(currentValue, 'Loading...'));
          final newValue = await Future<int>.delayed(const Duration(seconds: 5), () => currentValue - 1);
          emit(CounterState(newValue, 'Success'));
        },
        (emit, error, stackTrace) async {
          emit(CounterState(state.value, 'An error occurred'));
        },
        (emit) async {
          emit(CounterState(state.value, 'Idle'));
        },
      );
}

abstract class SequentialCubit<S extends Object> extends Cubit<S> {
  SequentialCubit(super.initialState);

  final _ControllerEventQueue _eventQueue = _ControllerEventQueue();

  /// Whether the event queue is closed.
  @nonVirtual
  bool get isProcessing => _eventQueue.length > 0;

  /// Fire when the event queue is empty.
  Future<void> get done => _eventQueue._processing ?? SynchronousFuture<void>(null);

  /// Use this method to handle asynchronous logic inside the cubit.
  @protected
  @mustCallSuper
  Future<R?> handle<R extends Object?>(
    FutureOr<R> Function(void Function(S state) emit) handler, [
    FutureOr<void> Function(void Function(S state) emit, Object error, StackTrace stackTrace)? errorHandler,
    FutureOr<void> Function(void Function(S state) emit)? doneHandler,
  ]) =>
      _eventQueue.push<R?>(
        () {
          final completer = Completer<R?>();
          void emit(S state) {
            if (isClosed || completer.isCompleted) return;
            super.emit(state);
          }

          Future<void> onError(Object error, StackTrace stackTrace) async {
            try {
              super.onError(error, stackTrace);
              if (isClosed || completer.isCompleted) return;
              await errorHandler?.call(emit, error, stackTrace);
            } on Object catch (error, stackTrace) {
              super.onError(error, stackTrace);
            }
          }

          runZonedGuarded<void>(
            () async {
              if (isClosed) return;
              R? result;
              try {
                result = await handler(emit);
              } on Object catch (error, stackTrace) {
                await onError(error, stackTrace);
              } finally {
                try {
                  await doneHandler?.call(emit);
                } on Object catch (error, stackTrace) {
                  super.onError(error, stackTrace);
                }
                completer.complete(result);
              }
            },
            onError,
          );
          return completer.future;
        },
      ).catchError((_, __) => null);

  @override
  @mustCallSuper
  Future<void> close() => super.close().whenComplete(_eventQueue.close);
}

/// {@nodoc}
final class _ControllerEventQueue {
  /// {@nodoc}
  _ControllerEventQueue();

  final DoubleLinkedQueue<_SequentialTask<Object?>> _queue = DoubleLinkedQueue<_SequentialTask<Object?>>();
  Future<void>? _processing;
  bool _isClosed = false;

  /// Event queue length.
  /// {@nodoc}
  int get length => _queue.length;

  /// Push it at the end of the queue.
  /// {@nodoc}
  Future<T> push<T>(FutureOr<T> Function() fn) {
    final task = _SequentialTask<T>(fn);
    _queue.add(task);
    _exec();
    return task.future;
  }

  /// Mark the queue as closed.
  /// The queue will be processed until it's empty.
  /// But all new and current events will be rejected with [WSClientClosed].
  /// {@nodoc}
  FutureOr<void> close() async {
    _isClosed = true;
    await _processing;
  }

  /// Execute the queue.
  /// {@nodoc}
  void _exec() => _processing ??= Future.doWhile(() async {
        final event = _queue.first;
        try {
          if (_isClosed) {
            event.reject(StateError('Controller\'s event queue are disposed'), StackTrace.current);
          } else {
            await event();
          }
        } on Object catch (error, stackTrace) {
          /* warning(
            error,
            stackTrace,
            'Error while processing event "${event.id}"',
          ); */
          Future<void>.sync(() => event.reject(error, stackTrace)).ignore();
        }
        _queue.removeFirst();
        final isEmpty = _queue.isEmpty;
        if (isEmpty) _processing = null;
        return !isEmpty;
      });
}

/// {@nodoc}
class _SequentialTask<T> {
  /// {@nodoc}
  _SequentialTask(FutureOr<T> Function() fn)
      : _fn = fn,
        _completer = Completer<T>();

  /// {@nodoc}
  final Completer<T> _completer;

  /// {@nodoc}
  final FutureOr<T> Function() _fn;

  /// {@nodoc}
  Future<T> get future => _completer.future;

  /// {@nodoc}
  FutureOr<T> call() async {
    final result = await _fn();
    if (!_completer.isCompleted) {
      _completer.complete(result);
    }
    return result;
  }

  /// {@nodoc}
  void reject(Object error, [StackTrace? stackTrace]) {
    if (_completer.isCompleted) return;
    _completer.completeError(error, stackTrace);
  }
}