KubeMQ
LearnEventsTutorials

Publish & Subscribe

Learn the basic pub/sub pattern with KubeMQ Events including multiple subscribers and message metadata.

What You Will Build

A publisher sending order events with metadata and tags, and two subscribers both receiving every event (fan-out).

One publisher fans out every order event to both subscribers at-most-once.

Prerequisites

Steps

Set Up the Publisher

The publisher sends order events with a JSON body, metadata, and tags for downstream filtering.

order_publisher.go
package main

import (
    "context"
    "fmt"
    "log"
    "time"

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

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

    orders := []struct {
        ID     string
        Amount float64
        Region string
    }{
        {"ORD-001", 99.99, "us-east"},
        {"ORD-002", 249.50, "eu-west"},
        {"ORD-003", 15.00, "us-east"},
    }

    for _, order := range orders {
        body := fmt.Sprintf(
            `{"orderId":"%s","amount":%.2f,"region":"%s"}`,
            order.ID, order.Amount, order.Region)

        err = client.SendEvent(ctx, kubemq.NewEvent().
            SetChannel("order-events").
            SetMetadata("order.created").
            SetBody([]byte(body)).
            SetTags(map[string]string{"region": order.Region}),
        )
        if err != nil {
            log.Printf("Failed to send event for %s: %v", order.ID, err)
            continue
        }
        log.Printf("Published order event: %s", order.ID)
        time.Sleep(500 * time.Millisecond)
    }
}
order_publisher.py
import json
import time
from kubemq.pubsub import Client as PubSubClient
from kubemq.pubsub import EventMessage

orders = [
    {"orderId": "ORD-001", "amount": 99.99, "region": "us-east"},
    {"orderId": "ORD-002", "amount": 249.50, "region": "eu-west"},
    {"orderId": "ORD-003", "amount": 15.00, "region": "us-east"},
]

client = PubSubClient(address="localhost:50000")
for order in orders:
    try:
        client.send_event(
            EventMessage(
                channel="order-events",
                metadata="order.created",
                body=json.dumps(order).encode("utf-8"),
                tags={"region": order["region"]},
            )
        )
        print(f"Published order event: {order['orderId']}")
    except Exception as e:
        print(f"Failed to send event for {order['orderId']}: {e}")
    time.sleep(0.5)
client.close()
order_publisher.js
const { KubeMQClient } = require("kubemq-js");

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

const orders = [
  { orderId: "ORD-001", amount: 99.99, region: "us-east" },
  { orderId: "ORD-002", amount: 249.50, region: "eu-west" },
  { orderId: "ORD-003", amount: 15.00, region: "us-east" },
];

for (const order of orders) {
  try {
    await client.sendEvent({
      channel: "order-events",
      metadata: "order.created",
      body: Buffer.from(JSON.stringify(order)),
      tags: { region: order.region },
    });
    console.log(`Published order event: ${order.orderId}`);
  } catch (err) {
    console.error(`Failed to send event for ${order.orderId}:`, err);
  }
  await new Promise((r) => setTimeout(r, 500));
}
OrderPublisher.java
PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("order-publisher")
    .build();

String[][] orders = {
    {"ORD-001", "99.99", "us-east"},
    {"ORD-002", "249.50", "eu-west"},
    {"ORD-003", "15.00", "us-east"},
};

for (String[] order : orders) {
    String body = String.format(
        "{\"orderId\":\"%s\",\"amount\":%s,\"region\":\"%s\"}",
        order[0], order[1], order[2]);
    try {
        client.sendEventsMessage(EventMessage.builder()
            .channel("order-events")
            .metadata("order.created")
            .body(body.getBytes())
            .tags(Map.of("region", order[2]))
            .build());
        System.out.println("Published order event: " + order[0]);
    } catch (Exception e) {
        System.err.println("Failed to send: " + e.getMessage());
    }
    Thread.sleep(500);
}
client.close();
OrderPublisher.cs
await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

var orders = new[]
{
    new { Id = "ORD-001", Amount = 99.99, Region = "us-east" },
    new { Id = "ORD-002", Amount = 249.50, Region = "eu-west" },
    new { Id = "ORD-003", Amount = 15.00, Region = "us-east" },
};

foreach (var order in orders)
{
    var body = $"{{\"orderId\":\"{order.Id}\",\"amount\":{order.Amount},\"region\":\"{order.Region}\"}}";
    await client.SendEventAsync(new EventMessage
    {
        Channel = "order-events",
        Metadata = "order.created",
        Body = Encoding.UTF8.GetBytes(body),
        Tags = new Dictionary<string, string> { ["region"] = order.Region },
    });
    Console.WriteLine($"Published order event: {order.Id}");
    await Task.Delay(500);
}
OrderPublisher.kt
val client = PubSubClient("localhost:50000")

data class Order(val id: String, val amount: Double, val region: String)
val orders = listOf(
    Order("ORD-001", 99.99, "us-east"),
    Order("ORD-002", 249.50, "eu-west"),
    Order("ORD-003", 15.00, "us-east"),
)

for (order in orders) {
    val body = """{"orderId":"${order.id}","amount":${order.amount},"region":"${order.region}"}"""
    client.sendEvent(EventMessage(
        channel = "order-events",
        metadata = "order.created",
        body = body.toByteArray(),
        tags = mapOf("region" to order.region),
    ))
    println("Published order event: ${order.id}")
    Thread.sleep(500)
}
client.close()
order_publisher.cpp
#include <kubemq/client.h>
#include <iostream>
#include <thread>
#include <chrono>

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

struct Order { std::string id; double amount; std::string region; };
std::vector<Order> orders = {
    {"ORD-001", 99.99, "us-east"},
    {"ORD-002", 249.50, "eu-west"},
    {"ORD-003", 15.00, "us-east"},
};

for (const auto& order : orders) {
    kubemq::EventMessage event;
    event.channel = "order-events";
    event.metadata = "order.created";
    event.body = "{\"orderId\":\"" + order.id + "\",\"amount\":" +
                 std::to_string(order.amount) + ",\"region\":\"" + order.region + "\"}";
    event.tags["region"] = order.region;

    client.sendEvent(event);
    std::cout << "Published order event: " << order.id << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
}
order_publisher.rs
use kubemq::prelude::*;
use kubemq::EventBuilder;
use std::collections::HashMap;
use std::time::Duration;

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

    let orders = [
        ("ORD-001", 99.99, "us-east"),
        ("ORD-002", 249.50, "eu-west"),
        ("ORD-003", 15.00, "us-east"),
    ];

    for (id, amount, region) in orders {
        let body = format!(
            r#"{{"orderId":"{id}","amount":{amount},"region":"{region}"}}"#
        );
        let mut tags = HashMap::new();
        tags.insert("region".to_string(), region.to_string());

        let event = EventBuilder::new()
            .channel("order-events")
            .metadata("order.created")
            .body(body.into_bytes())
            .tags(tags)
            .build();

        match client.send_event(event).await {
            Ok(_) => println!("Published order event: {id}"),
            Err(e) => eprintln!("Failed to send event for {id}: {e}"),
        }
        tokio::time::sleep(Duration::from_millis(500)).await;
    }

    client.close().await?;
    Ok(())
}
order_publisher.rb
require 'kubemq'
require 'json'

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

orders = [
  { orderId: 'ORD-001', amount: 99.99, region: 'us-east' },
  { orderId: 'ORD-002', amount: 249.50, region: 'eu-west' },
  { orderId: 'ORD-003', amount: 15.00, region: 'us-east' },
]

orders.each do |order|
  begin
    msg = KubeMQ::PubSub::EventMessage.new(
      channel: 'order-events',
      metadata: 'order.created',
      body: order.to_json,
      tags: { 'region' => order[:region] }
    )
    client.send_event(msg)
    puts "Published order event: #{order[:orderId]}"
  rescue KubeMQ::Error => e
    puts "Failed to send event for #{order[:orderId]}: #{e.message}"
  end
  sleep 0.5
end

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

orders = [
  %{order_id: "ORD-001", amount: 99.99, region: "us-east"},
  %{order_id: "ORD-002", amount: 249.50, region: "eu-west"},
  %{order_id: "ORD-003", amount: 15.00, region: "us-east"}
]

for order <- orders do
  body =
    ~s({"orderId":"#{order.order_id}","amount":#{order.amount},"region":"#{order.region}"})

  event =
    KubeMQ.Event.new(
      channel: "order-events",
      metadata: "order.created",
      body: body,
      tags: %{"region" => order.region}
    )

  case KubeMQ.Client.send_event(client, event) do
    :ok -> IO.puts("Published order event: #{order.order_id}")
    {:error, err} -> IO.puts("Failed to send event for #{order.order_id}: #{err.message}")
  end

  Process.sleep(500)
end

KubeMQ.Client.close(client)

Set Up Subscriber A

The notification service processes every order event.

notification_service.go
package main

import (
    "context"
    "fmt"
    "log"

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

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

    sub, err := client.SubscribeToEvents(ctx, "order-events", "",
        kubemq.WithOnEvent(func(event *kubemq.Event) {
            fmt.Printf("[Notification] New order: %s | metadata: %s\n",
                string(event.Body), event.Metadata)
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("[Notification] Error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()

    log.Println("[Notification] Service listening...")
    <-ctx.Done()
}
notification_service.py
import time
from kubemq.pubsub import Client as PubSubClient
from kubemq.pubsub import EventsSubscription, CancellationToken

def on_event(event):
    print(f"[Notification] New order: "
          f"{event.body.decode('utf-8')} | metadata: {event.metadata}")

client = PubSubClient(address="localhost:50000")
client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="order-events",
        on_receive_event_callback=on_event,
        on_error_callback=lambda e: print(f"[Notification] Error: {e}"),
    ),
    cancel=CancellationToken(),
)
print("[Notification] Service listening...")
time.sleep(300)
client.close()
notification_service.js
const { KubeMQClient } = require("kubemq-js");

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

client.subscribeToEvents({
  channel: "order-events",
  onEvent: (msg) =>
    console.log(
      `[Notification] New order: ${Buffer.from(msg.body).toString()}`
    ),
  onError: (err) => console.error("[Notification] Error:", err.message),
});

console.log("[Notification] Service listening...");
NotificationService.java
PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("notification-service")
    .build();

client.subscribeToEvents(EventsSubscription.builder()
    .channel("order-events")
    .onReceiveEventCallback(event ->
        System.out.println("[Notification] New order: "
            + new String(event.getBody())))
    .onErrorCallback(err ->
        System.err.println("[Notification] Error: " + err.getMessage()))
    .build());

System.out.println("[Notification] Service listening...");
Thread.sleep(300_000);
client.close();
NotificationService.cs
await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

Console.WriteLine("[Notification] Service listening...");
await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "order-events" }))
{
    Console.WriteLine($"[Notification] New order: "
        + $"{Encoding.UTF8.GetString(msg.Body.Span)}");
}
NotificationService.kt
val client = PubSubClient("localhost:50000")

client.subscribeToEvents(
    channel = "order-events",
    onEvent = { event ->
        println("[Notification] New order: ${String(event.body)}")
    },
    onError = { err ->
        System.err.println("[Notification] Error: ${err.message}")
    }
)

println("[Notification] Service listening...")
Thread.sleep(300_000)
client.close()
notification_service.cpp
auto client = kubemq::PubSubClient("localhost:50000");

client.subscribeToEvents("order-events", "",
    [](const kubemq::Event& event) {
        std::cout << "[Notification] New order: " << event.body << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "[Notification] Error: " << err << std::endl;
    }
);

std::cout << "[Notification] Service listening..." << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(300));
notification_service.rs
use kubemq::prelude::*;
use kubemq::Subscription;
use std::time::Duration;

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

    // Subscribe with an empty group -- every subscriber receives every event
    let sub: Subscription = client
        .subscribe_to_events(
            "order-events",
            "",
            |event| {
                Box::pin(async move {
                    println!(
                        "[Notification] New order: {} | metadata: {}",
                        String::from_utf8_lossy(&event.body),
                        event.metadata
                    );
                })
            },
            None,
        )
        .await?;

    println!("[Notification] Service listening...");
    tokio::time::sleep(Duration::from_secs(300)).await;

    sub.unsubscribe().await;
    client.close().await?;
    Ok(())
}
notification_service.rb
require 'kubemq'

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

sub = KubeMQ::PubSub::EventsSubscription.new(channel: 'order-events')
client.subscribe_to_events(
  sub,
  cancellation_token: cancel,
  on_error: ->(e) { puts "[Notification] Error: #{e.message}" }
) do |event|
  puts "[Notification] New order: #{event.body} | metadata: #{event.metadata}"
end

puts '[Notification] Service listening...'
sleep 300

cancel.cancel
client.close
notification_service.exs
{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "notification-service")

{:ok, _sub} =
  KubeMQ.Client.subscribe_to_events(client, "order-events",
    on_event: fn event ->
      IO.puts("[Notification] New order: #{event.body} | metadata: #{event.metadata}")
    end,
    on_error: fn err -> IO.puts("[Notification] Error: #{err.message}") end
  )

IO.puts("[Notification] Service listening...")
Process.sleep(300_000)

KubeMQ.Client.close(client)

Set Up Subscriber B

A second subscriber receives the same events independently for analytics processing.

analytics_service.go
sub, err := client.SubscribeToEvents(ctx, "order-events", "",
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[Analytics] Processing: %s\n", string(event.Body))
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("[Analytics] Error:", err)
    }),
)
analytics_service.py
def on_event(event):
    print(f"[Analytics] Processing: {event.body.decode('utf-8')}")

client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="order-events",
        on_receive_event_callback=on_event,
        on_error_callback=lambda e: print(f"[Analytics] Error: {e}"),
    ),
    cancel=CancellationToken(),
)
print("[Analytics] Service listening...")
analytics_service.js
client.subscribeToEvents({
  channel: "order-events",
  onEvent: (msg) =>
    console.log(
      `[Analytics] Processing: ${Buffer.from(msg.body).toString()}`
    ),
  onError: (err) => console.error("[Analytics] Error:", err.message),
});

console.log("[Analytics] Service listening...");
AnalyticsService.java
client.subscribeToEvents(EventsSubscription.builder()
    .channel("order-events")
    .onReceiveEventCallback(event ->
        System.out.println("[Analytics] Processing: "
            + new String(event.getBody())))
    .onErrorCallback(err ->
        System.err.println("[Analytics] Error: " + err.getMessage()))
    .build());

System.out.println("[Analytics] Service listening...");
AnalyticsService.cs
Console.WriteLine("[Analytics] Service listening...");
await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "order-events" }))
{
    Console.WriteLine($"[Analytics] Processing: "
        + $"{Encoding.UTF8.GetString(msg.Body.Span)}");
}
AnalyticsService.kt
client.subscribeToEvents(
    channel = "order-events",
    onEvent = { event ->
        println("[Analytics] Processing: ${String(event.body)}")
    },
    onError = { err ->
        System.err.println("[Analytics] Error: ${err.message}")
    }
)
println("[Analytics] Service listening...")
analytics_service.cpp
client.subscribeToEvents("order-events", "",
    [](const kubemq::Event& event) {
        std::cout << "[Analytics] Processing: " << event.body << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "[Analytics] Error: " << err << std::endl;
    }
);
std::cout << "[Analytics] Service listening..." << std::endl;
analytics_service.rs
// Same channel, empty group -- this subscriber receives the same events
// independently of the notification service.
let sub: Subscription = client
    .subscribe_to_events(
        "order-events",
        "",
        |event| {
            Box::pin(async move {
                println!(
                    "[Analytics] Processing: {}",
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;
println!("[Analytics] Service listening...");
analytics_service.rb
# Same channel, no group -- this subscriber receives the same events
# independently of the notification service.
sub = KubeMQ::PubSub::EventsSubscription.new(channel: 'order-events')
client.subscribe_to_events(
  sub,
  cancellation_token: cancel,
  on_error: ->(e) { puts "[Analytics] Error: #{e.message}" }
) do |event|
  puts "[Analytics] Processing: #{event.body}"
end
puts '[Analytics] Service listening...'
analytics_service.exs
# Same channel, no group -- this subscriber receives the same events
# independently of the notification service.
{:ok, _sub} =
  KubeMQ.Client.subscribe_to_events(client, "order-events",
    on_event: fn event ->
      IO.puts("[Analytics] Processing: #{event.body}")
    end,
    on_error: fn err -> IO.puts("[Analytics] Error: #{err.message}") end
  )

IO.puts("[Analytics] Service listening...")

Observe Fan-Out

Start both subscribers in separate terminals, then run the publisher. Both subscribers receive every event. No acknowledgment needed.

Notification Service output:

[Notification] New order: {"orderId":"ORD-001","amount":99.99,"region":"us-east"}
[Notification] New order: {"orderId":"ORD-002","amount":249.50,"region":"eu-west"}
[Notification] New order: {"orderId":"ORD-003","amount":15.00,"region":"us-east"}

Analytics Service output:

[Analytics] Processing: {"orderId":"ORD-001","amount":99.99,"region":"us-east"}
[Analytics] Processing: {"orderId":"ORD-002","amount":249.50,"region":"eu-west"}
[Analytics] Processing: {"orderId":"ORD-003","amount":15.00,"region":"us-east"}

How Fan-Out Works

KubeMQ delivers each published event to every active subscriber independently — no acknowledgment, no ordering coordination between subscribers.

Key Points

  • No acknowledgment needed — fire-and-forget delivery
  • Missed messages — if a subscriber is offline, events sent while it was disconnected are lost
  • Metadata and tags — propagated to all subscribers for downstream filtering and routing

If a subscriber is slow to process events, messages may be dropped after the write deadline (default 2 seconds). See Handle Slow Consumers for mitigation strategies.

Next Steps

Consumer Groups

Distribute events across a group.

Wildcard Subscriptions

Subscribe to multiple channels.

Was this page helpful?

Getting Started with Events

Build a fire-and-forget publisher and subscriber and send your first KubeMQ event in 5 minutes — at-most-once delivery, no persistence.

Consumer Groups

Distribute event processing across multiple consumers with load-balanced groups.

On this page

What You Will BuildPrerequisitesStepsSet Up the PublisherSet Up Subscriber ASet Up Subscriber BObserve Fan-OutHow Fan-Out WorksKey PointsNext Steps