【Flutter】Cola secuencial + reintentos en Riverpod: por qué separé Notifier y AsyncNotifier.family

1. Contexto

Cuando desarrollo una aplicación con Flutter, a veces encuentro una función en la que se ejecuta secuencialmente el proceso asincrónico de cada línea registrada y luego se refleja su resultado.

Por ejemplo, este patrón se aplica a los casos siguientes:

• Se cargan las imágenes a la vez, y después de analizarlas por IA, se muestran los resultados individualmente.
• Después de que se escanean los códigos de barras, se consultan las existencias.

¿En este caso, cómo podemos implementarlo?

En este blog, voy a explicar cómo diseñar este requerimiento usando Riverpod.

2. Requisito Previo

Tenemos que considerar las cosas siguientes para que se satisfagan en la app:

1. Todos los ítems se ejecutan secuencialmente para que no pasen el límite de solicitudes.
2. Desde el punto de vista de rendimiento de renderizado, todos los ítems se reconstruyen solo según su estado.
3. El Notifier para gestionar cola secuencial funciona como el desencadenante del proceso asincrónico.
4. No se escribe una lógica sobre la administración del encolamiento, el orden del ejecución y el reintento en la capa de presentación para mantener la app testable y modificable.

3. Mi pensamiento

Primero, pensé que podría realizar este requerimiento con solo un Notifier.

(1) QueueNotifier que administra todos los ítems y el cambio de sus estados

Dejé de adoptar este método porque la lisa completa necesita reconstruirse cada vez que se completa un proceso.

(2) Cada ítem cambia su estado propio y observa el cambio

El segundo es que cada línea hace ref.watch de AsyncNotifier.family, y se ejecuta el método de Repository Class o UseCase Class a la hora de llamar a build() de este Notifier.
Sin embargo, cuando lo uso, 1) se ignora el orden de ejecución y 2) se generan muchos solicitud y puede alcanzar el límite de tasa.

(3) Uso dos Notifier para separar responsabilidades

El último método que consideré es usar dos Notifier. Un Notifier administra el proceso asincrónico en la cola general. Y el otro cambia el estado del ítem según su resultado. O sea, es un método híbrido entre (1) y (2).

Lo que más atención requiere es cómo se inicializa el estado del Notifier de (2). Si build() devuelve un valor vacío como valor inicial, no se puede distinguir si el resultado aún no ha llegado o ya llegó en la capa de presentación.

Por eso, utilizo Completer<Result>().future en build() y invoco un objeto de Future incompleto para expresar el estado en el que todavía no se termina.

Nota:

En este caso, se utiliza typedef TargetId = int como identificador de cada objeto a procesar. Si quiere usar un objeto como identificador, conviene utilizar los paquetes, Equatable o freezed, que le permiten expresar igualdad de valores.

4. Implementación

Implemento no solo ejecución secuencial mediante una cola sino también una funcionalidad de reintento que permite tocar un ítem cuyo procesamiento asíncrono falló por un error temporal, para volver a intentarlo.

Se busca lograr una integración bidireccional como la que se muestra en el siguiente diagrama.

Queue

QueueState

• TargetStatus
• Target
• QueueState

typedef TargetId = int;

enum TargetStatus { pending, scanning, success, failure }

class Target {
  const Target({required this.id, required this.status});

  final TargetId id;
  final TargetStatus status;

  Target copyWith({TargetId? id, TargetStatus? status}) {
    return Target(id: id ?? this.id, status: status ?? this.status);
  }
}

@immutable
class QueueState {
  const QueueState({required this.targets});

  final List<Target> targets;

  QueueState updateStatus(TargetId id, TargetStatus status) {
    return copyWith(
      targets: [
        for (final target in targets)
          target.id == id ? target.copyWith(status: status) : target,
      ],
    );
  }

  Target? get firstPending {
    for (final target in targets) {
      if (target.status == TargetStatus.pending) {
        return target;
      }
    }
    return null;
  }

  QueueState copyWith({List<Target>? targets}) {
    return QueueState(targets: targets ?? this.targets);
  }
}

QueueNotifier

final queueProvider = NotifierProvider<QueueNotifier, QueueState>(
  QueueNotifier.new,
);

class QueueNotifier extends Notifier<QueueState> {
  bool _executing = false;

  @override
  QueueState build() => QueueState(
    targets: [
      for (int i = 0; i < 10; i++) Target(id: i, status: TargetStatus.pending),
    ],
  );

  Future<void> executeSequentially() async {
    if (_executing) {
      return;
    }
    _executing = true;

    final useCase = processUseCase();

    try {
      while (true) {
        final poppedTarget = state.firstPending;
        if (poppedTarget == null) {
          break;
        }

        state = state.updateStatus(poppedTarget.id, TargetStatus.scanning);

        try {
          final result = await useCase.execute(poppedTarget);
          ref
              .read(resultProvider(poppedTarget.id).notifier)
              .executed(Result(status: ResultStatus.success, data: result));
          state = state.updateStatus(poppedTarget.id, TargetStatus.success);
        } catch (e) {
          ref
              .read(resultProvider(poppedTarget.id).notifier)
              .executed(
                Result(status: ResultStatus.failure, message: e.toString()),
              );
          state = state.updateStatus(poppedTarget.id, TargetStatus.failure);
        }
      }
    } finally {
      _executing = false;
    }
  }

  Future<void> retry(TargetId id) async {
    state = state.updateStatus(id, TargetStatus.pending);
    await executeSequentially();
  }
}

Resultado

Result

• ResultStatus
• Result

enum ResultStatus{ success, failure }

@immutable
class Result {
  const Result({
    required this.status,
    this.data,
    this.message
  });
  final ResultStatus status;
  final Map<String, Object>? data;
  final String? message;
}

ResultNotifier

final resultProvider =
    AsyncNotifierProvider.family<ResultNotifier, Result, TargetId>(
      ResultNotifier.new,
    );

class ResultNotifier extends FamilyAsyncNotifier<Result, TargetId> {
  @override
  Future<Result> build(TargetId id) => Completer<Result>().future;

  void executed(Result result) {
    state = AsyncData(result);
  }

  Future<void> retry(TargetId id) async {
    state = const AsyncLoading();
    await ref.read(queueProvider.notifier).retry(id);
  }
}

Implementación en capa de presentación

Obtiene la lista completa desde QueueNotifier y el estado de cada ítem desde ResultNotfier. Así que el cambio del estado no provoca reconstrucción del otro.

 // Lista completa
final targets = ref.watch(queueProvider).targets;

 // Cada ítem
final asyncResult = ref.watch(resultProvider(target.id));
return asyncResult.when(
  loading: () => LoadingTile(),
  data: (result) => ResultTile(result),
  error: (e, _) => ErrorTile(),
);

5. Resumen

• Completer<Result>().future permite expresar que el resultado aún no ha llegado sin renunciar a la seguridad de tipos.
• Separar la gestión de la cola de la gestión del resultado permite acotar la reconstrucción parcial.
• Se logra que la ejecución secuencial y el reintento convivan sin conflicto por una transición de estado bidireccional con el guard _executing en el lado de la cola y AsyncLoading en el lado del resultado.