# API Reference (/integrations/spring-boot/reference/api) This page is the authoritative reference for the KubeMQ Spring Boot Starter's runtime API: the `KubeMQTemplate` send methods and fluent builders, the listener and handler annotation attributes, the Spring Cloud Stream binder properties, and the Actuator endpoints. For module coordinates and the full `kubemq.*` property tables, see the [configuration reference](/integrations/spring-boot/reference/configuration). ## KubeMQTemplate API [#kubemqtemplate-api] `KubeMQTemplate` is a thread-safe, injectable bean that covers all five messaging patterns. It wraps the native SDK clients (`PubSubClient`, `QueuesClient`, `CQClient`) with payload serialization and optional Micrometer observation. Inject it through the constructor: ```java title="OrderService.java" @Service public class OrderService { private final KubeMQTemplate template; public OrderService(KubeMQTemplate template) { this.template = template; } } ``` ### Send Methods [#send-methods] `channel` is the target channel, `data` is the payload (serialized by the configured `KubeMQMessageConverter`, with a built-in fallback for `byte[]` and `String`), and `tags` is an optional `Map`. Command and query sends take a `Duration timeout`. | Method | Returns | Description | | ------------------------------------------------------------------------------ | ------------------------------------------- | ----------------------------------------------------------------------- | | `sendEvent(String channel, Object data)` | `void` | Fire-and-forget event. | | `sendEvent(String channel, Object data, Map tags)` | `void` | Event with tags. | | `sendEventAsync(String channel, Object data)` | `CompletableFuture` | Async event. | | `sendEventAsync(String channel, Object data, Map tags)` | `CompletableFuture` | Async event with tags. | | `sendEventStore(String channel, Object data)` | `void` | Persistent, replayable event. | | `sendEventStore(String channel, Object data, Map tags)` | `void` | Persistent event with tags. | | `sendEventStoreAsync(String channel, Object data)` | `CompletableFuture` | Async persistent event. | | `sendEventStoreAsync(String channel, Object data, Map tags)` | `CompletableFuture` | Async persistent event with tags. | | `sendQueueMessage(String channel, Object data)` | `void` | Single durable queue message. | | `sendQueueMessage(String channel, Object data, Map tags)` | `void` | Queue message with tags. | | `sendQueueMessages(String channel, List data)` | `void` | Multiple queue messages to one channel. Empty/`null` lists are a no-op. | | `sendQueueMessageAsync(String channel, Object data)` | `CompletableFuture` | Async queue message. | | `sendQueueMessageAsync(String channel, Object data, Map tags)` | `CompletableFuture` | Async queue message with tags. | | `sendCommand(String channel, Object data, Duration timeout)` | `CommandResponseMessage` | RPC command; blocks until the handler responds or the timeout elapses. | | `sendCommandAsync(String channel, Object data, Duration timeout)` | `CompletableFuture` | Async RPC command. | | `sendQuery(String channel, Object data, Duration timeout)` | `QueryResponseMessage` | RPC query; blocks and returns the response body. | | `sendQueryAsync(String channel, Object data, Duration timeout)` | `CompletableFuture` | Async RPC query. | ```java title="SendExamples.java" // Events template.sendEvent("orders", order); template.sendEvent("orders", order, Map.of("region", "eu")); template.sendEventAsync("orders", order); // Events Store (persistent, replayable) template.sendEventStore("orders.audit", auditEntry); // Queues template.sendQueueMessage("queues.orders", order); template.sendQueueMessages("queues.orders", List.of(order1, order2, order3)); // Commands & Queries (request-response) CommandResponseMessage cmd = template.sendCommand("commands.device", payload, Duration.ofSeconds(10)); QueryResponseMessage qry = template.sendQuery("queries.user-lookup", request, Duration.ofSeconds(10)); ``` ### Fluent Builders [#fluent-builders] Each fluent factory returns a builder; chain `toChannel(...)`, optional `withTag` / `withTags` / `withMetadata`, plus pattern-specific options, then call the terminal `send()` (or `sendAsync()`). | Factory | Returns | Pattern-specific options | Terminal ops | | ------------------------------ | ----------------------------- | ------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------ | | `newEvent(Object data)` | `KubeMQEventMessageBuilder` | — | `send()` → `void`, `sendAsync()` → `CompletableFuture` | | `newEventStore(Object data)` | `KubeMQEventMessageBuilder` | — | `send()` → `void`, `sendAsync()` → `CompletableFuture` | | `newQueueMessage(Object data)` | `KubeMQQueueMessageBuilder` | `withDelay(Duration)`, `withExpiration(Duration)`, `withDeadLetterQueue(String, int)` | `send()` → `void`, `sendAsync()` → `CompletableFuture` | | `newCommand(Object data)` | `KubeMQCommandMessageBuilder` | `withTimeout(Duration)` | `send()` → `CommandResponseMessage`, `sendAsync()` → `CompletableFuture` | | `newQuery(Object data)` | `KubeMQQueryMessageBuilder` | `withTimeout(Duration)`, `withCacheKey(String)`, `withCacheTTL(Duration)` | `send()` → `QueryResponseMessage`, `sendAsync()` → `CompletableFuture` | ```java title="FluentBuilders.java" template.newEvent("Tagged event") .toChannel("spring-fluent.events") .withTag("source", "fluent-builder") .send(); template.newQueueMessage("Delayed queue msg") .toChannel("spring-fluent.queues") .withDelay(Duration.ofSeconds(1)) .withExpiration(Duration.ofSeconds(30)) .send(); CommandResponseMessage cmdResponse = template.newCommand("Fluent command") .toChannel("spring-fluent.commands") .withTimeout(Duration.ofSeconds(10)) .send(); QueryResponseMessage queryResponse = template.newQuery("Fluent query") .toChannel("spring-fluent.queries") .withTimeout(Duration.ofSeconds(10)) .withCacheKey("fluent-cache") .withCacheTTL(Duration.ofMinutes(5)) .send(); ``` Configure a `KubeMQMessageConverter` bean (for example a Jackson-based JSON converter) to serialize arbitrary payload types. Without a converter, `KubeMQTemplate` only accepts `byte[]` and `String` payloads and throws `IllegalArgumentException` for anything else. ## Listener Annotations [#listener-annotations] Listener and handler annotations turn bean methods into message consumers. Attribute values are `String` because they resolve SpEL expressions and `${...}` property placeholders at runtime; an empty default means "fall back to the `kubemq.listener.*` configuration". Place the annotations on `@Component` (or other Spring bean) methods. ### @KubeMQEventListener [#kubemqeventlistener] Receives `EventMessageReceived` from one or more channels (push-based, fire-and-forget). | Attribute | Type | Default | Description | | ------------------ | ---------- | ------------ | -------------------------------------------------------------- | | `channels` | `String[]` | *(required)* | Channel names to subscribe to. Supports SpEL and placeholders. | | `group` | `String` | `""` | Consumer group for shared subscriptions. Empty means no group. | | `concurrency` | `String` | `""` | Number of concurrent processors (resolved as integer). | | `containerFactory` | `String` | `""` | Bean name of a custom `KubeMQListenerContainerFactory`. | | `autoStartup` | `String` | `""` | Whether the listener auto-starts (resolved as boolean). | | `errorHandler` | `String` | `""` | Bean name of a custom `ErrorHandler`. | | `id` | `String` | `""` | Unique listener endpoint identifier. | ```java title="EventListenerExample.java" @KubeMQEventListener(channels = "${kubemq.channels.orders}", group = "order-group") public void onOrder(EventMessageReceived event) { // process event } ``` ### @KubeMQEventStoreListener [#kubemqeventstorelistener] Receives `EventStoreMessageReceived` with replay support. The `subscriptionType` controls the replay start position. | Attribute | Type | Default | Description | | ------------------- | ---------- | -------------- | -------------------------------------------------------------------------------------- | | `channels` | `String[]` | *(required)* | Channel names to subscribe to. Supports SpEL and placeholders. | | `group` | `String` | `""` | Consumer group for shared subscriptions. | | `subscriptionType` | `String` | `StartNewOnly` | Replay start type (see values below). | | `subscriptionValue` | `String` | `0` | Value for the subscription type, e.g. sequence number or timestamp (resolved as long). | | `concurrency` | `String` | `""` | Number of concurrent processors. | | `containerFactory` | `String` | `""` | Bean name of a custom `KubeMQListenerContainerFactory`. | | `autoStartup` | `String` | `""` | Whether the listener auto-starts. | | `errorHandler` | `String` | `""` | Bean name of a custom `ErrorHandler`. | | `id` | `String` | `""` | Unique listener endpoint identifier. | `subscriptionType` accepts one of: `StartNewOnly`, `StartFromFirst`, `StartFromLast`, `StartAtSequence`, `StartAtTime`, `StartAtTimeDelta`. ```java title="EventStoreListenerExample.java" @KubeMQEventStoreListener( channels = "events_store.orders", subscriptionType = "StartFromFirst" ) public void replayOrders(EventStoreMessageReceived event) { // replay from the first stored event } ``` ### @KubeMQQueueListener [#kubemqqueuelistener] Polls `QueueMessageReceived` (single) or `List` (batch mode) from one or more channels. | Attribute | Type | Default | Description | | ------------------- | ---------- | ------------ | ----------------------------------------------------------------- | | `channels` | `String[]` | *(required)* | Channel names to poll from. Supports SpEL and placeholders. | | `concurrency` | `String` | `""` | Number of concurrent poll loops. | | `pollTimeout` | `String` | `""` | Max seconds to wait for messages per poll cycle. | | `maxPollMessages` | `String` | `""` | Max messages returned per poll. | | `visibilityTimeout` | `String` | `""` | Seconds received messages stay hidden from other consumers. | | `autoAck` | `String` | `""` | Whether messages are acknowledged automatically on receipt. | | `batch` | `String` | `false` | Whether the method receives a `List` batch. | | `containerFactory` | `String` | `""` | Bean name of a custom `KubeMQListenerContainerFactory`. | | `autoStartup` | `String` | `""` | Whether the listener auto-starts. | | `errorHandler` | `String` | `""` | Bean name of a custom `ErrorHandler`. | | `id` | `String` | `""` | Unique listener endpoint identifier. | ```java title="QueueListenerExample.java" @KubeMQQueueListener(channels = "queues.orders", autoAck = "true") public void processOrder(QueueMessageReceived msg) { // process queue message } ``` ### @KubeMQCommandHandler [#kubemqcommandhandler] Handles one command channel (point-to-point). Returns `boolean` (simple mode) or `CommandResponseMessage` (full control). Uses singular `channel`. | Attribute | Type | Default | Description | | ------------------ | -------- | ------------ | --------------------------------------------------------------------------- | | `channel` | `String` | *(required)* | The single command channel to subscribe to. Supports SpEL and placeholders. | | `group` | `String` | `""` | Consumer group for shared subscriptions. | | `concurrency` | `String` | `""` | Number of concurrent command processors. | | `containerFactory` | `String` | `""` | Bean name of a custom `KubeMQListenerContainerFactory`. | | `autoStartup` | `String` | `""` | Whether the handler auto-starts. | | `errorHandler` | `String` | `""` | Bean name of a custom `ErrorHandler`. | | `id` | `String` | `""` | Unique listener endpoint identifier. | ```java title="CommandHandlerExample.java" @KubeMQCommandHandler(channel = "commands.device-control", group = "handlers") public boolean handleCommand(CommandMessageReceived cmd) { return true; // executed successfully } ``` ### @KubeMQQueryHandler [#kubemqqueryhandler] Handles one query channel (point-to-point). Must return a `QueryResponseMessage`. Uses singular `channel`. | Attribute | Type | Default | Description | | ------------------ | -------- | ------------ | ------------------------------------------------------------------------- | | `channel` | `String` | *(required)* | The single query channel to subscribe to. Supports SpEL and placeholders. | | `group` | `String` | `""` | Consumer group for shared subscriptions. | | `concurrency` | `String` | `""` | Number of concurrent query processors. | | `containerFactory` | `String` | `""` | Bean name of a custom `KubeMQListenerContainerFactory`. | | `autoStartup` | `String` | `""` | Whether the handler auto-starts. | | `errorHandler` | `String` | `""` | Bean name of a custom `ErrorHandler`. | | `id` | `String` | `""` | Unique listener endpoint identifier. | ```java title="QueryHandlerExample.java" @KubeMQQueryHandler(channel = "queries.user-lookup") public QueryResponseMessage handleQuery(QueryMessageReceived query) { return QueryResponseMessage.builder() .queryReceived(query) .body(lookupResult) .isExecuted(true) .build(); } ``` ## Spring Cloud Stream Binder [#spring-cloud-stream-binder] The `kubemq-spring-cloud-stream-binder` module registers a binder named `kubemq` (declared in `META-INF/spring.binders`). It binds three KubeMQ patterns, selected per binding via the `pattern` consumer/producer property. For the full programming-model walkthrough, see [Spring Cloud Stream Binder](/integrations/spring-boot/how-to/spring-cloud-stream-binder). `KubeMQPattern` values: | Value | Maps to | | -------------- | ------------------------------------ | | `EVENTS` | Fire-and-forget Events. | | `EVENTS_STORE` | Persistent, replayable Events Store. | | `QUEUES` | Durable point-to-point Queues. | Consumer properties are bound from `KubeMQConsumerProperties` under `spring.cloud.stream.kubemq.bindings..consumer.*` (or `spring.cloud.stream.kubemq.default.consumer.*` for all bindings): | Property | Type | Default | Description | | -------------------------- | ----------------- | -------------- | ------------------------------------------------------------- | | `pattern` | `KubeMQPattern` | `EVENTS` | Messaging pattern for this binding. | | `eventsStoreType` | `EventsStoreType` | `StartNewOnly` | Events Store replay start type. | | `eventsStoreSequenceValue` | `long` | `0` | Value for the Events Store start type (e.g. sequence number). | | `pollMaxMessages` | `int` | `1` | Max messages returned per poll (Queues). | | `pollWaitTimeoutInSeconds` | `int` | `5` | Poll wait timeout in seconds (Queues). | | `visibilitySeconds` | `int` | `30` | Visibility timeout in seconds (Queues). | | `autoAckMessages` | `boolean` | `false` | Whether queue messages are auto-acknowledged. | ```yaml title="application.yml" spring: cloud: function: definition: uppercase stream: default-binder: kubemq bindings: uppercase-in-0: destination: spring-scs.events.in group: scs-events-demo uppercase-out-0: destination: spring-scs.events.out kubemq: default: consumer: pattern: EVENTS producer: pattern: EVENTS kubemq: address: ${KUBEMQ_ADDRESS:localhost:50000} client-id: spring-scs-events ``` Set `pattern: QUEUES` (with the poll/visibility consumer properties) or `pattern: EVENTS_STORE` (with `eventsStoreType` / `eventsStoreSequenceValue`) to bind the other patterns. ## Actuator Endpoints [#actuator-endpoints] When `kubemq.health.enabled` is `true`, the starter contributes a `kubemq` health indicator to `/actuator/health`. It pings all three clients (`PubSubClient`, `QueuesClient`, `CQClient`) and caches the result for `kubemq.health.cache-duration`. | Endpoint | Output | | ------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------- | | `/actuator/health` | `kubemq` component reports `UP` with `host` and `version` details when all clients respond to a ping, or `DOWN` with an `error` detail on failure. | | `/actuator/kubemq` | Overall `status` (`connected` when all clients are reachable, otherwise `degraded`) plus a per-client map (`pubsub`, `queues`, `cq`). | The `KubeMQEndpoint` (`/actuator/kubemq`) reports each client as a `ClientStatus` record with `status`, `host`, `version`, and `errorType`: ```json title="GET /actuator/kubemq" { "status": "connected", "clients": { "pubsub": { "status": "connected", "host": "kubemq:50000", "version": "v3.x", "errorType": null }, "queues": { "status": "connected", "host": "kubemq:50000", "version": "v3.x", "errorType": null }, "cq": { "status": "connected", "host": "kubemq:50000", "version": "v3.x", "errorType": null } } } ``` Expose the endpoint through standard Actuator configuration: ```yaml title="application.yml" management: endpoints: web: exposure: include: health, kubemq endpoint: health: show-details: always ``` ## Related [#related]