Watcher

The Watcher state flow

Overview

A Watcher is a type of use case that streams data and continuously emits either a Left Event (error) or a Right Event (data) as long as the corresponding stream is active.

However, before a stream can start emitting events, it must first be started. If the stream fails to start, then a Left Value (error) will be emitted. If the stream starts successfully, then a Right Value (a VerboseStream containing the stream being listened to) will be emitted.

A Watcher can emit 7 possible class states, all of which inherit the WatcherState sealed class:

Watcher States	Description
WatcherInitial	The initial state or the state emitted when the use case has been reset.
StartWatching	The state emitted when the stream creation is in progress.
StartWatchSuccess	The state emitted when the stream creation succeeds.
StartWatchFailed	The state emitted when the stream creation fails.
WatchDataReceived	The state emitted when the stream emits a new data.
WatchErrorReceived	The state emitted when the stream emits an error.
WatchDone	The state emitted when the stream has been closed.

A Watcher also has the following properties that you can access:

Property	Description
value	The latest value returned when calling watch(). This may either be the leftValue if StartWatchFailed was recently emitted. Otherwise, this will be equal to the rightValue if StartWatchSuccess was more recent.
params	The latest params returned when calling watch(). This may either be the leftParams if StartWatchFailed was recently emitted. Otherwise, this will be equal to the rightParams if StartWatchSuccess was more recent.
leftValue	The last left value returned when a failed watch() was called.
leftParams	The last left params passed when a failed watch() was called.
rightValue	The last right value returned when a successful watch() was called. This references a VerboseStream containing the stream being listened to.
rightParams	The last right params passed when a successful watch() was called.
event	The latest event emitted by the Watcher. This can either be the leftEvent if WatchErrorReceived was recently emitted. Otherwise, this will be equal to the rightEvent if WatchDataReceived was more recent.
leftEvent	The last error event emitted by the Watcher.
rightEvent	The last data event emitted by the Watcher.

Creating a Watcher

Let's implement a Watcher use case called WatchFruitBasket that does the following:

• Add fruits to the basket.

• Emits all the fruits in the basket when new ones are added.

Begin by creating the following classes:

1. Create the Parameter class

This is the parameter class passed to the Watcher when creating and starting the stream.

/// The parameter for the [WatchFruitBasket] use case to specify the max 
/// capacity allowed in the fruit basket.
class WatchFruitBasketParams {
  const WatchFruitBasketParams({required this.maxCapacity});

  /// The max number of fruits allowed in the fruit basket.
  final int maxCapacity;

  @override
  String toString() => toMap().toString();

  Map<String, dynamic> toMap() => {'maxCapacity': maxCapacity};
}

2. Create the Left event class

This is the object returned when the Watcher emits a StartWatchFailed or a WatchErrorReceived state.

class Failure {
  const Failure(this.message);

  final String message;

  @override
  String toString() => 'Failure: $message';
}

3. Create the Right event class

This is the object returned when the Watcher emits a WatchDataReceived state.

/// The right value for the [FruitBasket] use case representing the fruit 
/// basket.
class FruitBasket {
  const FruitBasket(this.fruits);

  /// The fruits in the basket.
  final List<String> fruits;

  @override
  String toString() => 'FruitBasket: $fruits';
}

4. Create the Watcher

Pass the Parameter, Left and Right event classes to the Runner's generic arguments. Afterwards, implement the code logic in the onCall method:

/// A watcher for streaming fruits that goes inside the fruit basket.
class WatchFruitBasket
    extends Watcher<WatchFruitBasketParams, Failure, FruitBasket> {
  /// The stream controller for the fruit basket.
  StreamController<FruitBasket>? streamController;

  /// The max number of fruits allowed in the fruit basket.
  int? basketCapacity;

  /// The fruits currently in the fruit basket.
  List<String>? fruits;

  /// A callback function triggered when the [watch] method is called.
  @override
  FutureOr<Either<Failure, VerboseStream<Failure, FruitBasket>>> onCall(
    WatchFruitBasketParams params,
  ) async {
    if (params.maxCapacity < 1) {
      // When the basket capacity is less than 1, then a left value is returned
      return const Left(Failure('Basket capacity must be greater than 0'));
    }

    basketCapacity = params.maxCapacity;
    fruits = [];
    await streamController?.close();

    // Create a new stream controller
    streamController = StreamController<FruitBasket>();

    // Return a right value `VerboseStream` containing the stream that will be
    // listened to and its error converter
    //
    // A [VerboseStream] is a [Stream] wrapper which gives it an error
    // handler for converting [Exception]s into [Failure]s before it gets 
    // emitted
    return Right(
      VerboseStream(
        stream: streamController!.stream,
        errorConverter: (error, stackTrace) => Failure(error.toString()),
      ),
    );
  }

  /// A helper method to add fruits to the fruit basket.
  void addFruits(List<String> newFruits) {
    if (fruits == null || basketCapacity == null || streamController == null) {
      return;
    }

    if (fruits!.length + newFruits.length <= basketCapacity!) {
      fruits!.addAll(newFruits);
      streamController!.add(FruitBasket(fruits!));
    } else {
      streamController!.addError(Exception('Fruit Basket is full'));
    }
  }

  /// A helper method for closing the stream controller.
  Future<void> closeStream() async => streamController?.close();

  @override
  Future<void> close() {
    streamController?.close();
    return super.close();
  }
}

Using a Watcher

1. Start the Watcher and emit an event

To start the Watcher stream, call the watch method:

final watchFruitBasket = WatchFruitBasket();

// Initiate the stream
await watchFruitBasket.watch(
  params: const WatchFruitBasketParams(maxCapacity: 5),
);

// Use the helper method to add fruits and emit a Right event
await watchFruitBasket.addFruits(['Apple', 'Orange', 'Mango']);

The Watcher will emit the StartWatching state while the stream is being initiated. If it is successful, then a StartWatchSuccess state together with a Right Value will be emitted. Otherwise, a StartWatchFailed state will be emitted containing Left Value.

The Right Value holds a VerboseStream object which contains the stream being listened to.

Once the Watcher successfully starts watching, a WatchDataReceived state will be emitted containing the Right Event whenever the stream receives new data. On the other hand, if the stream receives an error, then a WatchErrorReceived will be emitted alongside a Left Event.

2. View the Watcher properties

The Watcher gives you access to the following properties:

// The latest value returned when calling `watch()`. This may either be the 
// `leftValue` if `StartWatchFailed` was recently emitted. Otherwise, this 
// will be equal to the `rightValue` if `StartWatchSuccess` was more recent
print('Current value: ${watcher.value}');

// The latest params returned when calling `watch()`. This may either be the 
// `leftParams` if `StartWatchFailed` was recently emitted. Otherwise, this 
// will be equal to the `rightParams` if `StartWatchSuccess` was more recent
print('Current params: ${watcher.params}');

// The last left value returned when a failed `watch()` was called
print('Last left value: ${watcher.leftValue}');

// The last left params passed when a failed `watch()` was called
print('Last left params: ${watcher.leftParams}');

// The last right value returned when a successful `watch()` was called
print('Last right value: ${watcher.rightValue}');

// The last right params passed when a successful `watch()` was called
print('Last right params: ${watcher.rightParams}');

// The latest event emitted by the Watcher. This can either be the 
// `leftEvent` if `WatchErrorReceived` was recently emitted. Otherwise, this 
// will be equal to the `rightEvent` if `WatchDataReceived` was more recent
print('Current event: ${watcher.event}');

// The last error event emitted by Watcher
print('Last left event: ${watcher.leftEvent}');

// The last data event emitted by the Watcher
print('Last right event: ${watcher.rightEvent}');

3. Reset the Watcher

To clear the Watcher and reset it back to its initial state, call the reset method:

runner.reset();

Flutter Example

Every use case is a descendent of BLoC cubit. Hence, we can manage its states via the flutter_bloc package.

BlocProvider(
  lazy: false,
  create: (context) =>
    // Initialize and eagerly run the Watcher
    WatchFruitBasket()
      ..watch(params: const WatchFruitBasketParams(maxCapacity: 5)),
  child: BlocConsumer<WatchFruitBasket, WatcherState>(
    listener: (context, watcherState) {
      if (watcherState is WatcherInitial) {
      } else if (watcherState is StartWatching) {
        // Handle start watch
      } else if (watcherState is StartWatchFailed<Failure>) {
        // Handle start watch failure
        print('Left value: ${watcherState.leftValue}');
      } else if (watcherState is StartWatchSuccess) {
        // Handle start watch success. Add fruits to emit an event
        context.read<WatchFruitBasket>().addFruits(['Apple', 'Orange']);
      } else if (watcherState is WatchErrorReceived<Failure>) {
        // Handle an error event
        print('Left event: ${watcherState.leftEvent}');
      } else if (watcherState is WatchDataReceived<FruitBasket>) {
        // Handle a data event
        print('Right event: ${watcherState.rightEvent}');
      } else if (watcherState is WatchDone) {
        // Handle stream closed
      }
    },
    builder: (context, watcherState) {
      switch (watcherState) {
        case WatcherInitial():
          return Container();
        case StartWatching():
          return const Center(child: CircularProgressIndicator());
        case StartWatchFailed():
          return Center(
            child: Text('Start watch failed: ${watcherState.leftValue}'),
          );
        case StartWatchSuccess():
          return const Center(child: Text('Start watch success'));
        case WatchErrorReceived():
          return Center(
            child: Text('Left event: ${watcherState.leftEvent}'),
          );
        case WatchDataReceived():
          return Center(
            child: Text(
              'Right event: ${watcherState.rightEvent}',
            ),
          );
        case WatchDone():
          return const Center(child: Text('Watch done'));
      }
    },
  ),
);

Dart Example

To run the Dart demo:

1. Clone to codenic_clean_arch repository.

2. Navigate to the packages/codenic_bloc_use_case directory.

3. Run the example code:

dart run example/main.dart