Architecture

An e-commerce order service publishes status updates. Multiple downstream services — email, push notifications, and analytics — each subscribe independently and process events in real time.

Each event fans out to every ungrouped subscriber; the analytics workers share one consumer group, so each event reaches exactly one of them.

Implementation

Order Status Publisher

When an order changes status, publish an event with the order details and status metadata.

order_status_publisher.go

package main

import (
    "context"
    "encoding/json"
    "log"
    "time"

    "github.com/kubemq-io/kubemq-go/v2"
)

type OrderEvent struct {
    OrderID   string  `json:"orderId"`
    Status    string  `json:"status"`
    Customer  string  `json:"customer"`
    Amount    float64 `json:"amount"`
    Timestamp int64   `json:"timestamp"`
}

func publishOrderStatus(ctx context.Context, client *kubemq.Client, event OrderEvent) error {
    body, _ := json.Marshal(event)
    return client.SendEvent(ctx, kubemq.NewEvent().
        SetChannel("order-notifications").
        SetMetadata("order."+event.Status).
        SetBody(body).
        SetTags(map[string]string{
            "status":   event.Status,
            "customer": event.Customer,
        }),
    )
}

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    events := []OrderEvent{
        {"ORD-001", "created", "alice@example.com", 99.99, time.Now().UnixMilli()},
        {"ORD-001", "confirmed", "alice@example.com", 99.99, time.Now().UnixMilli()},
        {"ORD-001", "shipped", "alice@example.com", 99.99, time.Now().UnixMilli()},
    }

    for _, event := range events {
        if err := publishOrderStatus(ctx, client, event); err != nil {
            log.Printf("Failed to publish %s: %v", event.OrderID, err)
        }
        time.Sleep(time.Second)
    }
}

order_status_publisher.py

import json
import time
from kubemq.pubsub import Client as PubSubClient
from kubemq.pubsub import EventMessage

client = PubSubClient(address="localhost:50000")

events = [
    {"orderId": "ORD-001", "status": "created", "customer": "alice@example.com", "amount": 99.99},
    {"orderId": "ORD-001", "status": "confirmed", "customer": "alice@example.com", "amount": 99.99},
    {"orderId": "ORD-001", "status": "shipped", "customer": "alice@example.com", "amount": 99.99},
]

for event in events:
    client.send_event(
        EventMessage(
            channel="order-notifications",
            metadata=f"order.{event['status']}",
            body=json.dumps(event).encode("utf-8"),
            tags={"status": event["status"], "customer": event["customer"]},
        )
    )
    print(f"Published: {event['orderId']} -> {event['status']}")
    time.sleep(1)

client.close()

order_status_publisher.js

const { KubeMQClient } = require("kubemq-js");

const client = new KubeMQClient({ address: "localhost:50000" });

const events = [
  { orderId: "ORD-001", status: "created", customer: "alice@example.com", amount: 99.99 },
  { orderId: "ORD-001", status: "confirmed", customer: "alice@example.com", amount: 99.99 },
  { orderId: "ORD-001", status: "shipped", customer: "alice@example.com", amount: 99.99 },
];

for (const event of events) {
  await client.sendEvent({
    channel: "order-notifications",
    metadata: `order.${event.status}`,
    body: Buffer.from(JSON.stringify(event)),
    tags: { status: event.status, customer: event.customer },
  });
  console.log(`Published: ${event.orderId} -> ${event.status}`);
  await new Promise((r) => setTimeout(r, 1000));
}

OrderStatusPublisher.java

PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("order-service")
    .build();

String[][] events = {
    {"ORD-001", "created", "alice@example.com", "99.99"},
    {"ORD-001", "confirmed", "alice@example.com", "99.99"},
    {"ORD-001", "shipped", "alice@example.com", "99.99"},
};

for (String[] event : events) {
    String body = String.format(
        "{\"orderId\":\"%s\",\"status\":\"%s\",\"customer\":\"%s\",\"amount\":%s}",
        event[0], event[1], event[2], event[3]);

    client.sendEventsMessage(EventMessage.builder()
        .channel("order-notifications")
        .metadata("order." + event[1])
        .body(body.getBytes())
        .tags(Map.of("status", event[1], "customer", event[2]))
        .build());

    System.out.printf("Published: %s -> %s%n", event[0], event[1]);
    Thread.sleep(1000);
}
client.close();

OrderStatusPublisher.cs

await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

var events = new[]
{
    new { OrderId = "ORD-001", Status = "created", Customer = "alice@example.com", Amount = 99.99 },
    new { OrderId = "ORD-001", Status = "confirmed", Customer = "alice@example.com", Amount = 99.99 },
    new { OrderId = "ORD-001", Status = "shipped", Customer = "alice@example.com", Amount = 99.99 },
};

foreach (var evt in events)
{
    await client.SendEventAsync(new EventMessage
    {
        Channel = "order-notifications",
        Metadata = $"order.{evt.Status}",
        Body = Encoding.UTF8.GetBytes(
            $"{{\"orderId\":\"{evt.OrderId}\",\"status\":\"{evt.Status}\",\"amount\":{evt.Amount}}}"),
        Tags = new Dictionary<string, string>
        {
            ["status"] = evt.Status, ["customer"] = evt.Customer
        }
    });
    Console.WriteLine($"Published: {evt.OrderId} -> {evt.Status}");
    await Task.Delay(1000);
}

OrderStatusPublisher.kt

val client = PubSubClient("localhost:50000")

data class OrderEvent(val orderId: String, val status: String, val customer: String, val amount: Double)

val events = listOf(
    OrderEvent("ORD-001", "created", "alice@example.com", 99.99),
    OrderEvent("ORD-001", "confirmed", "alice@example.com", 99.99),
    OrderEvent("ORD-001", "shipped", "alice@example.com", 99.99),
)

for (event in events) {
    val body = """{"orderId":"${event.orderId}","status":"${event.status}","amount":${event.amount}}"""
    client.sendEvent(EventMessage(
        channel = "order-notifications",
        metadata = "order.${event.status}",
        body = body.toByteArray(),
        tags = mapOf("status" to event.status, "customer" to event.customer),
    ))
    println("Published: ${event.orderId} -> ${event.status}")
    Thread.sleep(1000)
}
client.close()

order_status_publisher.cpp

auto client = kubemq::PubSubClient("localhost:50000");

struct OrderEvent { std::string id, status, customer; double amount; };
std::vector<OrderEvent> events = {
    {"ORD-001", "created", "alice@example.com", 99.99},
    {"ORD-001", "confirmed", "alice@example.com", 99.99},
    {"ORD-001", "shipped", "alice@example.com", 99.99},
};

for (const auto& evt : events) {
    kubemq::EventMessage event;
    event.channel = "order-notifications";
    event.metadata = "order." + evt.status;
    event.body = "{\"orderId\":\"" + evt.id + "\",\"status\":\"" + evt.status + "\"}";
    event.tags["status"] = evt.status;
    event.tags["customer"] = evt.customer;

    client.sendEvent(event);
    std::cout << "Published: " << evt.id << " -> " << evt.status << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(1));
}

order_status_publisher.rs

use kubemq::prelude::*;
use kubemq::EventBuilder;
use std::time::Duration;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let statuses = ["created", "confirmed", "shipped"];

    for status in statuses {
        let body = format!(
            "{{\"orderId\":\"ORD-001\",\"status\":\"{}\",\"amount\":99.99}}",
            status
        );
        let event = EventBuilder::new()
            .channel("order-notifications")
            .metadata(format!("order.{}", status))
            .body(body.into_bytes())
            .add_tag("status", status)
            .add_tag("customer", "alice@example.com")
            .build();

        client.send_event(event).await?;
        println!("Published: ORD-001 -> {}", status);
        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    client.close().await?;
    Ok(())
}

order_status_publisher.rb

require 'kubemq'

client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'order-service')

statuses = %w[created confirmed shipped]

statuses.each do |status|
  body = %({"orderId":"ORD-001","status":"#{status}","amount":99.99})
  msg = KubeMQ::PubSub::EventMessage.new(
    channel: 'order-notifications',
    metadata: "order.#{status}",
    body: body,
    tags: { 'status' => status, 'customer' => 'alice@example.com' }
  )
  client.send_event(msg)
  puts "Published: ORD-001 -> #{status}"
  sleep 1
end

client.close

order_status_publisher.exs

{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-service")

for status <- ["created", "confirmed", "shipped"] do
  body = ~s({"orderId":"ORD-001","status":"#{status}","amount":99.99})

  event =
    KubeMQ.Event.new(
      channel: "order-notifications",
      metadata: "order.#{status}",
      body: body,
      tags: %{"status" => status, "customer" => "alice@example.com"}
    )

  :ok = KubeMQ.Client.send_event(client, event)
  IO.puts("Published: ORD-001 -> #{status}")
  Process.sleep(1_000)
end

KubeMQ.Client.close(client)

Email Notification Subscriber

The email service receives all events and sends confirmation emails for specific status changes.

email_service.go

sub, err := client.SubscribeToEvents(ctx, "order-notifications", "",
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        if event.Tags["status"] == "shipped" {
            fmt.Printf("[Email] Sending shipping confirmation to %s for order %s\n",
                event.Tags["customer"], string(event.Body))
        }
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("[Email] Error:", err)
    }),
)

email_service.py

def on_event(event):
    if event.tags.get("status") == "shipped":
        print(f"[Email] Sending shipping confirmation to {event.tags['customer']}")

client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="order-notifications",
        on_receive_event_callback=on_event,
        on_error_callback=lambda e: print(f"[Email] Error: {e}"),
    ),
    cancel=CancellationToken(),
)

email_service.js

client.subscribeToEvents({
  channel: "order-notifications",
  onEvent: (msg) => {
    if (msg.tags?.status === "shipped") {
      console.log(`[Email] Sending shipping confirmation to ${msg.tags.customer}`);
    }
  },
  onError: (err) => console.error("[Email] Error:", err.message),
});

EmailService.java

client.subscribeToEvents(EventsSubscription.builder()
    .channel("order-notifications")
    .onReceiveEventCallback(event -> {
        if ("shipped".equals(event.getTags().get("status"))) {
            System.out.printf("[Email] Sending shipping confirmation to %s%n",
                event.getTags().get("customer"));
        }
    })
    .onErrorCallback(err ->
        System.err.println("[Email] Error: " + err.getMessage()))
    .build());

EmailService.cs

await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "order-notifications" }))
{
    if (msg.Tags?.GetValueOrDefault("status") == "shipped")
    {
        Console.WriteLine($"[Email] Sending shipping confirmation to "
            + $"{msg.Tags["customer"]}");
    }
}

EmailService.kt

client.subscribeToEvents(
    channel = "order-notifications",
    onEvent = { event ->
        if (event.tags["status"] == "shipped") {
            println("[Email] Sending shipping confirmation to ${event.tags["customer"]}")
        }
    },
    onError = { err -> System.err.println("[Email] Error: ${err.message}") }
)

email_service.cpp

client.subscribeToEvents("order-notifications", "",
    [](const kubemq::Event& event) {
        if (event.tags.at("status") == "shipped") {
            std::cout << "[Email] Sending shipping confirmation to "
                      << event.tags.at("customer") << std::endl;
        }
    },
    [](const std::string& err) {
        std::cerr << "[Email] Error: " << err << std::endl;
    }
);

email_service.rs

// Ungrouped subscriber: receives every event on the channel.
let sub = client
    .subscribe_to_events(
        "order-notifications",
        "",
        |event| {
            Box::pin(async move {
                if event.tags.get("status").map(String::as_str) == Some("shipped") {
                    let customer = event.tags.get("customer").cloned().unwrap_or_default();
                    println!("[Email] Sending shipping confirmation to {}", customer);
                }
            })
        },
        None,
    )
    .await?;

email_service.rb

cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsSubscription.new(channel: 'order-notifications')
client.subscribe_to_events(sub, cancellation_token: cancel,
                                on_error: ->(e) { puts "[Email] Error: #{e.message}" }) do |event|
  if event.tags['status'] == 'shipped'
    puts "[Email] Sending shipping confirmation to #{event.tags['customer']}"
  end
end

email_service.exs

# Ungrouped subscriber: receives every event on the channel.
{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, "order-notifications",
    on_event: fn event ->
      if event.tags["status"] == "shipped" do
        IO.puts("[Email] Sending shipping confirmation to #{event.tags["customer"]}")
      end
    end
  )

Analytics Subscriber (with Consumer Group)

Analytics workers use a consumer group for load-balanced processing.

analytics_worker.go

sub, err := client.SubscribeToEvents(ctx, "order-notifications", "analytics",
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[Analytics] Recording metric: %s\n", event.Metadata)
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("[Analytics] Error:", err)
    }),
)

analytics_worker.py

client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="order-notifications",
        group="analytics",
        on_receive_event_callback=lambda e: print(
            f"[Analytics] Recording metric: {e.metadata}"
        ),
        on_error_callback=lambda e: print(f"[Analytics] Error: {e}"),
    ),
    cancel=CancellationToken(),
)

analytics_worker.js

client.subscribeToEvents({
  channel: "order-notifications",
  group: "analytics",
  onEvent: (msg) =>
    console.log(`[Analytics] Recording metric: ${msg.metadata}`),
  onError: (err) => console.error("[Analytics] Error:", err.message),
});

AnalyticsWorker.java

client.subscribeToEvents(EventsSubscription.builder()
    .channel("order-notifications")
    .group("analytics")
    .onReceiveEventCallback(event ->
        System.out.println("[Analytics] Recording metric: " + event.getMetadata()))
    .onErrorCallback(err ->
        System.err.println("[Analytics] Error: " + err.getMessage()))
    .build());

AnalyticsWorker.cs

await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "order-notifications", Group = "analytics" }))
{
    Console.WriteLine($"[Analytics] Recording metric: {msg.Metadata}");
}

AnalyticsWorker.kt

client.subscribeToEvents(
    channel = "order-notifications",
    group = "analytics",
    onEvent = { event ->
        println("[Analytics] Recording metric: ${event.metadata}")
    },
    onError = { err -> System.err.println("[Analytics] Error: ${err.message}") }
)

analytics_worker.cpp

client.subscribeToEvents("order-notifications", "analytics",
    [](const kubemq::Event& event) {
        std::cout << "[Analytics] Recording metric: "
                  << event.metadata << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "[Analytics] Error: " << err << std::endl;
    }
);

analytics_worker.rs

// Same group on every worker: each event reaches exactly one worker.
let sub = client
    .subscribe_to_events(
        "order-notifications",
        "analytics",
        |event| {
            Box::pin(async move {
                println!("[Analytics] Recording metric: {}", event.metadata);
            })
        },
        None,
    )
    .await?;

analytics_worker.rb

cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsSubscription.new(channel: 'order-notifications', group: 'analytics')
client.subscribe_to_events(sub, cancellation_token: cancel,
                                on_error: ->(e) { puts "[Analytics] Error: #{e.message}" }) do |event|
  puts "[Analytics] Recording metric: #{event.metadata}"
end

analytics_worker.exs

# Same group on every worker: each event reaches exactly one worker.
{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, "order-notifications",
    group: "analytics",
    on_event: fn event ->
      IO.puts("[Analytics] Recording metric: #{event.metadata}")
    end
  )

Production Considerations

Consumer Groups for load-balanced delivery
Filter Events for tag-based filtering patterns
Events Store for guaranteed delivery scenarios

Real-Time Notifications System

Architecture

Implementation

Order Status Publisher

Email Notification Subscriber

Analytics Subscriber (with Consumer Group)

Production Considerations

On this page

Real-Time Notifications System

Architecture

Implementation

Order Status Publisher

Email Notification Subscriber

Analytics Subscriber (with Consumer Group)

Production Considerations

What if a subscriber is down?

How to monitor event flow?

How to scale the notification system?

Related

On this page