# Basic Pub/Sub (/sdks/java/tutorials/basic-pubsub)



## Overview [#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 [#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](/sdks/java))

## Code [#code]

```java title="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 [#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.

## Related [#related]

* [Pattern overview](/learn/events/getting-started)
* [Java SDK Reference](/sdks/java/reference/events)
* [Cancel Subscription](/sdks/java/how-to/events/cancel-subscription)
* [Consumer Group](/sdks/java/how-to/events/consumer-group)