run method

Future<void> run()

Starts the consumer loops and resolves only after stop is called (or an unrecoverable error). Run this as the body of a worker process.

Implementation

Future<void> run() async {
  if (_running) return;
  _running = true;
  app.logger.info(
      'worker started: queues=${queues.join(',')} concurrency=$concurrency');
  _controlSub = app.controlChannel.subscribe().listen(
        _handleControlMessage,
        onError: (Object e, StackTrace st) =>
            app.logger.warning('control channel error: $e'),
      );
  await _emitWorker(QueueEventType.workerOnline);
  _heartbeat = Timer.periodic(heartbeatInterval, (_) {
    _emitWorker(QueueEventType.workerHeartbeat);
    _pruneRevokedIds();
  });
  for (var i = 0; i < concurrency; i++) {
    _loops.add(_consumeLoop(i));
  }
  await Future.wait(_loops);
  app.logger.info('worker stopped');
}