# Real-Time Notifications System (/learn/events/scenarios/real-time-notifications)



## Architecture [#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.

<Mermaid
  chart="graph LR
    OS[&#x22;Order Service&#x22;]
    K{{&#x22;Events channel<br/>order-notifications&#x22;}}
    ES[&#x22;Email Service&#x22;]
    PS[&#x22;Push Service&#x22;]
    A1[&#x22;Analytics Worker 1&#x22;]
    A2[&#x22;Analytics Worker 2&#x22;]

    OS -- publish --> K
    K -- &#x22;fan-out&#x22; --> ES
    K -- &#x22;fan-out&#x22; --> PS
    K -- &#x22;group: analytics&#x22; --> A1
    K -- &#x22;group: analytics&#x22; --> A2

    class K events
    class OS,ES,PS,A1,A2 client"
/>

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

## Implementation [#implementation]

### Order Status Publisher [#order-status-publisher]

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

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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)
        }
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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()
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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));
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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();
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="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);
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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()
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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));
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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)
    ```
  </Tab>
</Tabs>

### Email Notification Subscriber [#email-notification-subscriber]

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

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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)
        }),
    )
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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(),
    )
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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),
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="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"]}");
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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}") }
    )
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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;
        }
    );
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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?;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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
      )
    ```
  </Tab>
</Tabs>

### Analytics Subscriber (with Consumer Group) [#analytics-subscriber-with-consumer-group]

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

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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)
        }),
    )
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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(),
    )
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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),
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="AnalyticsWorker.cs"
    await foreach (var msg in client.SubscribeToEventsAsync(
        new EventsSubscription { Channel = "order-notifications", Group = "analytics" }))
    {
        Console.WriteLine($"[Analytics] Recording metric: {msg.Metadata}");
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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}") }
    )
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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;
        }
    );
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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?;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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
      )
    ```
  </Tab>
</Tabs>

## Production Considerations [#production-considerations]

<Accordions>
  <Accordion title="What if a subscriber is down?">
    Events use at-most-once delivery. If the email service is down when a "shipped" event is published, that notification is lost. For critical notifications, consider using [Events Store](/learn/events-store) to guarantee delivery, or implement a heartbeat/health check that alerts when a subscriber disconnects.
  </Accordion>

  <Accordion title="How to monitor event flow?">
    Add an ungrouped monitor subscriber that logs all events for observability. Use tags to track event counts per status type. Monitor server-side logs for `writeDeadline` warnings that indicate slow consumers.
  </Accordion>

  <Accordion title="How to scale the notification system?">
    Use consumer groups for services that can be parallelized (like analytics). Services that must see every event (like email) should remain ungrouped. Scale ungrouped services vertically or use application-level buffering.
  </Accordion>
</Accordions>

## Related [#related]

* [Consumer Groups](/learn/events/tutorials/consumer-groups) for load-balanced delivery
* [Filter Events](/learn/events/how-to/filter-events) for tag-based filtering patterns
* [Events Store](/learn/events-store) for guaranteed delivery scenarios