Overview

Publishing events one at a time over a fresh call each time caps throughput when you need to push hundreds or thousands of events per second — log forwarding, sensor telemetry, change-data-capture feeds — where per-call overhead dominates. The Java SDK avoids that cost by keeping events on a persistent path under the hood: client.publishEvent(message) writes onto an internal bidirectional gRPC stream shared by the client rather than opening a new call per event, so a tight send loop doesn't pay a round-trip for each message.

That streaming path is what lets the loop in this sample push many events back-to-back — each publishEvent call just enqueues the next event on the open connection, which is why elapsed time barely grows per additional message.

Gotchas: publishing is fire-and-forget — publishEvent returning doesn't mean a subscriber received the event, only that it left the client; a subscriber that isn't ready yet will simply miss events sent before it subscribed. Because sends aren't acknowledged individually, give the broker a moment (or wait on a CountDownLatch as this sample does) before assuming delivery, and watch each subscription's onErrorCallback rather than expecting publishEvent itself to raise on delivery problems.

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

StreamSendExample.java

package io.kubemq.example.events;

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

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Stream Send Example
 *
 * Demonstrates sending multiple events over a persistent gRPC stream
 * for high-throughput event publishing.
 */
public class StreamSendExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-events-stream-send-client";
    private static final String CHANNEL = "java-events.stream-send";

    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);

        int messageCount = 10;
        CountDownLatch latch = new CountDownLatch(messageCount);

        EventsSubscription subscription = EventsSubscription.builder()
                .channel(CHANNEL)
                .onReceiveEventCallback(event -> {
                    System.out.println("  Stream received: " + new String(event.getBody()));
                    latch.countDown();
                })
                .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
                .build();

        // Subscribe to handle incoming events on this channel
        client.subscribeToEvents(subscription);
        // Wait for the subscriber to be ready
        Thread.sleep(500);

        // Send multiple events via the internal gRPC stream
        System.out.println("Sending " + messageCount + " events via stream...\n");
        long start = System.currentTimeMillis();

        for (int i = 1; i <= messageCount; i++) {
            EventMessage message = EventMessage.builder()
                    .id(UUID.randomUUID().toString())
                    .channel(CHANNEL)
                    .body(("Stream event #" + i).getBytes())
                    .metadata("stream-batch")
                    .build();

            // publishEvent sends via the internal gRPC stream
            client.publishEvent(message);
        }

        long elapsed = System.currentTimeMillis() - start;
        System.out.println("Sent " + messageCount + " events in " + elapsed + "ms");
        System.out.println("Throughput: " + (messageCount * 1000 / Math.max(elapsed, 1)) + " msg/sec\n");

        // 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("Stream send example completed.");
    }
}

How It Works

client.publishEvent(message) internally reuses a persistent bidirectional gRPC stream for all sends; there is no per-message connection overhead, which is why the loop achieves high throughput even without batching.
Elapsed time and throughput are measured around the tight send loop to demonstrate the difference between gRPC stream overhead and individual request overhead.
The subscriber's CountDownLatch confirms all 10 events arrive before cleanup, showing that stream-sent events are durably delivered to active subscribers.

Stream Send

Overview

Prerequisites

Code

How It Works

Related

On this page