replayDeadLetters method

@override Future<DeadLetterReplayResult> replayDeadLetters(String queue, { int limit = 50, DateTime? since, Duration? delay, bool dryRun = false, })

Replays at most limit dead letter entries back onto the active queue.

When since is provided, only entries with a deadAt greater than or equal to the timestamp are considered. If delay is specified, replayed envelopes are scheduled with the provided delay. When dryRun is true, the method MUST NOT modify broker state and instead return the entries that would have been replayed.

Implementation

@override
Future<DeadLetterReplayResult> replayDeadLetters(
  String queue, {
  int limit = 50,
  DateTime? since,
  Duration? delay,
  bool dryRun = false,
}) async {
  if (limit <= 0) {
    return DeadLetterReplayResult(entries: const [], dryRun: dryRun);
  }
  final state = _state(queue);
  final candidates = state.deadLetters.where((entry) {
    if (since == null) return true;
    return !entry.deadAt.isBefore(since);
  }).toList()..sort((a, b) => a.deadAt.compareTo(b.deadAt));
  final selected = candidates.take(limit).toList();
  if (dryRun || selected.isEmpty) {
    return DeadLetterReplayResult(entries: selected, dryRun: true);
  }
  final now = stemNow();
  for (final entry in selected) {
    state.deadLetters.remove(entry);
    final replayEnvelope = entry.envelope.copyWith(
      attempt: entry.envelope.attempt + 1,
      notBefore: delay != null ? now.add(delay) : null,
    );
    await publish(replayEnvelope.copyWith(queue: queue));
  }
  return DeadLetterReplayResult(entries: selected, dryRun: false);
}