enqueue<TPayload> method

@override Future<DVJobEnvelope<TPayload>> enqueue<TPayload>(String queue, TPayload payload, { int priority = 0, int maxAttempts = 3, Duration backoff = const Duration(seconds: 30), String? tenant, })

Stores payload on queue.

tenant is the tenant the job was dispatched under, and an adapter stores it and hands it back on the envelope reserve returns. Dropping it is not a smaller feature: the worker would run the job as no tenant, which reads nothing on a scoped table and writes rows nobody can see.

Implementation

@override
Future<DVJobEnvelope<TPayload>> enqueue<TPayload>(
  String queue,
  TPayload payload, {
  int priority = 0,
  int maxAttempts = 3,
  Duration backoff = const Duration(seconds: 30),
  String? tenant,
}) async {
  await initialize();
  final codec = DVJobPayloadCodecs._byType[TPayload];
  if (codec == null) {
    throw StateError(
      'No DVJobPayloadCodec registered for $TPayload, so it cannot be '
      'persisted. Register one, or use DVInMemoryQueueAdapter.',
    );
  }

  final envelope = DVJobEnvelope<TPayload>(
    id: 'job-${DateTime.now().microsecondsSinceEpoch}-${_sequence++}',
    queue: queue,
    payloadType: TPayload,
    payload: payload,
    priority: priority,
    maxAttempts: maxAttempts,
    backoff: backoff,
    createdAt: DateTime.now(),
    attempts: 0,
    state: DVJobState.queued,
    tenant: tenant,
  );

  await database.execute(
    'INSERT INTO $tableName (id, queue, payload_name, payload, priority, '
    'max_attempts, backoff_ms, created_at, attempts, state, last_error, '
    'tenant) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)',
    <Object?>[
      envelope.id,
      queue,
      codec.name,
      jsonEncode(codec.encode(payload)),
      priority,
      maxAttempts,
      backoff.inMilliseconds,
      envelope.createdAt.millisecondsSinceEpoch,
      0,
      DVJobState.queued.name,
      tenant,
    ],
  );
  return envelope;
}