process method

Future<TaskProcessOutcome> process(Envelope envelope, { int? deliveryAttempt, TaskStatus? existingStatus, TaskExecutionControl control = const TaskExecutionControl(), })

Processes envelope and returns a transport-independent outcome.

deliveryAttempt overrides the envelope attempt for runtimes whose native retry counter is stored outside the message body. It is zero-based. existingStatus allows the caller to suppress a previously completed duplicate without giving the processor ownership of persistence.

Implementation

Future<TaskProcessOutcome> process(
  Envelope envelope, {
  int? deliveryAttempt,
  TaskStatus? existingStatus,
  TaskExecutionControl control = const TaskExecutionControl(),
}) async {
  if (deliveryAttempt != null && deliveryAttempt < 0) {
    throw ArgumentError.value(
      deliveryAttempt,
      'deliveryAttempt',
      'must be zero or greater',
    );
  }
  final effectiveEnvelope = deliveryAttempt == null
      ? envelope
      : envelope.copyWith(attempt: deliveryAttempt);

  if (existingStatus?.state.isTerminal ?? false) {
    return TaskProcessSkipped(
      envelope: effectiveEnvelope,
      existingStatus: existingStatus!,
    );
  }

  final handler = registry.resolve(effectiveEnvelope.name);
  if (handler == null) {
    return TaskProcessRejected(
      envelope: effectiveEnvelope,
      reason: TaskRejectionReason.unregisteredTask,
    );
  }

  final resolvedSigner = signer;
  if (resolvedSigner != null) {
    try {
      // Verify the wire body, not host-supplied delivery metadata. Native
      // retries retain the original signed body while advancing the counter.
      await resolvedSigner.verify(envelope);
    } on Object catch (error, stackTrace) {
      return TaskProcessRejected(
        envelope: effectiveEnvelope,
        reason: TaskRejectionReason.invalidSignature,
        error: error,
        stackTrace: stackTrace,
      );
    }
  }

  if (_isExpired(effectiveEnvelope)) {
    return TaskProcessCancelled(
      envelope: effectiveEnvelope,
      reason: TaskCancellationReason.expired,
    );
  }

  late final Map<String, Object?> decodedArgs;
  try {
    decodedArgs = _decodeArgs(effectiveEnvelope, handler);
  } on Object catch (error, stackTrace) {
    return TaskProcessRejected(
      envelope: effectiveEnvelope,
      reason: TaskRejectionReason.invalidPayload,
      error: error,
      stackTrace: stackTrace,
    );
  }

  final context = TaskContext(
    id: effectiveEnvelope.id,
    args: effectiveEnvelope.args,
    attempt: effectiveEnvelope.attempt,
    headers: effectiveEnvelope.headers,
    meta: effectiveEnvelope.meta,
    heartbeat: control.heartbeat,
    extendLease: control.extendLease,
    progress: control.progress,
    cancellation: control.cancellation,
    enqueuer: control.enqueuer,
    workflows: control.workflows,
    workflowEvents: control.workflowEvents,
  );

  try {
    control.cancellation.throwIfCancelled();
    final value = await invoke(
      context,
      () => _invokeHandler(
        handler,
        context,
        decodedArgs,
        hardTimeout: _resolveHardTimeLimit(
          effectiveEnvelope,
          handler.options,
        ),
      ),
    );
    return TaskProcessSuccess(envelope: effectiveEnvelope, value: value);
  } on TaskCancellationException catch (error) {
    return TaskProcessCancelled(
      envelope: effectiveEnvelope,
      reason: TaskCancellationReason.cancelled,
      error: error,
    );
  } on TaskRetryRequest catch (request, stackTrace) {
    return classifyRetry(
      effectiveEnvelope,
      handler,
      request,
      stackTrace,
    );
  } on Object catch (error, stackTrace) {
    await _notifyError(context, error, stackTrace);
    return classifyFailure(
      effectiveEnvelope,
      handler,
      error,
      stackTrace,
    );
  }
}