KubeMQ
LearnEvents

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.

Prerequisites: KubeMQ server running on localhost:50000 and your SDK installed. See Getting Started for setup.

What You Will Build

A notification publisher that sends order events and a subscriber that receives them in real time.

The publisher fans out one event to every connected subscriber. Because Events are at-most-once, a subscriber that is offline (Subscriber C) misses the message — there is no replay.

Steps

Install the SDK

go get github.com/kubemq-io/kubemq-go/v2
pip install kubemq
npm install kubemq-js
<dependency>
    <groupId>io.kubemq.sdk</groupId>
    <artifactId>kubemq-sdk-Java</artifactId>
    <version>3.1.1</version>
</dependency>
<PackageReference Include="KubeMQ.SDK.CSharp" Version="3.0.1" />
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1")
vcpkg install kubemq
[dependencies]
kubemq = "1.0"
tokio = { version = "1", features = ["full"] }
gem install kubemq
# mix.exs
def deps do
  [{:kubemq, "~> 1.0"}]
end

Create a Subscriber

Start the subscriber first. Events are fire-and-forget, so the subscriber must be connected before the publisher sends events.

subscriber.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-notifications", "",
        kubemq.WithOnEvent(func(event *kubemq.Event) {
            fmt.Printf("Received: %s\n", string(event.Body))
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Subscription error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()

    log.Println("Subscriber listening on 'order-notifications'...")
    <-ctx.Done()
}
subscriber.py
import time
from kubemq.pubsub import Client as PubSubClient
from kubemq.pubsub import EventsSubscription, CancellationToken

def on_event(event):
    print(f"Received: {event.body.decode('utf-8')}")

def on_error(err):
    print(f"Subscription error: {err}")

client = PubSubClient(address="localhost:50000")
cancel = CancellationToken()
client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="order-notifications",
        on_receive_event_callback=on_event,
        on_error_callback=on_error,
    ),
    cancel=cancel,
)
print("Subscriber listening on 'order-notifications'...")
time.sleep(120)
client.close()
subscriber.js
const { KubeMQClient } = require("kubemq-js");

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

client.subscribeToEvents({
  channel: "order-notifications",
  onEvent: (msg) =>
    console.log(`Received: ${Buffer.from(msg.body).toString()}`),
  onError: (err) => console.error("Subscription error:", err.message),
});

console.log("Subscriber listening on 'order-notifications'...");
Subscriber.java
import io.kubemq.sdk.pubsub.PubSubClient;
import io.kubemq.sdk.pubsub.EventsSubscription;

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

client.subscribeToEvents(EventsSubscription.builder()
    .channel("order-notifications")
    .onReceiveEventCallback(event ->
        System.out.println("Received: " + new String(event.getBody())))
    .onErrorCallback(err ->
        System.err.println("Subscription error: " + err.getMessage()))
    .build());

System.out.println("Subscriber listening on 'order-notifications'...");
Thread.sleep(120_000);
client.close();
Subscriber.cs
using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Events;
using System.Text;

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

Console.WriteLine("Subscriber listening on 'order-notifications'...");
await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "order-notifications" }))
{
    Console.WriteLine($"Received: {Encoding.UTF8.GetString(msg.Body.Span)}");
}
Subscriber.kt
import io.kubemq.sdk.client.KubeMQClient
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val pubsub = KubeMQClient.pubSub {
        address = "localhost:50000"
        clientId = "order-subscriber"
    }

    println("Subscriber listening on 'order-notifications'...")
    pubsub.subscribeToEvents {
        channel = "order-notifications"
    }.collect { event ->
        println("Received: ${String(event.body)}")
    }
}
subscriber.cpp
#include <kubemq/client.h>
#include <iostream>

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

client.subscribeToEvents("order-notifications", "",
    [](const kubemq::Event& event) {
        std::cout << "Received: " << event.body << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "Subscription error: " << err << std::endl;
    }
);

std::cout << "Subscriber listening on 'order-notifications'..." << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(120));
subscriber.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?;

    let sub: Subscription = client
        .subscribe_to_events(
            "order-notifications",
            "",
            |event| {
                Box::pin(async move {
                    println!("Received: {}", String::from_utf8_lossy(&event.body));
                })
            },
            None,
        )
        .await?;

    println!("Subscriber listening on 'order-notifications'...");
    tokio::time::sleep(Duration::from_secs(120)).await;

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

client = KubeMQ::PubSubClient.new(address: "localhost:50000", client_id: "order-subscriber")
cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsSubscription.new(channel: "order-notifications")
client.subscribe_to_events(sub, cancellation_token: cancel,
                           on_error: ->(e) { puts "Subscription error: #{e.message}" }) do |event|
  puts "Received: #{event.body}"
end

puts "Subscriber listening on 'order-notifications'..."
sleep 120
cancel.cancel
client.close
subscriber.exs
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-subscriber")

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, "order-notifications",
    on_event: fn event -> IO.puts("Received: #{event.body}") end,
    on_error: fn err -> IO.puts("Subscription error: #{err.message}") end
  )

IO.puts("Subscriber listening on 'order-notifications'...")
Process.sleep(120_000)

KubeMQ.Subscription.cancel(sub)
KubeMQ.Client.close(client)

Create a Publisher

In a separate terminal, run the publisher. The subscriber receives the event instantly.

publisher.go
package main

import (
    "context"
    "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()

    err = client.SendEvent(ctx, kubemq.NewEvent().
        SetChannel("order-notifications").
        SetBody([]byte(`{"orderId":"ORD-1234","status":"created"}`)),
    )
    if err != nil {
        log.Fatal(err)
    }
    log.Println("Event published to 'order-notifications'")
}
publisher.py
from kubemq.pubsub import Client as PubSubClient
from kubemq.pubsub import EventMessage

client = PubSubClient(address="localhost:50000")
client.send_event(
    EventMessage(
        channel="order-notifications",
        body=b'{"orderId":"ORD-1234","status":"created"}',
    )
)
print("Event published to 'order-notifications'")
client.close()
publisher.js
const { KubeMQClient } = require("kubemq-js");

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

await client.sendEvent({
  channel: "order-notifications",
  body: Buffer.from(JSON.stringify({ orderId: "ORD-1234", status: "created" })),
});

console.log("Event published to 'order-notifications'");
Publisher.java
import io.kubemq.sdk.pubsub.PubSubClient;
import io.kubemq.sdk.pubsub.EventMessage;

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

client.sendEventsMessage(EventMessage.builder()
    .channel("order-notifications")
    .body("{\"orderId\":\"ORD-1234\",\"status\":\"created\"}".getBytes())
    .build());

System.out.println("Event published to 'order-notifications'");
client.close();
Publisher.cs
using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Events;
using System.Text;

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

await client.SendEventAsync(new EventMessage
{
    Channel = "order-notifications",
    Body = Encoding.UTF8.GetBytes(
        "{\"orderId\":\"ORD-1234\",\"status\":\"created\"}")
});

Console.WriteLine("Event published to 'order-notifications'");
Publisher.kt
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.eventMessage
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val pubsub = KubeMQClient.pubSub {
        address = "localhost:50000"
        clientId = "order-publisher"
    }

    pubsub.publishEvent(eventMessage {
        channel = "order-notifications"
        body = """{"orderId":"ORD-1234","status":"created"}""".toByteArray()
    })

    println("Event published to 'order-notifications'")
}
publisher.cpp
#include <kubemq/client.h>
#include <iostream>

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

kubemq::EventMessage event;
event.channel = "order-notifications";
event.body = R"({"orderId":"ORD-1234","status":"created"})";

client.sendEvent(event);
std::cout << "Event published to 'order-notifications'" << std::endl;
publisher.rs
use kubemq::prelude::*;
use kubemq::EventBuilder;

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

    let event = EventBuilder::new()
        .channel("order-notifications")
        .body(br#"{"orderId":"ORD-1234","status":"created"}"#.to_vec())
        .build();

    client.send_event(event).await?;
    println!("Event published to 'order-notifications'");

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

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

client.send_event(KubeMQ::PubSub::EventMessage.new(
  channel: "order-notifications",
  body: '{"orderId":"ORD-1234","status":"created"}'
))

puts "Event published to 'order-notifications'"
client.close
publisher.exs
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-publisher")

event = KubeMQ.Event.new(
  channel: "order-notifications",
  body: ~s({"orderId":"ORD-1234","status":"created"})
)

:ok = KubeMQ.Client.send_event(client, event)
IO.puts("Event published to 'order-notifications'")

KubeMQ.Client.close(client)

Run the Example

  1. Start the subscriber in one terminal
  2. Run the publisher in a separate terminal
  3. The subscriber receives the event in real time

Verify

The subscriber terminal should display:

Subscriber listening on 'order-notifications'...
Received: {"orderId":"ORD-1234","status":"created"}

The publisher terminal should display:

Event published to 'order-notifications'

What Just Happened

  1. The subscriber connected to KubeMQ and registered interest in order-notifications
  2. The publisher sent an event containing an order payload to the same channel
  3. KubeMQ delivered the event to the subscriber in real time
  4. Because Events are fire-and-forget, the publisher does not wait for acknowledgment

Events have at-most-once delivery. If the subscriber was not connected when the event was published, the message would be lost. For persistent delivery with replay, use Events Store.

Next Steps

Publish & Subscribe

Multiple subscribers, metadata, and tags.

Consumer Groups

Load balance across subscribers.

Events Reference

Message structure and validation rules.

Was this page helpful?

Events — Real-Time Pub/Sub

Publish and subscribe to real-time events with fire-and-forget delivery and at-most-once semantics.

Publish & Subscribe

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

On this page

What You Will BuildStepsInstall the SDKCreate a SubscriberCreate a PublisherRun the ExampleVerifyWhat Just HappenedNext Steps