Custom Topics

This guide targets the documented 1.3.x release line. Features marked Since 1.3.0 require @connectum/events 1.3.0 or later; they are unavailable from the published 1.2.x package until 1.3.0 is released.

By default, the EventBus routes events using the protobuf message's typeName (e.g., orders.v1.OrderCreated). You can override this with a custom topic name using proto options or publish-time overrides.

Outcome: choose one canonical topic source and verify publishers and handlers resolve the same name. Delivery and broker semantics remain in Events and Adapter selection.

Default Topic Naming

When you register event handlers with events.service(), the EventRouter resolves the topic for each method:

1. Check for a custom (connectum.events.v1.event).topic proto option on the method
2. Fall back to method.input.typeName (the fully-qualified protobuf message name)

An empty topic string in the option is treated as if the option were not set, so the method falls back to method.input.typeName.

For example, this handler subscribes to the topic orders.v1.OrderCreated:

service InventoryEventHandlers {
  // Topic: "orders.v1.OrderCreated" (default — from message typeName)
  rpc OnOrderCreated(OrderCreated) returns (google.protobuf.Empty);
}

Proto Option: Custom Topic

Import the Connectum events options proto and set a custom topic on any method:

syntax = "proto3";

package orders.v1;

import "google/protobuf/empty.proto";
import "connectum/events/v1/options.proto";

service InventoryEventHandlers {
  // Default topic: "orders.v1.OrderCreated"
  rpc OnOrderCreated(OrderCreated) returns (google.protobuf.Empty);

  // Custom topic: "orders.cancelled"
  rpc OnOrderCancelled(OrderCancelled) returns (google.protobuf.Empty) {
    option (connectum.events.v1.event).topic = "orders.cancelled";
  }
}

The connectum/events/v1/options.proto file defines the method option:

// connectum/events/v1/options.proto
syntax = "proto2";

package connectum.events.v1;

import "google/protobuf/descriptor.proto";

message EventOptions {
  optional string topic = 1;
}

extend google.protobuf.MethodOptions {
  optional EventOptions event = 50102;
}

TIP

The options.proto file is included in the @connectum/events package proto directory. Add it to your buf.yaml dependencies or copy it into your project's proto tree.

Since 1.3.0, @connectum/events also exports the generated code of this proto at @connectum/events/gen/connectum/events/v1/options_pb.js — the module the EventBus itself uses to resolve topics. Import it instead of generating a local copy: run protoc-gen-es (2.15.0 or later) with map_imports=connectum/events/v1/:@connectum/events/gen, and if you copied the proto into your tree, keep it out of generation with exclude_paths:

version: v2
clean: true
inputs:
  - directory: proto
    exclude_paths:
      - proto/connectum/events/v1
plugins:
  - local: protoc-gen-es
    out: gen
    opt:
      - target=ts
      - import_extension=.ts
      - map_imports=connectum/events/v1/:@connectum/events/gen

connectum init --events sets this up; see Scaffolding.

Publishing to Custom Topics

The EventBus resolves the publish topic automatically when the publisher process has the relevant service registered — either via routes (subscriber side) or via the publishes option (publisher-only side). In that case no PublishOptions.topic override is needed:

import { OrderCancelledSchema, OrderEventService } from '#gen/orders/v1/orders_pb.js';

// Publisher-only process: declare the service in `publishes` so the
// custom topic from the proto option is resolved automatically.
const eventBus = createEventBus({ adapter, publishes: [OrderEventService] });

// Topic resolves to "orders.cancelled" from the proto option — no manual override needed.
await eventBus.publish(OrderCancelledSchema, {
  orderId: 'abc-123',
  reason: 'Changed my mind',
});

If the publisher has neither routes nor publishes covering the event type, the lookup will be empty and publish() falls back to schema.typeName. In that case you must pass the topic explicitly:

// Fallback: no routes/publishes registered — specify the topic manually.
await eventBus.publish(OrderCancelledSchema, {
  orderId: 'abc-123',
  reason: 'Changed my mind',
}, { topic: 'orders.cancelled' });

Consistency Required

Whichever resolution path you use, the topic must match what the subscriber declared in the proto option. A mismatch means the subscriber will never receive the event.

Topic Resolution Flow

Wildcard Topic Matching

A subscription can use wildcard patterns for topic matching:

Pattern	Matches	Does Not Match
orders.*	orders.created, orders.cancelled	orders.v1.created
orders.>	orders.created, orders.v1.created, orders.v1.created.eu	inventory.reserved
orders.v1.*	orders.v1.OrderCreated, orders.v1.OrderCancelled	orders.v1.sub.topic

Two wildcard tokens are supported:

• * -- matches exactly one dot-separated segment
• > -- matches one or more trailing segments

When one bus has both an exact route and a wildcard pattern that match the same incoming topic, the exact route handles the event. The order in which two matching wildcard patterns are tried is not defined; do not rely on it.

Support and caveats differ by adapter:

Adapter	Wildcards	What to know
MemoryAdapter	Yes	Same matcher as above.
NatsAdapter	Yes, native	The server evaluates the filter for every message, so subjects first used after the subscription are delivered. When several patterns of one subscription match the same subject (orders.created, orders.*, orders.>), the handler still runs once per event; see overlapping patterns.
KafkaAdapter	Yes, as a regex over topic names	A matching topic created later is picked up at consumerOptions.topicDiscoveryInterval (default 5 min; false keeps the list fixed; in a group, after the longest interval among its members). A pattern that opens with a wildcard never matches topics starting with __ (Kafka's internal topics).
AmqpAdapter	Yes, on a topic exchange	> becomes *.#, so user.> needs at least one more segment. Other exchange types do not match patterns: a direct exchange binds the pattern as a literal key (rejected with topologyMode: "assert"), and a fanout exchange delivers every key to the queue, as does a headers exchange unless topology.bindings declares the queue's bindings (then the declared x-match bindings decide), so EventBus drops events with no matching handler. See the AMQP module page.
RedisAdapter	No	A pattern containing * or > is rejected when the subscription is made, so bus.start() fails with RedisAdapter: wildcard pattern "..." is not supported. Redis Streams requires explicit topic names. Subscribe to exact topic names; see Redis wildcards.

Kafka: topics created after the subscription

Kafka expands a wildcard when the subscription starts, and the adapter refreshes it every five minutes by default (consumerOptions.topicDiscoveryInterval). A discovered topic is read from its first message; each discovery rebalances the consumer group, so consumption pauses for a few seconds. A restart before the next check reads a new topic from its end unless fromBeginning is set. Details: Kafka wildcard subscriptions.

Best Practices

Use default topics for simple cases

If your events have unique message types, the default typeName works well and requires no extra configuration:

// Publisher uses the default topic
await eventBus.publish(OrderCreatedSchema, data);

// Subscriber listens on the default topic automatically
events.service(InventoryEventHandlers, {
  onOrderCreated: async (msg, ctx) => { /* ... */ },
});

Use custom topics for shared message types

Custom topics give events domain-oriented names. Use distinct message types when registering several such topics on one EventBus:

service OrderEventHandlers {
  rpc OnOrderConfirmed(OrderConfirmed) returns (google.protobuf.Empty) {
    option (connectum.events.v1.event).topic = "orders.confirmed";
  }

  rpc OnOrderShipped(OrderShipped) returns (google.protobuf.Empty) {
    option (connectum.events.v1.event).topic = "orders.shipped";
  }
}

The publish lookup maps one message typeName to one topic. If routes or publishes register the same message type under different topics on one bus, start() rejects with Ambiguous publish topic, even if later publishes would pass an explicit topic. To reuse one payload type across different topics, keep those registrations on separate buses. A publisher with no conflicting registrations can choose the destination with PublishOptions.topic.

Keep topic naming consistent

Adopt a convention across your project:

Convention	Example	When to Use
Default (typeName)	orders.v1.OrderCreated	Single message type per event
Domain-based	orders.created, inventory.reserved	Shared messages, simpler names
Hierarchical	domain.aggregate.event	Complex event taxonomies

Related

• Events Overview -- architecture and core concepts
• Getting Started -- step-by-step setup
• Adapters -- wildcard support per adapter
• @connectum/events -- Package Guide