Overview

This tutorial builds the "hello world" of KubeMQ messaging: a publisher and a subscriber talking over the Events pattern. Events are fire-and-forget — the broker fans a message out to every subscriber currently listening on the channel and moves on. There's no persistence, no acknowledgment, and no replay, which makes this the pattern to reach for when you need low-latency, high-throughput broadcast (metrics ticks, live status updates, cache-invalidation signals) and can tolerate losing a message if nobody is listening at the moment it's sent.

You'll wire up subscribeToEvents(subscription) with an EventsSubscription built from an onReceiveEventCallback, give the subscription a moment to register with the server, then call publishEvent(message) to publish an EventMessage. Every connected subscriber on the channel gets its own copy, as opposed to a consumer group where only one member would receive it. Gotchas: if the subscriber isn't fully established before you publish, the event is simply gone — there's no queue catching it, which is why the sample sleeps briefly before sending; and because delivery isn't acknowledged, a crashed or disconnected subscriber never knows it missed anything.

Prerequisites

KubeMQ server running on localhost:50000
Java SDK installed (implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1' (Gradle) or Maven dependency from Getting Started)

Code

BasicPubSubExample.java

package io.kubemq.example.events;

import io.kubemq.sdk.common.ServerInfo;
import io.kubemq.sdk.pubsub.*;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * Basic Pub/Sub Example
 *
 * Demonstrates publishing and subscribing to non-persistent events.
 */
public class BasicPubSubExample {

    // TODO: Replace with your KubeMQ server address
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-events-basic-pubsub-client";
    private static final String CHANNEL = "java-events.basic-pubsub";

    public static void main(String[] args) throws InterruptedException {
        // Create a client connected to the KubeMQ server
        PubSubClient client = PubSubClient.builder()
                .address(ADDRESS)
                .clientId(CLIENT_ID)
                .build();

        // Verify connection to the server
        ServerInfo info = client.ping();
        System.out.println("Connected to: " + info.getHost());

        // Create the events channel
        client.createEventsChannel(CHANNEL);

        CountDownLatch latch = new CountDownLatch(3);

        Consumer<EventMessageReceived> onReceive = event -> {
            System.out.println("Received event:");
            System.out.println("  ID: " + event.getId());
            System.out.println("  Channel: " + event.getChannel());
            System.out.println("  Body: " + new String(event.getBody()));
            System.out.println("  Tags: " + event.getTags());
            latch.countDown();
        };

        Consumer<io.kubemq.sdk.exception.KubeMQException> onError = error -> {
            System.err.println("Error: " + error.getMessage());
        };

        EventsSubscription subscription = EventsSubscription.builder()
                .channel(CHANNEL)
                .onReceiveEventCallback(onReceive)
                .onErrorCallback(onError)
                .build();

        // Subscribe to handle incoming events on this channel
        client.subscribeToEvents(subscription);
        System.out.println("Subscribed to events on: " + CHANNEL);

        // Wait for the subscriber to be ready
        Thread.sleep(500);

        // Send event messages
        for (int i = 1; i <= 3; i++) {
            Map<String, String> tags = new HashMap<>();
            tags.put("sequence", String.valueOf(i));

            EventMessage message = EventMessage.builder()
                    .id(UUID.randomUUID().toString())
                    .channel(CHANNEL)
                    .metadata("Event metadata")
                    .body(("Hello KubeMQ event #" + i).getBytes())
                    .tags(tags)
                    .build();

            client.publishEvent(message);
            System.out.println("Sent event #" + i);
        }

        // Wait for the subscriber to receive all messages
        latch.await(5, TimeUnit.SECONDS);

        // Clean up resources
        subscription.cancel();
        client.deleteEventsChannel(CHANNEL);
        client.close();
        System.out.println("\nBasic pub/sub example completed.");
    }
}

// Expected output:
// Connected to: <host>
// Subscribed to events on: java-events.basic-pubsub
// Sent event #1
// Sent event #2
// Sent event #3
// Received event:
//   ID: <message-id>
//   Channel: java-events.basic-pubsub
//   Body: Hello KubeMQ event #1
//   Tags: {sequence=1}
// Received event:
//   ID: <message-id>
//   Channel: java-events.basic-pubsub
//   Body: Hello KubeMQ event #2
//   Tags: {sequence=2}
// Received event:
//   ID: <message-id>
//   Channel: java-events.basic-pubsub
//   Body: Hello KubeMQ event #3
//   Tags: {sequence=3}
//
// Basic pub/sub example completed.

How It Works

PubSubClient.builder() creates a gRPC connection to the broker; the client is reused for both subscribe and publish.
EventsSubscription.builder() registers an onReceiveEventCallback (Consumer<EventMessageReceived>) that runs on a background gRPC thread for every arriving event.
client.subscribeToEvents(subscription) opens the server-streaming gRPC call; a brief Thread.sleep lets the stream establish before publishing.
EventMessage.builder() constructs each message with an ID, body, metadata, and tag map; client.publishEvent(message) sends it fire-and-forget over the shared gRPC channel.
CountDownLatch synchronises the main thread: it waits until all three events have been received before cleanup.

Basic Pub/Sub

Overview

Prerequisites

Code

How It Works

Related

On this page