Messaging

Use messaging for work that shouldn't block a request, such as sending an invoice after an order is placed. Annotate a controller method with @Consumes, and return a broker from AppConfig.createBroker(). Revali doesn't run a broker itself: you deploy one, such as Redis, the same way you deploy a database.

Setup#

Messaging needs both of these. Without either one, no consumers are registered and nothing warns you:

1. A controller method annotated with @Consumes(topic, group:).
2. A broker returned from AppConfig.createBroker(). The default returns null.

import 'package:revali_router/revali_router.dart';

@Controller('orders')
class OrdersController {
  const OrdersController(this._invoices);

  final InvoiceService _invoices;

  @Consumes('order.placed', group: 'billing')
  Future<void> onPlaced(BrokerMessage message) async {
    await _invoices.create(message.json);
  }
}

import 'package:revali_redis/revali_redis.dart';
import 'package:revali_router/revali_router.dart';

@App()
final class MainApp extends AppConfig {
  MainApp() : super.fromEnv();

  @override
  Future<MessageBroker?> createBroker() => RedisBroker.connect(
        host: Env.current.string('REDIS_HOST', orElse: 'localhost'),
        consumerName: Env.current.string('HOSTNAME', orElse: 'revali'),
      );
}

The framework owns the broker from then on, and drains and closes it during shutdown. createBroker() runs once in every worker isolate.

Publish from anywhere that has the broker. Pass the trace headers so the consumer's logs can be matched to the request that published the message:

await broker.publish(
  'order.placed',
  jsonEncode(order),
  headers: TraceContext.current?.outboundHeaders() ?? const {},
);

Handler Rules#

• A handler takes either one BrokerMessage parameter or none.
• message.json decodes the payload and throws if it isn't JSON. message.payload is the raw string. message.headers holds metadata.
• Consumers live on controllers alongside routes and use the same controller instance. A controller with InstanceType.factory is created once per message.
• Guards and middleware don't run for messages, because a message has no caller to reject. Check authorization in the handler, using data the publisher put in the payload.
• Each message gets its own TraceContext, taken from the message headers, and its own request scope.

These mistakes fail generation with an error that names the controller and method:

• more than one parameter
• a parameter that isn't a BrokerMessage
• two @Consumes annotations on one method
• a method that is both a route and a consumer
• an empty topic or group

Consumer Groups#

• Handlers in the same group split the messages between them: each message goes to exactly one handler. Two replicas in group billing don't both create the invoice.
• Different groups on the same topic each get their own copy of every message.

group has no default. A group name is a stored identity, and renaming a group re-reads the whole stream.

@Consumes('order.placed', group: 'billing')
Future<void> bill(BrokerMessage message) async { /* ... */ }

@Consumes('order.placed', group: 'shipping')   // also receives every message
void ship(BrokerMessage message) { /* ... */ }

Delivery Is At Least Once#

Handlers must be idempotent. A handler that throws doesn't acknowledge its message, so the message is delivered again. A message whose acknowledgment was lost (because the process died, for example) is also delivered again. BrokerMessage.id identifies one delivery, not the event, so deduplicate on an id from your own data, such as the order id. Or make the effect safe to repeat, for example with an upsert.

Shutdown: Consumers Drain Before HTTP#

On SIGTERM, consumers are drained first, then HTTP requests, then onServerStopped() runs. Draining pauses the subscriptions, so no new messages are taken, and waits for the handlers already running, for up to shutdownTimeout. The request drain waits for up to shutdownTimeout too, so shutdown can take twice that long in the worst case.

Testing with InMemoryBroker#

InMemoryBroker runs in-process. It supports consumer groups, redelivers a message when its handler throws (turn that off with redeliver: false), and keeps messages published before anyone subscribed. Nothing survives a restart. published lists every message the broker accepted.

Keep the broker in a static field in a test flavor, so the test can reach the same instance the server uses:

import 'package:revali_router/revali_router.dart';

class TestBroker {
  static InMemoryBroker? instance;
}

@App(flavor: 'test')
final class TestApp extends AppConfig {
  const TestApp() : super(host: 'localhost', port: 0);

  @override
  Future<MessageBroker?> createBroker() async =>
      TestBroker.instance = InMemoryBroker();
}

test('handles order.placed', () async {
  final server = TestServer();
  await createServer(server);

  await TestBroker.instance!.publish('order.placed', '{"id": 1}');

  // assert on the handler's side effects
});

Generate with --flavor test so the server uses this app. See Testing.

Redis (revali_redis)#

revali_redis implements MessageBroker on Redis Streams. It has no dependencies other than revali_core.

dependencies:
  revali_redis:

Every option can be passed to RedisBroker.connect:

The broker behaves as follows:

• The consumer group is created together with the stream, so consumers can start before anything is published. Starting more replicas is safe.
• The broker reconnects when the connection drops. If Redis restarts without persistence and loses the consumer group, the broker creates the group again and keeps reading.
• TLS and password authentication aren't supported yet.

Name Your Replicas#

Redis tracks pending messages per consumer name. When two replicas share a name, neither can see the other's pending messages. Give each replica its own name, such as the pod name or hostname.

Worker isolates are named for you. The parent keeps the configured name, and workers get <name>-1, <name>-2, and so on. If you write your own broker, pass the name through IsolateIdentity.scopeName(name) to get the same behavior.