Real-time chat with RabbitMQ

This guide extends the chat pattern from the in-memory guide by replacing the in-process state with two external systems: a persistence layer that loads message history on startup, and RabbitMQ that delivers new messages to all connected server instances in real time.

The observable query and the React component are unchanged from the in-memory version. All the differences are in how the backend fills the room’s history and publishes new messages to it.

The backend is shown in C#, Kotlin, Java, and TypeScript; the RabbitMQ publisher and consumer are plain application code and are shown in C# only. Arc for TypeScript is a source preview: no package is published to npm, and it does not have full parity with Arc on .NET. How the backends differ summarizes the per-language differences.

🎯What you'll build

• A ChatRoom that loads its initial history from a persistence layer
• A SendMessage command that publishes through a publishing contract rather than writing directly to the room
• A RabbitMQ publisher and a consumer that routes each message to the appropriate room (shown for C#)
• The same observable query and React component from the in-memory guide, unchanged

Architecture

SendMessage publishes to RabbitMQ and returns. The consumer, running on every server instance, receives the message from the queue and routes it to the correct ChatRoom. The room publishes its updated history, and every subscriber of the room’s query receives the new message.

Folder structure

Chat/
├── IChatPersistence.cs        ← Persistence abstraction
├── ChatRoom.cs                ← ChatRoom + ChatService (persistence-backed)
├── IChatPublisher.cs          ← Publishing abstraction
├── ChatRoomPage.cs            ← ChatMessage read model + SendMessage command
├── RabbitMqChatPublisher.cs   ← IChatPublisher over RabbitMQ
├── ChatSubscriber.cs          ← BackgroundService — RabbitMQ consumer
└── ChatRoomPage.tsx           ← React component (unchanged from in-memory guide)

This is the C# layout. Kotlin and TypeScript keep the same backend files as .kt or .ts, and Java puts each public type in its own file. The RabbitMQ publisher and consumer are shown for C# only (Step 2).

Step 1 — The backend

The backend keeps the shape of the in-memory guide and changes where messages come from and go to:

• The persistence contract loads a room’s history, oldest first. The implementation depends on your storage choice (SQL, MongoDB, a Chronicle read model — any will do).
• The room starts from that history and exposes Receive, which the broker consumer calls for every message that arrives.
• The chat service is still one instance for the application. It loads a room’s history the first time the room is asked for.
• The publishing contract hides the broker from the command. The message’s Id is created when it is sent and travels in the envelope, so every server instance stores the same message under the same Id.
• The read model and its query are the same as in the in-memory guide, except in TypeScript, where the query awaits the room because its history loads asynchronously.
• The command publishes an envelope instead of writing to the room. The room changes when the message comes back from the broker.

C#

using System.Collections.Concurrent;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using Cratis.Arc.Commands.ModelBound;
using Cratis.Arc.Queries.ModelBound;
using Cratis.Concepts;

namespace MyApp.Chat;

// Chat/IChatPersistence.cs
public interface IChatPersistence
{
    // Loads a room's messages, oldest first.
    Task<IEnumerable<ChatMessage>> GetHistoryAsync(string roomName);
}

// Chat/ChatRoom.cs
public class ChatRoom(IEnumerable<ChatMessage> history)
{
    readonly object _lock = new();
    readonly List<ChatMessage> _history = [.. history];

    public BehaviorSubject<IEnumerable<ChatMessage>> Messages { get; } = new([.. history]);

    // Called for every message that arrives from the broker.
    public void Receive(ChatMessage message)
    {
        lock (_lock)
        {
            _history.Add(message);
            Messages.OnNext([.. _history]);
        }
    }
}

// Register as a singleton: builder.Services.AddSingleton<ChatService>();
public class ChatService(IChatPersistence persistence)
{
    readonly ConcurrentDictionary<string, Lazy<ChatRoom>> _rooms = new();

    // Loads the history once, the first time a room is asked for.
    public ChatRoom GetChatRoom(string name)
    {
        var room = _rooms.GetOrAdd(name, roomName => new Lazy<ChatRoom>(() =>
            new ChatRoom(persistence.GetHistoryAsync(roomName).GetAwaiter().GetResult())));
        try
        {
            return room.Value;
        }
        catch
        {
            // A Lazy keeps a failed load's exception, so forget it and let the next subscriber try again.
            _rooms.TryRemove(new KeyValuePair<string, Lazy<ChatRoom>>(name, room));
            throw;
        }
    }
}

// Chat/IChatPublisher.cs
public interface IChatPublisher
{
    Task Publish(ChatMessageEnvelope envelope);
}

// Chat/ChatRoomPage.cs
public record ChatMessageId(Guid Value) : ConceptAs<Guid>(Value)
{
    public static ChatMessageId New() => new(Guid.NewGuid());
}

// The wire format on the broker: the message plus the room it belongs to.
public record ChatMessageEnvelope(string RoomName, ChatMessageId Id, string User, DateTimeOffset SentAt, string Message);

[ReadModel]
public record ChatMessage(ChatMessageId Id, string User, DateTimeOffset SentAt, string Message)
{
    public static ISubject<IEnumerable<ChatMessage>> ForRoom(string roomName, ChatService chatService)
    {
        var room = chatService.GetChatRoom(roomName);
        // A relay per subscriber, seeded with the room's current history. It follows the room only
        // while Arc is subscribed: Observable.Using ends the room subscription when the client leaves.
        var relay = new BehaviorSubject<IEnumerable<ChatMessage>>(room.Messages.Value);
        return Subject.Create<IEnumerable<ChatMessage>>(relay, Observable.Using(() => room.Messages.Subscribe(relay), _ => relay));
    }
}

[Command]
public record SendMessage(string RoomName, string User, string Message)
{
    // Publishes to the broker; the room updates when the message comes back from it.
    public Task Handle(IChatPublisher publisher) =>
        publisher.Publish(new(RoomName, ChatMessageId.New(), User, DateTimeOffset.UtcNow, Message));
}

View C# snippet source on GitHub

Java

import io.cratis.arc.artifacts.Command;
import io.cratis.arc.artifacts.FromServices;
import io.cratis.arc.artifacts.ReadModel;
import io.cratis.arc.authorization.AllowAnonymous;
import io.cratis.arc.concepts.ConceptAs;
import io.cratis.arc.queries.ObservableState;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Flow;
import org.springframework.stereotype.Component;

// Chat/ChatPersistence.java
public interface ChatPersistence {
    // Loads a room's messages, oldest first.
    List<ChatMessage> history(String roomName);
}

// Chat/ChatRoom.java
public final class ChatRoom {
    private final Object lock = new Object();
    private final ObservableState<List<ChatMessage>> messages;

    public ChatRoom(List<ChatMessage> history) {
        messages = new ObservableState<>(List.copyOf(history));
    }

    public Flow.Publisher<List<ChatMessage>> messages() {
        return messages;
    }

    // Called for every message that arrives from the broker.
    public void receive(ChatMessage message) {
        synchronized (lock) {
            var history = new ArrayList<>(messages.get());
            history.add(message);
            messages.set(List.copyOf(history));
        }
    }
}

// Chat/ChatService.java
// Loads the history once, the first time a room is asked for.
@Component
public final class ChatService {
    private final Map<String, ChatRoom> rooms = new ConcurrentHashMap<>();
    private final ChatPersistence persistence;

    public ChatService(ChatPersistence persistence) {
        this.persistence = persistence;
    }

    public ChatRoom getChatRoom(String name) {
        return rooms.computeIfAbsent(name, key -> new ChatRoom(persistence.history(key)));
    }
}

// Chat/ChatPublisher.java
public interface ChatPublisher {
    void publish(ChatMessageEnvelope envelope);
}

// Chat/ChatMessageId.java
public record ChatMessageId(UUID value) implements ConceptAs<UUID> {
    public static ChatMessageId newId() {
        return new ChatMessageId(UUID.randomUUID());
    }
}

// Chat/ChatMessageEnvelope.java
// The wire format on the broker: the message plus the room it belongs to.
public record ChatMessageEnvelope(
    String roomName, ChatMessageId id, String user, OffsetDateTime sentAt, String message) { }

// Chat/ChatMessage.java
@ReadModel
@AllowAnonymous
public record ChatMessage(ChatMessageId id, String user, OffsetDateTime sentAt, String message) {
    public static Flow.Publisher<List<ChatMessage>> forRoom(String roomName, @FromServices ChatService chatService) {
        return chatService.getChatRoom(roomName).messages();
    }
}

// Chat/SendMessage.java
@Command
@AllowAnonymous
public record SendMessage(String roomName, String user, String message) {
    // Publishes to the broker; the room updates when the message comes back from it.
    public void handle(@FromServices ChatPublisher publisher) {
        publisher.publish(new ChatMessageEnvelope(
            roomName, ChatMessageId.newId(), user, OffsetDateTime.now(ZoneOffset.UTC), message));
    }
}

View Java snippet source on GitHub

Kotlin

import io.cratis.arc.artifacts.Command
import io.cratis.arc.artifacts.FromServices
import io.cratis.arc.artifacts.ReadModel
import io.cratis.arc.authorization.AllowAnonymous
import io.cratis.arc.concepts.ConceptAs
import java.time.OffsetDateTime
import java.time.ZoneOffset
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import org.springframework.stereotype.Component

// Chat/ChatPersistence.kt
interface ChatPersistence {
    // Loads a room's messages, oldest first.
    fun history(roomName: String): List<ChatMessage>
}

// Chat/ChatRoom.kt
class ChatRoom(history: List<ChatMessage>) {
    private val state = MutableStateFlow(history)

    val messages: StateFlow<List<ChatMessage>> = state.asStateFlow()

    // Called for every message that arrives from the broker.
    fun receive(message: ChatMessage) {
        state.update { it + message }
    }
}

// Loads the history once, the first time a room is asked for.
@Component
class ChatService(private val persistence: ChatPersistence) {
    private val rooms = ConcurrentHashMap<String, ChatRoom>()

    fun getChatRoom(name: String): ChatRoom = rooms.computeIfAbsent(name) { ChatRoom(persistence.history(it)) }
}

// Chat/ChatPublisher.kt
interface ChatPublisher {
    fun publish(envelope: ChatMessageEnvelope)
}

// Chat/ChatRoomPage.kt
data class ChatMessageId(private val id: UUID) : ConceptAs<UUID> {
    override fun value(): UUID = id

    companion object {
        fun new(): ChatMessageId = ChatMessageId(UUID.randomUUID())
    }
}

// The wire format on the broker: the message plus the room it belongs to.
data class ChatMessageEnvelope(
    val roomName: String,
    val id: ChatMessageId,
    val user: String,
    val sentAt: OffsetDateTime,
    val message: String
)

@ReadModel
@AllowAnonymous
data class ChatMessage(val id: ChatMessageId, val user: String, val sentAt: OffsetDateTime, val message: String) {
    companion object {
        @JvmStatic
        fun forRoom(roomName: String, @FromServices chatService: ChatService): Flow<List<ChatMessage>> =
            chatService.getChatRoom(roomName).messages
    }
}

@Command
@AllowAnonymous
data class SendMessage(val roomName: String, val user: String, val message: String) {
    // Publishes to the broker; the room updates when the message comes back from it.
    fun handle(@FromServices publisher: ChatPublisher) {
        publisher.publish(ChatMessageEnvelope(roomName, ChatMessageId.new(), user, OffsetDateTime.now(ZoneOffset.UTC), message))
    }
}

View Kotlin snippet source on GitHub

TypeScript (preview)

Chat/ChatPersistence.ts

import type { ChatMessage } from './ChatRoomPage.js';

// An abstract class, so it can be the service token; register your implementation for it.
export abstract class ChatPersistence {
    // Loads a room's messages, oldest first.
    abstract history(roomName: string): Promise<ChatMessage[]>;
}

// Chat/ChatRoom.ts
import { injectable, singleton } from '@cratis/arc.core';
import { BehaviorSubject } from 'rxjs';
import { ChatPersistence } from './ChatPersistence.js';
// Type-only: ChatRoomPage.ts imports ChatService from this file at runtime.
import type { ChatMessage } from './ChatRoomPage.js';

export class ChatRoom {
    readonly messages: BehaviorSubject<ChatMessage[]>;

    constructor(history: ChatMessage[]) {
        this.messages = new BehaviorSubject(history);
    }

    // Called for every message that arrives from the broker.
    receive(message: ChatMessage): void {
        this.messages.next([...this.messages.value, message]);
    }
}

// Loads the history once, the first time a room is asked for.
@singleton()
@injectable(ChatPersistence)
export class ChatService {
    readonly #rooms = new Map<string, Promise<ChatRoom>>();

    constructor(private readonly persistence: ChatPersistence) {}

    getChatRoom(name: string): Promise<ChatRoom> {
        let room = this.#rooms.get(name);
        if (!room) {
            room = this.persistence.history(name).then(history => new ChatRoom(history));
            this.#rooms.set(name, room);
            // Forget a failed load, so the next subscriber tries again.
            room.catch(() => this.#rooms.delete(name));
        }
        return room;
    }
}

// Chat/ChatPublisher.ts
import type { ChatMessageEnvelope } from './ChatRoomPage.js';

export abstract class ChatPublisher {
    abstract publish(envelope: ChatMessageEnvelope): Promise<void>;
}

// Chat/ChatRoomPage.ts
import { ConceptAs, field, Guid } from '@cratis/fundamentals';
import { argument, command, inject, query, readModel, service } from '@cratis/arc.core';
import { BehaviorSubject } from 'rxjs';
import { ChatPublisher } from './ChatPublisher.js';
import { ChatService } from './ChatRoom.js';

export class ChatMessageId extends ConceptAs<Guid> {
    static readonly valueType = Guid;

    static create(): ChatMessageId {
        return new ChatMessageId(Guid.create());
    }
}

// The wire format on the broker: the message plus the room it belongs to.
export interface ChatMessageEnvelope {
    roomName: string;
    id: string;
    user: string;
    sentAt: string;
    message: string;
}

@readModel()
export class ChatMessage {
    @field(ChatMessageId) id: ChatMessageId;
    @field(String) user: string;
    @field(Date) sentAt: Date;
    @field(String) message: string;

    constructor(id = ChatMessageId.create(), user = '', sentAt = new Date(), message = '') {
        this.id = id;
        this.user = user;
        this.sentAt = sentAt;
        this.message = message;
    }

    @query({ observable: true }, argument('roomName', String), service(ChatService))
    static async forRoom(roomName: string, chatService: ChatService): Promise<BehaviorSubject<ChatMessage[]>> {
        return (await chatService.getChatRoom(roomName)).messages;
    }
}

@command()
export class SendMessage {
    @field(String) roomName!: string;
    @field(String) user!: string;
    @field(String) message!: string;

    // Publishes to the broker; the room updates when the message comes back from it.
    @inject(ChatPublisher)
    handle(publisher: ChatPublisher): Promise<void> {
        return publisher.publish({
            roomName: this.roomName,
            id: ChatMessageId.create().value.toString(),
            user: this.user,
            sentAt: new Date().toISOString(),
            message: this.message
        });
    }
}

View TypeScript (preview) snippet source on GitHub

Per backend:

• C# blocks on the first load of a room’s history and loads each room once, even when two requests ask for it at the same time. A Lazy keeps a failed load’s exception, so the service removes a failed room and the next subscriber tries again. For production, consider warming the rooms when the application starts, with a hosted service.
• Kotlin and Java load the history inside computeIfAbsent, so a room is loaded once. The persistence contract is a blocking call, as with JDBC.
• TypeScript loads the history asynchronously, so getChatRoom and the query are async. The service caches the pending load, so concurrent subscribers share one load, and forgets a failed load so the next subscriber tries again. ChatPersistence and ChatPublisher are abstract classes because a TypeScript interface does not exist at runtime to be a service token.

Register your implementations of the two contracts with the application’s services:

• C#: see the registrations after Step 2.
• Kotlin and Java: make each implementation a Spring bean, for example with @Component.
• TypeScript: builder.services.addSingleton(ChatPersistence, YourChatPersistence) and builder.services.addSingleton(ChatPublisher, YourChatPublisher).

Step 2 — RabbitMQ

RabbitMQ is plain application code, not an Arc integration. This step shows it for C# with the RabbitMQ .NET client; with Kotlin, Java or TypeScript, use your platform’s AMQP client in the same roles.

The publisher implements IChatPublisher by publishing the envelope to the chat.messages fanout exchange:

Chat/RabbitMqChatPublisher.cs

using System.Text.Json;
using RabbitMQ.Client;

namespace MyApp.Chat;

public class RabbitMqChatPublisher(IChannel channel) : IChatPublisher
{
    public async Task Publish(ChatMessageEnvelope envelope) =>
        await channel.BasicPublishAsync(
            exchange: "chat.messages",
            routingKey: string.Empty,
            body: JsonSerializer.SerializeToUtf8Bytes(envelope));
}

ChatSubscriber runs for the lifetime of the application. It declares a transient queue bound to the exchange, consumes every published message, and routes it to the correct ChatRoom.

Using a fanout exchange with an exclusive, auto-delete queue means every server instance receives every message. A live chat system needs that, because clients may be connected to any instance.

Chat/ChatSubscriber.cs

using System.Text.Json;
using Microsoft.Extensions.Hosting;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace MyApp.Chat;

/// <summary>
/// Background service that consumes chat messages from RabbitMQ and routes them
/// to the appropriate <see cref="ChatRoom"/>.
/// </summary>
public class ChatSubscriber(
    IConnectionFactory connectionFactory,
    ChatService chatService) : BackgroundService
{
    /// <inheritdoc/>
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using var connection = await connectionFactory.CreateConnectionAsync(stoppingToken);
        await using var channel = await connection.CreateChannelAsync(cancellationToken: stoppingToken);

        // Fanout exchange — every bound queue receives every published message.
        await channel.ExchangeDeclareAsync(
            exchange: "chat.messages",
            type: ExchangeType.Fanout,
            durable: true,
            cancellationToken: stoppingToken);

        // Exclusive, auto-delete queue — scoped to this server instance.
        var queue = await channel.QueueDeclareAsync(
            exclusive: true,
            autoDelete: true,
            cancellationToken: stoppingToken);

        await channel.QueueBindAsync(
            queue: queue.QueueName,
            exchange: "chat.messages",
            routingKey: string.Empty,
            cancellationToken: stoppingToken);

        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            var envelope = JsonSerializer.Deserialize<ChatMessageEnvelope>(ea.Body.Span)!;
            var message = new ChatMessage(envelope.Id, envelope.User, envelope.SentAt, envelope.Message);
            chatService.GetChatRoom(envelope.RoomName).Receive(message);
            await channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken: stoppingToken);
        };

        await channel.BasicConsumeAsync(
            queue: queue.QueueName,
            autoAck: false,
            consumer: consumer,
            cancellationToken: stoppingToken);

        await Task.Delay(Timeout.Infinite, stoppingToken);
    }
}

Register the services in your Program.cs:

builder.Services.AddSingleton<ChatService>();
builder.Services.AddHostedService<ChatSubscriber>();
builder.Services.AddSingleton<IChatPersistence, YourChatPersistenceImplementation>();
builder.Services.AddSingleton<IChatPublisher, RabbitMqChatPublisher>();

// Register IConnectionFactory pointing at your RabbitMQ instance.
builder.Services.AddSingleton<IConnectionFactory>(_ =>
    new ConnectionFactory { HostName = "localhost" });

// Register a singleton IChannel for the publisher.
builder.Services.AddSingleton<IConnection>(sp =>
    sp.GetRequiredService<IConnectionFactory>()
      .CreateConnectionAsync().GetAwaiter().GetResult());
builder.Services.AddSingleton<IChannel>(sp =>
    sp.GetRequiredService<IConnection>()
      .CreateChannelAsync().GetAwaiter().GetResult());

Build the backend to generate the proxies. They are the same ForRoom.ts, SendMessage.ts, and ChatMessage.ts as in the in-memory version, so the frontend does not change.

Step 3 — Delta mode

Delta mode works identically here. The room still publishes the full list on each message, but Arc only sends the difference to each client:

• First connection: the client receives the full persisted history as the initial payload.
• Each new message: Arc matches messages by Id, so the ChangeSet holds one item in added and nothing else.

The use() hook applies each ChangeSet transparently, so messagesResult.data always holds the complete current collection.

See delta mode and the in-memory guide for how the ChangeSet is computed and why ChatMessage has an Id.

Step 4 — The React component

The React component is identical to the one in the in-memory guide. The observable query contract, ForRoom.use({ roomName }) returning messagesResult.data, does not change regardless of how the backend sources its data.

What changed from the in-memory version

Aspect	In-Memory	RabbitMQ
Initial history	Empty []	Loaded from persistence layer
New messages enter via	The room’s Send	The broker consumer, through the room’s Receive
SendMessage handler	Sends to the room through the chat service	Publishes an envelope through the publishing contract
Scales across instances	No, state is process-local	Yes, every instance consumes from the exchange
Observable query	Unchanged	Unchanged
React component	Unchanged	Unchanged

The observable query and the frontend are unaffected by the backend data source. The query returns an observable source of the room’s messages (a subject in C# and TypeScript, a Flow in Kotlin, a Flow.Publisher in Java), and that return type is a contract between the query and Arc, not between the query and any specific storage or messaging technology.

✓Recap

You replaced the in-process state with a persistence layer and a RabbitMQ fanout exchange, and the query and the React component stayed the same. Next, manage the message list in the frontend with useChangeStream().