Explicit Event Store Subscriptions
Chronicle supports explicit subscriptions — manual registration of subscriptions between event stores using the Subscriptions API. When you configure a subscription explicitly, you control:
- Which event types are forwarded
- Subscription lifecycle and management
- The subscription identifier for reference and cleanup
Explicit subscriptions are ideal when:
- You need fine-grained control over which event types are forwarded
- The source service’s event types are not in a shared NuGet package
- You want to dynamically create or remove subscriptions
- Events come from a legacy or external system
Setting Up an Explicit Subscription
Section titled “Setting Up an Explicit Subscription”Use the Subscriptions property on IEventStore to subscribe:
using Cratis.Chronicle;using Cratis.Chronicle.Events;
[EventType]public record SubscriptionsExplicitShipmentDispatched(string OrderId);
public static class SubscriptionsExplicitBasic{ public static Task Run(IEventStore eventStore) => eventStore.Subscriptions.Subscribe( "orders-from-fulfillment", "fulfillment-service", builder => builder.WithEventType<SubscriptionsExplicitShipmentDispatched>());}import io.cratis.chronicle.EventStoreimport io.cratis.chronicle.events.EventType
@EventTypedata class SubscriptionsExplicitPayrollRunCompleted(val employeeId: String, val amount: Double)
suspend fun subscribeToPayroll(store: EventStore) { store.eventStoreSubscriptions.subscribe("payroll-inbox", "PayrollEventStore") { builder -> builder.withEventType(SubscriptionsExplicitPayrollRunCompleted::class) }}import io.cratis.chronicle.EventStore;import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.java.EventStoreSubscriptionBuilderJavaBridge;import io.cratis.chronicle.java.EventStoreSubscriptionsServiceJavaBridge;
@EventTyperecord SubscriptionsExplicitPayrollRunCompleted(String employeeId, double amount) {}
class SubscriptionsExplicitBasic { void subscribeToPayroll(EventStore store) { EventStoreSubscriptionsServiceJavaBridge.subscribe(store.getEventStoreSubscriptions(), "payroll-inbox", "PayrollEventStore", builder -> { EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitPayrollRunCompleted.class); }); }}defmodule MyApp.Events.SubscriptionsExplicitShipmentDispatched do use Chronicle.Events.EventType, id: "subscriptions-explicit-shipment-dispatched"
defstruct [:order_id]end
defmodule MyApp.SubscriptionsExplicitBasic do alias Chronicle.EventStoreSubscriptions.DefinitionBuilder alias MyApp.Events.SubscriptionsExplicitShipmentDispatched
def run do Chronicle.subscribe_to_event_store( "orders-from-fulfillment", "fulfillment-service", fn builder -> DefinitionBuilder.with_event_type(builder, SubscriptionsExplicitShipmentDispatched) end, [] ) endendimport { eventType, IEventStore } from '@cratis/chronicle';
@eventType()class SubscriptionsExplicitShipmentDispatched { constructor(readonly orderId: string) {}}
class SubscriptionsExplicitBasic { static async run(store: IEventStore): Promise<void> { await store.subscriptions.subscribe( 'orders-from-fulfillment', 'fulfillment-service', builder => builder.withEventType(SubscriptionsExplicitShipmentDispatched) ); }}The three arguments are:
| Argument | Description |
|---|---|
| Subscription ID | A stable, unique string that identifies this subscription within the target event store |
| Source event store | The name of the event store whose outbox to subscribe to |
| Configuration callback | Optional — filters which event types to forward |
Subscription ID
Section titled “Subscription ID”The subscription ID must be unique within the target event store. It serves as the persistent identifier for the subscription on the kernel side. If you call Subscribe with the same ID, the kernel treats it as idempotent — no duplicate subscription is created.
A typical naming convention:
using Cratis.Chronicle;using Cratis.Chronicle.Events;
[EventType]public record SubscriptionsExplicitStockAdjusted(string ItemId, int Delta);
public static class SubscriptionsExplicitNamingConvention{ public static async Task Run(IEventStore eventStore) { // subscription-id format: {target}-from-{source} await eventStore.Subscriptions.Subscribe( "orders-from-fulfillment", "fulfillment-service", builder => builder.WithEventType<SubscriptionsExplicitShipmentDispatched>());
await eventStore.Subscriptions.Subscribe( "inventory-from-warehouse", "warehouse-service", builder => builder.WithEventType<SubscriptionsExplicitStockAdjusted>()); }}import io.cratis.chronicle.EventStore
suspend fun subscribeWithStableId(store: EventStore) { // Use a stable, descriptive id — it identifies this subscription across restarts // and is how you target it later with unsubscribe(). store.eventStoreSubscriptions.subscribe("payroll-inbox-v1", "PayrollEventStore") { // No filter configured here — see the filtering example for withEventType. }}import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.java.EventStoreSubscriptionsServiceJavaBridge;
class SubscriptionsExplicitNamingConvention { void subscribeWithStableId(EventStore store) { // Use a stable, descriptive id — it identifies this subscription across restarts // and is how you target it later with unsubscribe(). EventStoreSubscriptionsServiceJavaBridge.subscribe(store.getEventStoreSubscriptions(), "payroll-inbox-v1", "PayrollEventStore", builder -> { }); }}defmodule MyApp.Events.SubscriptionsExplicitStockAdjusted do use Chronicle.Events.EventType, id: "subscriptions-explicit-stock-adjusted"
defstruct [:item_id, :delta]end
defmodule MyApp.SubscriptionsExplicitNamingConvention do alias Chronicle.EventStoreSubscriptions.DefinitionBuilder alias MyApp.Events.{SubscriptionsExplicitShipmentDispatched, SubscriptionsExplicitStockAdjusted}
def run do # subscription-id format: {target}-from-{source} :ok = Chronicle.subscribe_to_event_store( "orders-from-fulfillment", "fulfillment-service", fn builder -> DefinitionBuilder.with_event_type(builder, SubscriptionsExplicitShipmentDispatched) end, [] )
Chronicle.subscribe_to_event_store( "inventory-from-warehouse", "warehouse-service", fn builder -> DefinitionBuilder.with_event_type(builder, SubscriptionsExplicitStockAdjusted) end, [] ) endendimport { eventType, IEventStore } from '@cratis/chronicle';
@eventType()class SubscriptionsExplicitStockAdjusted { constructor( readonly itemId: string, readonly delta: number ) {}}
class SubscriptionsExplicitNamingConvention { static async run(store: IEventStore): Promise<void> { // subscription-id format: {target}-from-{source} await store.subscriptions.subscribe( 'orders-from-fulfillment', 'fulfillment-service', builder => builder.withEventType(SubscriptionsExplicitShipmentDispatched) );
await store.subscriptions.subscribe( 'inventory-from-warehouse', 'warehouse-service', builder => builder.withEventType(SubscriptionsExplicitStockAdjusted) ); }}Filtering Event Types
Section titled “Filtering Event Types”Without a configuration callback, all events from the source outbox are forwarded. Use WithEventType<T>() to limit the subscription to specific types:
using Cratis.Chronicle;using Cratis.Chronicle.Events;
[EventType]public record SubscriptionsExplicitStockReserved(string ItemId, int Quantity);
public static class SubscriptionsExplicitFiltering{ public static Task Run(IEventStore eventStore) => eventStore.Subscriptions.Subscribe( "inventory-updates", "warehouse-service", builder => builder .WithEventType<SubscriptionsExplicitStockAdjusted>() .WithEventType<SubscriptionsExplicitStockReserved>());}import io.cratis.chronicle.EventStoreimport io.cratis.chronicle.events.EventType
@EventTypedata class SubscriptionsExplicitFilteringPayrollRunCompleted(val employeeId: String)
@EventTypedata class SubscriptionsExplicitFilteringPayrollRunFailed(val employeeId: String, val reason: String)
suspend fun subscribeToPayrollOutcomes(store: EventStore) { store.eventStoreSubscriptions.subscribe("payroll-outcomes", "PayrollEventStore") { builder -> builder .withEventType(SubscriptionsExplicitFilteringPayrollRunCompleted::class) .withEventType(SubscriptionsExplicitFilteringPayrollRunFailed::class) }}import io.cratis.chronicle.EventStore;import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.java.EventStoreSubscriptionBuilderJavaBridge;import io.cratis.chronicle.java.EventStoreSubscriptionsServiceJavaBridge;
@EventTyperecord SubscriptionsExplicitFilteringPayrollRunCompleted(String employeeId) {}
@EventTyperecord SubscriptionsExplicitFilteringPayrollRunFailed(String employeeId, String reason) {}
class SubscriptionsExplicitFiltering { void subscribeToPayrollOutcomes(EventStore store) { EventStoreSubscriptionsServiceJavaBridge.subscribe(store.getEventStoreSubscriptions(), "payroll-outcomes", "PayrollEventStore", builder -> { EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitFilteringPayrollRunCompleted.class); EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitFilteringPayrollRunFailed.class); }); }}defmodule MyApp.Events.SubscriptionsExplicitStockReserved do use Chronicle.Events.EventType, id: "subscriptions-explicit-stock-reserved"
defstruct [:item_id, :quantity]end
defmodule MyApp.SubscriptionsExplicitFiltering do alias Chronicle.EventStoreSubscriptions.DefinitionBuilder alias MyApp.Events.{SubscriptionsExplicitStockAdjusted, SubscriptionsExplicitStockReserved}
def run do Chronicle.subscribe_to_event_store( "inventory-updates", "warehouse-service", fn builder -> builder |> DefinitionBuilder.with_event_type(SubscriptionsExplicitStockAdjusted) |> DefinitionBuilder.with_event_type(SubscriptionsExplicitStockReserved) end, [] ) endendimport { eventType, IEventStore } from '@cratis/chronicle';
@eventType()class SubscriptionsExplicitStockReserved { constructor( readonly itemId: string, readonly quantity: number ) {}}
class SubscriptionsExplicitFiltering { static async run(store: IEventStore): Promise<void> { await store.subscriptions.subscribe( 'inventory-updates', 'warehouse-service', builder => builder .withEventType(SubscriptionsExplicitStockAdjusted) .withEventType(SubscriptionsExplicitStockReserved) ); }}If you need to subscribe to all events without filtering, omit the configuration callback entirely:
using Cratis.Chronicle;
public static class SubscriptionsExplicitNoFilter{ public static Task Run(IEventStore eventStore) => // All events from fulfillment-service outbox will be forwarded eventStore.Subscriptions.Subscribe( "all-fulfillment-events", "fulfillment-service");}import io.cratis.chronicle.EventStore
suspend fun subscribeToEverything(store: EventStore) { // No withEventType calls — every event type from the source outbox is subscribed to. store.eventStoreSubscriptions.subscribe("payroll-firehose", "PayrollEventStore") { // Intentionally left unconfigured. }}import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.java.EventStoreSubscriptionsServiceJavaBridge;
class SubscriptionsExplicitNoFilter { void subscribeToEverything(EventStore store) { // No withEventType calls — every event type from the source outbox is subscribed to. EventStoreSubscriptionsServiceJavaBridge.subscribe(store.getEventStoreSubscriptions(), "payroll-firehose", "PayrollEventStore", builder -> { }); }}defmodule MyApp.SubscriptionsExplicitNoFilter do def run do # All events from fulfillment-service outbox will be forwarded Chronicle.subscribe_to_event_store("all-fulfillment-events", "fulfillment-service", []) endendimport { IEventStore } from '@cratis/chronicle';
class SubscriptionsExplicitNoFilter { static async run(store: IEventStore): Promise<void> { // All events from fulfillment-service outbox will be forwarded await store.subscriptions.subscribe('all-fulfillment-events', 'fulfillment-service'); }}Accessing Forwarded Events
Section titled “Accessing Forwarded Events”Events forwarded via an explicit subscription are placed into the inbox sequence for the source event store. Access inbox events through any observer (reactor, projection, reducer):
using Cratis.Chronicle.Events;using Cratis.Chronicle.Reactors;
[EventType]public record SubscriptionsExplicitOrderPlaced(Guid OrderId, decimal Amount);
public class SubscriptionsExplicitIncomingOrdersReactor : IReactor{ public Task OrderPlaced(SubscriptionsExplicitOrderPlaced @event, EventContext context) => HandleIncomingOrderAsync(@event.OrderId, @event.Amount);
Task HandleIncomingOrderAsync(Guid id, decimal amount) => Task.CompletedTask;}import io.cratis.chronicle.events.EventContextimport io.cratis.chronicle.events.EventTypeimport io.cratis.chronicle.observation.Reactor
@EventTypedata class SubscriptionsExplicitOrderPlaced(val orderId: String, val amount: Double)
@Reactorclass SubscriptionsExplicitIncomingOrdersReactor { fun orderPlaced(event: SubscriptionsExplicitOrderPlaced, context: EventContext) { handleIncomingOrder(event.orderId, event.amount) }
private fun handleIncomingOrder(id: String, amount: Double) {}}import io.cratis.chronicle.events.EventContext;import io.cratis.chronicle.events.EventType;import io.cratis.chronicle.observation.Reactor;
@EventTyperecord SubscriptionsExplicitOrderPlaced(String orderId, double amount) {}
@Reactorclass SubscriptionsExplicitIncomingOrdersReactor { void orderPlaced(SubscriptionsExplicitOrderPlaced event, EventContext context) { handleIncomingOrder(event.orderId(), event.amount()); }
private void handleIncomingOrder(String id, double amount) {}}defmodule MyApp.Events.SubscriptionsExplicitOrderPlaced do use Chronicle.Events.EventType, id: "subscriptions-explicit-order-placed"
defstruct [:order_id, :amount]end
defmodule MyApp.Reactors.SubscriptionsExplicitIncomingOrdersReactor do use Chronicle.Reactors.Reactor
alias MyApp.Events.SubscriptionsExplicitOrderPlaced
@handles SubscriptionsExplicitOrderPlaced
@impl true def handle(%SubscriptionsExplicitOrderPlaced{}, _context), do: :okendimport { eventType, reactor } from '@cratis/chronicle';
@eventType()class SubscriptionsExplicitOrderPlaced { constructor( readonly orderId: string, readonly amount: number ) {}}
@reactor()class SubscriptionsExplicitIncomingOrdersReactor { // Method name must be the exact camelCase of the event's class name - // Chronicle discovers handlers by name, not by parameter type. async subscriptionsExplicitOrderPlaced(event: SubscriptionsExplicitOrderPlaced): Promise<void> { await this.handleIncomingOrder(event.orderId, event.amount); }
private async handleIncomingOrder(orderId: string, amount: number): Promise<void> {}}The kernel automatically routes incoming events to the appropriate inbox-{sourceEventStore} event sequence. You do not need to reference the subscription ID when observing the events.
Removing a Subscription
Section titled “Removing a Subscription”using Cratis.Chronicle;
public static class SubscriptionsExplicitUnsubscribe{ public static Task Run(IEventStore eventStore) => eventStore.Subscriptions.Unsubscribe("orders-from-fulfillment");}import io.cratis.chronicle.EventStore
suspend fun unsubscribeFromPayroll(store: EventStore) { store.eventStoreSubscriptions.unsubscribe("payroll-inbox")}import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.java.EventStoreSubscriptionsServiceJavaBridge;
class SubscriptionsExplicitUnsubscribe { void unsubscribeFromPayroll(EventStore store) { EventStoreSubscriptionsServiceJavaBridge.unsubscribe(store.getEventStoreSubscriptions(), "payroll-inbox"); }}defmodule MyApp.SubscriptionsExplicitUnsubscribe do def run do Chronicle.unsubscribe_from_event_store("orders-from-fulfillment") endendimport { IEventStore } from '@cratis/chronicle';
class SubscriptionsExplicitUnsubscribe { static async run(store: IEventStore): Promise<void> { await store.subscriptions.unsubscribe('orders-from-fulfillment'); }}Calling Unsubscribe with the subscription ID:
- Stops event forwarding from the source outbox
- Removes the persistent subscription definition from the kernel
- Existing inbox events are not deleted — they remain available for observation
Once removed, no new events are forwarded to the inbox. If you later recreate the subscription with the same ID, forwarding resumes from that point forward.
Idempotent Registration
Section titled “Idempotent Registration”Subscriptions are identified by their subscription ID. Calling Subscribe with the same ID more than once is safe — if the subscription already exists, no duplicate subscription is created.
This makes it safe to register all subscriptions at application startup without checking if they already exist:
using Cratis.Chronicle;using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.Hosting;
public static class SubscriptionsExplicitStartupRegistration{ public static async Task Configure(string[] args) { var builder = Host.CreateApplicationBuilder(args); builder.AddCratisChronicle(options => options.EventStore = "Quickstart");
// Safe to call on every application startup var app = builder.Build();
var eventStore = app.Services.GetRequiredService<IEventStore>();
await eventStore.Subscriptions.Subscribe( "orders-from-fulfillment", "fulfillment-service", builder => builder.WithEventType<SubscriptionsExplicitShipmentDispatched>());
await eventStore.Subscriptions.Subscribe( "inventory-from-warehouse", "warehouse-service", builder => builder.WithEventType<SubscriptionsExplicitStockAdjusted>());
await app.RunAsync(); }}import io.cratis.chronicle.ChronicleClientimport io.cratis.chronicle.ChronicleOptions
suspend fun configureSubscriptionsAtStartup() { val client = ChronicleClient(ChronicleOptions.development()) val eventStore = client.getEventStore("Quickstart")
// Safe to call on every application startup - Subscribe is idempotent by subscription id eventStore.eventStoreSubscriptions.subscribe("orders-from-fulfillment", "fulfillment-service") { builder -> builder.withEventType(SubscriptionsExplicitShipmentDispatched::class) }
eventStore.eventStoreSubscriptions.subscribe("inventory-from-warehouse", "warehouse-service") { builder -> builder.withEventType(SubscriptionsExplicitStockAdjusted::class) }}import io.cratis.chronicle.ChronicleClient;import io.cratis.chronicle.ChronicleOptions;import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.java.EventStoreSubscriptionBuilderJavaBridge;import io.cratis.chronicle.java.EventStoreSubscriptionsServiceJavaBridge;
class SubscriptionsExplicitStartupRegistration { static void configure() { ChronicleClient client = new ChronicleClient(ChronicleOptions.Companion.development()); EventStore eventStore = (EventStore) client.getEventStore("Quickstart", "Default");
// Safe to call on every application startup - subscribe is idempotent by subscription id EventStoreSubscriptionsServiceJavaBridge.subscribe(eventStore.getEventStoreSubscriptions(), "orders-from-fulfillment", "fulfillment-service", builder -> { EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitShipmentDispatched.class); });
EventStoreSubscriptionsServiceJavaBridge.subscribe(eventStore.getEventStoreSubscriptions(), "inventory-from-warehouse", "warehouse-service", builder -> { EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitStockAdjusted.class); }); }}defmodule MyApp.Events.SubscriptionsStartupShipmentDispatched do use Chronicle.Events.EventType, id: "subscriptions-startup-shipment-dispatched"
defstruct [:order_id]end
defmodule MyApp.Events.SubscriptionsStartupStockAdjusted do use Chronicle.Events.EventType, id: "subscriptions-startup-stock-adjusted"
defstruct [:item_id, :delta]end
defmodule MyApp.SubscriptionsStartupApplication do use Application
alias Chronicle.EventStoreSubscriptions.DefinitionBuilder alias MyApp.Events.{SubscriptionsStartupShipmentDispatched, SubscriptionsStartupStockAdjusted}
@impl true def start(_type, _args) do children = [ {Chronicle.Client, connection_string: "chronicle://localhost:35000", event_store: "quickstart"} ]
supervisor = Supervisor.start_link(children, strategy: :one_for_one)
# Safe to call on every application startup — Chronicle treats a repeated # subscription id as idempotent and creates no duplicate. :ok = Chronicle.subscribe_to_event_store( "orders-from-fulfillment", "fulfillment-service", fn builder -> DefinitionBuilder.with_event_type(builder, SubscriptionsStartupShipmentDispatched) end, [] )
:ok = Chronicle.subscribe_to_event_store( "inventory-from-warehouse", "warehouse-service", fn builder -> DefinitionBuilder.with_event_type(builder, SubscriptionsStartupStockAdjusted) end, [] )
supervisor endendimport { ChronicleClient, ChronicleOptions, eventType, IEventStore } from '@cratis/chronicle';
@eventType()class SubscriptionsExplicitStartupShipmentDispatched { constructor(readonly orderId: string, readonly trackingNumber: string) {}}
@eventType()class SubscriptionsExplicitStartupStockAdjusted { constructor(readonly sku: string, readonly delta: number) {}}
// Safe to call on every application startupasync function runSubscriptionsExplicitStartupRegistration(): Promise<void> { const client = new ChronicleClient(ChronicleOptions.fromConnectionString('chronicle://localhost:35000')); const eventStore: IEventStore = await client.getEventStore('Quickstart');
await eventStore.subscriptions.subscribe( 'orders-from-fulfillment', 'fulfillment-service', builder => builder.withEventType(SubscriptionsExplicitStartupShipmentDispatched));
await eventStore.subscriptions.subscribe( 'inventory-from-warehouse', 'warehouse-service', builder => builder.withEventType(SubscriptionsExplicitStartupStockAdjusted));}If any of these subscriptions already exist on the kernel, they are left unchanged.
Subscription Lifecycle
Section titled “Subscription Lifecycle”Explicit subscriptions are kernel-managed and persistent:
- Created when you call
Subscribe, stored on the kernel, and assigned a subscription ID - Persisted — survive client disconnections and kernel restarts
- Re-established automatically when the kernel starts, even if the client is not connected
- Removed only when you explicitly call
Unsubscribe
Typical Setup Pattern
Section titled “Typical Setup Pattern”A common pattern is to register all subscriptions once during application startup:
using Cratis.Chronicle;using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.Hosting;
public static class SubscriptionsExplicitTypicalPattern{ public static async Task RegisterSubscriptions(IEventStore eventStore) { await eventStore.Subscriptions.Subscribe( "orders-from-fulfillment", "fulfillment-service", builder => builder.WithEventType<SubscriptionsExplicitShipmentDispatched>());
await eventStore.Subscriptions.Subscribe( "inventory-updates", "warehouse-service", builder => builder .WithEventType<SubscriptionsExplicitStockAdjusted>() .WithEventType<SubscriptionsExplicitStockReserved>()); }
public static async Task Configure(string[] args) { var hostBuilder = Host.CreateApplicationBuilder(args); hostBuilder.AddCratisChronicle(options => options.EventStore = "Quickstart");
var app = hostBuilder.Build(); var eventStore = app.Services.GetRequiredService<IEventStore>(); await RegisterSubscriptions(eventStore); await app.RunAsync(); }}import io.cratis.chronicle.ChronicleClientimport io.cratis.chronicle.ChronicleOptionsimport io.cratis.chronicle.IEventStoreimport io.cratis.chronicle.events.EventType
@EventTypedata class SubscriptionsExplicitShipmentDispatched(val shipmentId: String = "")
@EventTypedata class SubscriptionsExplicitStockAdjusted(val sku: String = "", val delta: Int = 0)
@EventTypedata class SubscriptionsExplicitStockReserved(val sku: String = "", val quantity: Int = 0)
suspend fun registerSubscriptions(eventStore: IEventStore) { eventStore.eventStoreSubscriptions.subscribe("orders-from-fulfillment", "fulfillment-service") { builder -> builder.withEventType(SubscriptionsExplicitShipmentDispatched::class) }
eventStore.eventStoreSubscriptions.subscribe("inventory-updates", "warehouse-service") { builder -> builder .withEventType(SubscriptionsExplicitStockAdjusted::class) .withEventType(SubscriptionsExplicitStockReserved::class) }}
suspend fun configureSubscriptionsTypicalPattern() { val client = ChronicleClient(ChronicleOptions.development()) val eventStore = client.getEventStore("Quickstart") registerSubscriptions(eventStore)}import io.cratis.chronicle.ChronicleClient;import io.cratis.chronicle.ChronicleOptions;import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.events.EventType;import io.cratis.chronicle.java.EventStoreSubscriptionBuilderJavaBridge;import io.cratis.chronicle.java.EventStoreSubscriptionsServiceJavaBridge;
@EventTyperecord SubscriptionsExplicitShipmentDispatched(String shipmentId) {}
@EventTyperecord SubscriptionsExplicitStockAdjusted(String sku, int delta) {}
@EventTyperecord SubscriptionsExplicitStockReserved(String sku, int quantity) {}
class SubscriptionsExplicitTypicalPattern { static void registerSubscriptions(EventStore eventStore) { EventStoreSubscriptionsServiceJavaBridge.subscribe(eventStore.getEventStoreSubscriptions(), "orders-from-fulfillment", "fulfillment-service", builder -> { EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitShipmentDispatched.class); });
EventStoreSubscriptionsServiceJavaBridge.subscribe(eventStore.getEventStoreSubscriptions(), "inventory-updates", "warehouse-service", builder -> { EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitStockAdjusted.class); EventStoreSubscriptionBuilderJavaBridge.withEventType(builder, SubscriptionsExplicitStockReserved.class); }); }
static void configure() { ChronicleClient client = new ChronicleClient(ChronicleOptions.Companion.development()); EventStore eventStore = (EventStore) client.getEventStore("Quickstart", "Default"); registerSubscriptions(eventStore); }}defmodule MyApp.Events.SubscriptionsTypicalShipmentDispatched do use Chronicle.Events.EventType, id: "subscriptions-typical-shipment-dispatched"
defstruct [:order_id]end
defmodule MyApp.Events.SubscriptionsTypicalStockAdjusted do use Chronicle.Events.EventType, id: "subscriptions-typical-stock-adjusted"
defstruct [:item_id, :delta]end
defmodule MyApp.Events.SubscriptionsTypicalStockReserved do use Chronicle.Events.EventType, id: "subscriptions-typical-stock-reserved"
defstruct [:item_id, :quantity]end
defmodule MyApp.SubscriptionsTypicalRegistration do @moduledoc false
alias Chronicle.EventStoreSubscriptions.DefinitionBuilder
alias MyApp.Events.{ SubscriptionsTypicalShipmentDispatched, SubscriptionsTypicalStockAdjusted, SubscriptionsTypicalStockReserved }
def register_subscriptions do :ok = Chronicle.subscribe_to_event_store( "orders-from-fulfillment", "fulfillment-service", fn builder -> DefinitionBuilder.with_event_type(builder, SubscriptionsTypicalShipmentDispatched) end, [] )
:ok = Chronicle.subscribe_to_event_store( "inventory-updates", "warehouse-service", fn builder -> builder |> DefinitionBuilder.with_event_type(SubscriptionsTypicalStockAdjusted) |> DefinitionBuilder.with_event_type(SubscriptionsTypicalStockReserved) end, [] ) endendimport { eventType, IEventStore } from '@cratis/chronicle';
@eventType()class SubscriptionsExplicitTypicalShipmentDispatched { constructor(readonly orderId: string, readonly trackingNumber: string) {}}
@eventType()class SubscriptionsExplicitTypicalStockAdjusted { constructor(readonly sku: string, readonly delta: number) {}}
@eventType()class SubscriptionsExplicitTypicalStockReserved { constructor(readonly sku: string, readonly quantity: number) {}}
async function registerSubscriptionsExplicitTypicalPattern(eventStore: IEventStore): Promise<void> { await eventStore.subscriptions.subscribe( 'orders-from-fulfillment', 'fulfillment-service', builder => builder.withEventType(SubscriptionsExplicitTypicalShipmentDispatched));
await eventStore.subscriptions.subscribe( 'inventory-updates', 'warehouse-service', builder => builder .withEventType(SubscriptionsExplicitTypicalStockAdjusted) .withEventType(SubscriptionsExplicitTypicalStockReserved));}When to Use Explicit Subscriptions
Section titled “When to Use Explicit Subscriptions”Use explicit subscriptions when:
- You need fine-grained control over which event types are forwarded
- Event types are not published in a shared NuGet package
- You need to dynamically create or remove subscriptions at runtime
- The subscription relationship is consuming-service-specific
Use implicit subscriptions when:
- Event types are published in a shared NuGet package with
[EventStore]attributes - All events from a source should be automatically forwarded
- You prefer zero configuration and automatic routing
See Also
Section titled “See Also”- Implicit Event Store Subscriptions — automatic subscription via
[EventStore]attributes - Outbox and Inbox — conceptual explanation of how events flow between stores
- Reactors — processing inbox events with reactors
- Projections — building read models from inbox events