diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 0000000..3147d39
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1,240 @@
+# AGENTS.md
+
+Coding rules and project conventions for AI agents and human contributors working on
+**spring-cloud-stream-starter-request-reply** (artifactId `spring-cloud-stream-starter-request-reply`,
+groupId `community.solace.spring.cloud`). This file applies to this repository only.
+
+The library adds synchronous request/reply and request/multi‑reply semantics on top of Spring Cloud
+Stream, primarily for the Solace binder. When in doubt about behaviour, read the source under
+`src/main/java/community/solace/spring/cloud/requestreply` — it is the source of truth, not this file.
+
+## 1. Technology stack
+
+- **Language:** Java 17 (`17`). Do not use language features beyond Java 17.
+- **Build:** Maven, with `spring-boot-starter-parent` as the parent POM. It is a **library / Spring Boot
+ starter**, not an application: the `spring-boot-maven-plugin` repackage goal is disabled and the POM is
+ flattened for publishing.
+- **Frameworks:** Spring Boot, Spring Cloud Stream, Spring Cloud Function, Spring Integration, Spring
+ Messaging.
+- **Reactive:** Project Reactor (`Flux`, `Mono`, `FluxSink`) for multi‑reply streaming.
+- **Observability:** Micrometer (`micrometer-core` for metrics, `context-propagation` for tracing/MDC).
+- **Solace:** Solace `sol-jcsmp` types (`SDTStream`, `Destination`) are used **optionally** — the Solace
+ header parser is only registered when `com.solacesystems.jcsmp.Destination` is on the classpath.
+- **Testing:** JUnit 5, Spring Boot Test, `spring-cloud-stream-test-binder`, Mockito, Reactor Test,
+ Awaitility.
+- Do not add heavyweight dependencies. Solace types must stay optional so the library keeps working with
+ other binders (e.g. the test binder).
+
+## 2. Project structure
+
+```
+src/main/java/community/solace/spring/cloud/requestreply/
+├── service/
+│ ├── RequestReplyService.java # public API interface (requester side)
+│ ├── RequestReplyServiceImpl.java # requester implementation (correlation, timeout, executor)
+│ ├── ResponseHandler.java # per-request bookkeeping, dedup, completion latch
+│ ├── RequestReplyFunctionRegistrar.java # registers a reply consumer per bindingMapping
+│ ├── RequestReplyAutoConfiguration.java # main auto-configuration
+│ ├── MessageConverter.java
+│ ├── header/
+│ │ ├── RequestReplyMessageHeaderSupportService.java # responder-side wrap/wrapList/wrapFlux helpers
+│ │ └── parser/ # ordered Message/MessageHeaders parsers (correlationId,
+│ │ │ # destination, replyTo, replyIndex, totalReplies, errorMessage)
+│ │ ├── SolaceHeaderParser.java SpringHeaderParser.java HttpHeaderParser.java ...
+│ ├── logging/ # RequestReplyLogger + DefaultRequestReplyLogger
+│ └── messageinterceptor/ # RequestSendingInterceptor, ReplyWrappingInterceptor
+├── config/
+│ ├── RequestReplyProperties.java # @ConfigurationProperties("spring.cloud.stream.requestreply")
+│ ├── BinderMappings.java # a single bindingMapping entry
+│ ├── logging/LoggerAutoConfiguration.java
+│ └── messageinterceptor/ # Noop*Interceptor beans + MessageInterceptorAutoConfiguration
+├── env/ # ${replyTopicWithWildcards|...} property source
+├── util/ # MessageChunker, CheckedExceptionWrapper
+└── exception/ # RequestReplyException
+src/main/resources/META-INF/
+├── spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports # auto-config registration
+└── spring.factories # EnvironmentPostProcessor registration
+examples/ # standalone runnable example apps (own POMs)
+doc/ # protocol diagrams and AsyncAPI spec
+```
+
+- Auto‑configuration classes are registered in
+ `META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports`. Any new
+ auto‑configuration must be added there (never rely on component scan alone from a starter).
+- The `${replyTopicWithWildcards|...}` `EnvironmentPostProcessor` is registered in `spring.factories`.
+- Modules under `examples/` are independent Maven projects; keep them compiling but do not couple library
+ code to them.
+
+## 3. Coding standards & conventions
+
+Formatting is defined in `.editorconfig` (IntelliJ scheme). Key rules:
+
+- **Indent:** 4 spaces, no tabs. **Line ending:** LF. **Encoding:** UTF‑8. Trailing whitespace trimmed.
+ Files do **not** end with a final newline (`insert_final_newline = false`).
+- **Import order:** `java.**`, then `javax.**`, then everything else, then `org.springframework.**`, then
+ static imports (`$*`), each group separated by a blank line. No wildcard/on‑demand imports (threshold
+ 999) — use single‑class imports.
+- **Braces:** always use braces, even for single‑statement `if`/`for`/`while` (`*_brace_force = always`).
+- **Naming:** implementation classes end in `Impl` (e.g. `RequestReplyServiceImpl`); test classes end in
+ `Tests`.
+- Prefer `@Nullable` (Spring) on nullable returns/params, as existing parser code does.
+- Public API methods carry Javadoc; keep it accurate when you change signatures or behaviour.
+- Do not introduce Lombok in `src/main` (it is not a main dependency here; the examples use it, main code
+ does not).
+- Update `CHANGELOG.md` (Keep a Changelog format, semantic versioning) for any user‑visible change.
+
+## 4. Request/reply model & public API contract
+
+The two public entry points are `RequestReplyService` (requester) and
+`RequestReplyMessageHeaderSupportService` (responder). Treat their signatures and semantics as a stable
+contract; breaking changes require a major version bump and a CHANGELOG entry.
+
+**Requester (`RequestReplyService`):** every method is generic in request `Q` and response `A`, takes an
+`expectedClass` and a `Duration timeoutPeriod`, and has an overload accepting
+`Map additionalHeaders`. Flavours:
+
+- `requestAndAwaitReplyTo{Topic,Binding}` → blocking, returns `A`, declares `InterruptedException`,
+ `TimeoutException`, `RemoteErrorException`.
+- `requestReplyTo{Topic,Binding}` → non‑blocking, returns `CompletableFuture` (single response).
+- `requestReplyTo{Topic,Binding}Reactive` → returns `Flux` (zero to N responses).
+
+`…Topic` variants resolve the binding by matching the destination against
+`bindingMapping[].topicPatterns` (first match wins; no match → `IllegalArgumentException`). `…Binding`
+variants send to the binding's `-out-0` destination.
+
+**Responder (`RequestReplyMessageHeaderSupportService`):**
+
+- `wrap(fn, applicationExceptions...)` → single `Message` response. Returning `null` drops the message.
+- `wrapList(fn, bindingName, applicationExceptions...)` → `List` of known size.
+- `wrapFlux(fn, bindingName[, groupTimeout])` → streaming `Flux` of unknown size.
+
+Invariants the wrappers must preserve:
+
+- copy the **correlationId** onto every reply,
+- set the reply **destination** (`BinderHeaders.TARGET_DESTINATION`) from the request's reply‑to header,
+ applying `variableReplacements`,
+- set `totalReplies` and `replyIndex` headers for multi‑reply, and always emit a terminal (empty) message
+ to close the stream,
+- forward a matching `applicationException` as an error reply (`errorMessage` header) which the requester
+ surfaces as `RemoteErrorException` (or an errored `Flux`),
+- copy any headers listed in `copyHeadersOnWrap`.
+
+**Reply routing:** for each `bindingMapping`, `RequestReplyFunctionRegistrar` registers a
+`Consumer>` bean named after the binding, wired to `-in-0`, delegating to
+`RequestReplyServiceImpl.onReplyReceived`. Never register this consumer in a way that runs in sliced test
+contexts (see §6). If a bean with the binding name already exists, do not override it.
+
+## 5. Testing
+
+- Run the full suite with `mvn verify`. Do not use `--offline` unless dependencies are already cached.
+- Tests live under `src/test/java/...` mirroring the main package layout. Unit tests end in `Tests`;
+ Spring‑context / sample‑app integration tests live under `sampleapps/` and `integration/` and use the
+ `spring-cloud-stream-test-binder`.
+- Use JUnit 5 + AssertJ/Mockito; use Awaitility (already a dependency) for async assertions rather than
+ `Thread.sleep`. Use Reactor Test (`StepVerifier`) for `Flux`/`Mono` behaviour.
+- Any change to header parsing, deduplication (`ResponseHandler`), grouping (`MessageChunker`), or the
+ wrap helpers must come with tests — these are the correctness‑critical parts.
+- The starter must remain loadable **and** excludable: keep the tests that verify
+ `@ImportAutoConfiguration(exclude = RequestReplyAutoConfiguration.class)` and
+ `spring.autoconfigure.exclude` behaviour green.
+- Do not require a live Solace broker for `src/test`; use the test binder. (The `examples/` apps may point
+ at a public broker, but they are not part of the unit test suite.)
+
+## 6. Project-specific patterns & invariants
+
+- **Solace is optional.** Guard Solace‑specific beans/parsers with `@ConditionalOnClass` (as
+ `RequestReplyAutoConfiguration` does for `SolaceHeaderParser`). Never make core request/reply logic hard‑
+ depend on `sol-jcsmp`.
+- **Reply consumers are contributed by an `ImportBeanDefinitionRegistrar`, not an
+ `ApplicationContextInitializer`.** Spring Boot applies initializers to *every* context (including sliced
+ test contexts) and they cannot be excluded, which previously broke `@JsonTest`/`@WebMvcTest` slices. Keep
+ this registration inside `RequestReplyFunctionRegistrar` imported by the auto‑configuration so that
+ excluding the auto‑configuration also disables the consumers.
+- **Header access goes through the ordered parser chain**, never by reading a hard‑coded header key inline.
+ To support a new binder or header convention, add a parser bean and order it with `@Order` (lower =
+ higher priority). Header name constants live in `SpringHeaderParser` (`totalReplies`, `replyIndex`,
+ `groupedMessages`, `groupedContentType`, `errorMessage`) — reuse them.
+- **Correlation ids** are generated as a random UUID unless the request already carries one.
+- **Reply‑to uniqueness:** reply topics should be process‑unique. Use `${replyTopicWithWildcards|uuid}`
+ (generated once at process start) — not `${random.uuid}` (new value per reference). The requester listens
+ on the wildcarded topic via `${replyTopicWithWildcards||}`.
+- **Deduplication** (`ResponseHandler`) is by `replyIndex`, backed by a `BitSet`; terminal (finish/error)
+ messages bypass dedup. The unknown/streaming growth bound is read as a **JVM system property**
+ `spring.cloud.stream.requestreply.dedup.maxBitsWhenUnknown` (via `Integer.getInteger`, default 100000) —
+ it is **not** a Spring `@ConfigurationProperties` value, so it is set with `-D`, not `application.yaml`.
+- **Grouping thresholds** are load‑bearing constants: flush at 1 MB, at 10 000 messages, or after the group
+ timeout (default 200 ms). Changing them affects broker behaviour and interop — do so deliberately and
+ document it.
+- **Threading & context:** requests run on a shared executor wrapped with
+ `ContextExecutorService`/`ContextSnapshotFactory` so Micrometer context (trace id, MDC) propagates across
+ every pipeline stage. Preserve the executor‑wrapping approach; do not replace it with per‑task snapshots.
+- **State is in memory** (`PENDING_RESPONSES`), so the library is not fail‑safe or horizontally scalable
+ for a single request. Do not silently introduce assumptions that a reply may be handled by a different
+ instance than the one that sent the request.
+- **Configuration property namespace** is `spring.cloud.stream.requestreply`. Add new options to
+ `RequestReplyProperties` / `BinderMappings` with getters/setters and document them in the README property
+ table. Unknown YAML keys under a binding mapping are silently ignored by relaxed binding — do not rely on
+ keys that have no field.
+
+## 7. Example
+
+A minimal requester + responder pair. Configuration ties the binding name across
+`spring.cloud.function.definition`, `requestreply.bindingMapping`, and the `-in-0`/`-out-0` bindings.
+
+```yaml
+spring:
+ cloud:
+ function:
+ definition: temperatureQuery
+ stream:
+ requestreply:
+ bindingMapping:
+ - binding: temperatureQuery
+ replyTopic: reply/temperature/@project.artifactId@_${HOSTNAME}_${replyTopicWithWildcards|uuid}
+ topicPatterns:
+ - request/temperature/.*
+ bindings:
+ temperatureQuery-in-0:
+ destination: ${replyTopicWithWildcards|temperatureQuery|*}
+ contentType: "application/json"
+ binder: solace
+ temperatureQuery-out-0:
+ binder: solace
+```
+
+```java
+// Requester
+SensorReading reading = requestReplyService.requestAndAwaitReplyToTopic(
+ request,
+ "request/temperature/celsius/livingroom",
+ SensorReading.class,
+ Duration.ofSeconds(30)
+);
+
+// Responder (Spring Cloud Function bean)
+@Bean
+public Function, Message> temperatureQuery(
+ RequestReplyMessageHeaderSupportService headerSupport
+) {
+ return headerSupport.wrap(req -> {
+ SensorReading r = new SensorReading();
+ r.setTemperature(21.5);
+ return r; // return null to drop; throw a forwarded exception to send an error reply
+ }, MyBusinessException.class);
+}
+```
+
+## Related projects
+
+Only public ecosystem projects are relevant here — do not reference private/internal systems:
+
+- [Spring Cloud Stream](https://spring.io/projects/spring-cloud-stream) — the messaging abstraction this
+ library builds on.
+- [Spring Cloud Function](https://spring.io/projects/spring-cloud-function) — the functional model used for
+ responders and reply consumers.
+- [Spring Boot](https://spring.io/projects/spring-boot) — auto‑configuration and starter mechanics.
+- [Solace Spring Cloud / PubSub+ binder](https://github.com/SolaceProducts/solace-spring-cloud) — the
+ primary target binder.
+- [Project Reactor](https://projectreactor.io/) — reactive types for multi‑reply.
+- [Micrometer Context Propagation](https://docs.micrometer.io/context-propagation/reference/) — tracing/MDC
+ propagation across threads.
diff --git a/README.md b/README.md
index 463151a..99a0b1d 100644
--- a/README.md
+++ b/README.md
@@ -1,10 +1,52 @@
-# `spring-boot-starter-request-reply`
-## Description
-This Spring Boot Starter provides request-reply functionality for [Spring Cloud Stream Binders].
+# Spring Cloud Stream Request/Reply
+
+> Synchronous **request/reply** (and request / multi‑reply) semantics on top of
+> [Spring Cloud Stream](https://spring.io/projects/spring-cloud-stream) — primarily for the
+> [Solace PubSub+](https://github.com/SolaceProducts/solace-spring-cloud) binder, but pluggable for others.
+
+[](https://central.sonatype.com/artifact/community.solace.spring.cloud/spring-cloud-stream-starter-request-reply)
+[](https://github.com/solacecommunity/spring-cloud-stream-request-reply/actions/workflows/validate.yml)
+[](LICENSE)
+
+## Overview
+
+Message brokers are inherently asynchronous and fire‑and‑forget: you publish a message to a topic and
+you do not, by default, get an answer back. Many use cases, however, are naturally
+**request/reply** — "ask a question, wait for the answer" — for example querying the last sensor
+reading for a location, or fanning a query out to many responders and collecting their results.
+
+This Spring Boot starter adds that request/reply layer on top of Spring Cloud Stream. As the caller you
+get a plain, synchronous (or reactive) method call; under the hood the library:
+
+- generates a **correlation id** for the request,
+- publishes it to your request destination with a **reply‑to** topic that is unique to your process,
+- **correlates** the incoming reply(ies) back to the original call, and
+- returns the result to you (blocking, as a `CompletableFuture`, or as a reactive `Flux`) — or throws a
+ `TimeoutException` if no answer arrives within the timeout you specified.
+
+It supports both **single response** ("one question → one answer") and **request / multi‑reply**
+("one question → zero to N answers") patterns, with optional error propagation from the responder back
+to the caller.
+
+## Features
+
+- Synchronous request/reply as a single method call (`requestAndAwaitReplyToTopic` / `…ToBinding`).
+- Non‑blocking variants returning `CompletableFuture` or reactive `Flux`.
+- **Request / multi‑reply**: receive zero to N answers for one request.
+- Per‑request **timeouts** and automatic cleanup of in‑flight bookkeeping.
+- Automatic **correlation id** generation and reply‑to routing.
+- Helper methods (`wrap`, `wrapList`, `wrapFlux`) for the responder side that set the reply headers for
+ you and can **forward selected exceptions** back to the requester.
+- **Message grouping** for large multi‑reply result sets to reduce broker/header overhead.
+- **Reply deduplication** by `replyIndex` to survive duplicate delivery (e.g. broker reconnects).
+- **Micrometer context propagation** (tracing / MDC) across the internal asynchronous pipeline.
+- Pluggable **message / header parsers** to adapt to binders other than Solace.
+- Customizable **logging** and request/reply **message interceptors**.
-## Spring Cloud Version Compatibility
+## Compatibility
-Consult the table below to determine which version you need to use:
+The starter builds on Spring Boot, Spring Cloud Stream and the Solace `sol-jcsmp` client. Pick the
+version that matches your Spring Cloud release train:
| Spring Cloud | spring-cloud-stream-starter-request-reply | Spring Boot | sol-jcsmp |
|--------------|-------------------------------------------|-------------|-----------|
@@ -25,9 +67,11 @@ Consult the table below to determine which version you need to use:
| 2023.0.2 | 5.1.3 | 3.3.0 | 10.23.0 |
| 2023.0.1 | 5.1.2 | 3.2.5 | 10.23.0 |
-## Usage
-### Dependencies
-To enable the request/reply functionality, please add the following section to your Maven pom.xml
+Java 17+ is required.
+
+## Getting started
+
+### 1. Add the dependency
```xml
@@ -37,31 +81,144 @@ To enable the request/reply functionality, please add the following section to y
```
+The starter is auto‑configured; adding it to the classpath is enough to expose the
+`RequestReplyService` and `RequestReplyMessageHeaderSupportService` beans. You still need a Spring
+Cloud Stream binder on the classpath (e.g. the Solace binder) and the usual binder configuration.
+
+### 2. Send a request (requester side)
+
+Autowire `RequestReplyService` and call it:
+
+```java
+SensorReading response = requestReplyService.requestAndAwaitReplyToTopic(
+ reading, // the request payload
+ "last_value/temperature/celsius/" + location, // where to send the request
+ SensorReading.class, // how to map the reply
+ Duration.ofSeconds(30) // give up after 30s
+);
+```
+
+### 3. Reply to a request (responder side)
+
+A responder is an ordinary Spring Cloud Function; wrap it with
+`RequestReplyMessageHeaderSupportService` so the correct reply headers are set automatically:
+
+```java
+@Bean
+public Function, Message> responseToRequest(
+ RequestReplyMessageHeaderSupportService headerSupport
+) {
+ return headerSupport.wrap(request -> {
+ SensorReading response = new SensorReading();
+ response.setTemperature(21.5);
+ return response;
+ });
+}
+```
+
+The [`examples/`](examples) directory contains full runnable applications for the requester and
+responder sides, custom logging and custom reply‑to header handling.
+
+## How it works
+
+The request destination, the binding, and the reply‑to topic are wired together through three
+configuration keys. Understanding how they relate makes the configuration below straightforward.
+
+### Correlation
+
+Each request carries a **correlation id**. If your request is a plain payload, the library generates a
+random UUID; if it is already a `org.springframework.messaging.Message` that carries a correlation id,
+that id is reused. Every reply must echo the correlation id so the requester can match it to the
+pending call. Correlation ids and other metadata are read from messages by an ordered chain of
+**header parsers** (see [Extending to other binders](#extending-to-other-binders)).
+
+### Reply‑to and dynamic reply topics
+
+The requester tells the responder where to answer by putting a **reply‑to** topic into the outgoing
+message (`spring.cloud.stream.requestreply.bindingMapping[].replyTopic`). This topic should be unique
+per process so that replies come back only to the instance that asked. Best practice:
+
+- include the `HOSTNAME` to make debugging easier, and
+- include a **process‑stable UUID** via `${replyTopicWithWildcards|uuid}`. Do **not** use Spring's
+ `${random.uuid}` here — it produces a new UUID on every reference. `${replyTopicWithWildcards|uuid}`
+ is generated once at process start.
+
+Because a reply topic may contain `{placeholder}` segments that the responder substitutes before
+answering (see [Variable replacement](#variable-replacement)), the requester must listen on a
+**wildcarded** version of the topic. The `${replyTopicWithWildcards||*}` placeholder takes the
+`replyTopic` of the named binding and replaces every `{placeholder}` with the wildcard you pass (`*`
+for Solace):
+
+
+
+### Routing a request through a binding
+
+When you call `requestAndAwaitReplyToTopic(...)` / `requestReplyToTopicReactive(...)`, the request topic
+is matched against every `bindingMapping[].topicPatterns` (regular expressions); the **first match**
+selects the binding.
+
+
+
+The selected `bindingMapping[].binding` is used to look up `…-out-0` so the library knows which
+`binder`, `contentType`, etc. to use for sending. (When you send to a topic, any configured
+`…-out-0.destination` is ignored — the topic you passed wins.)
+
+
+
+For each configured `bindingMapping`, the library also registers a reply consumer on `…-in-0` so that
+incoming replies are routed back to the pending request. You do **not** need to declare this consumer
+function yourself — but you **must** list the binding name in `spring.cloud.function.definition`.
+
+
+
+### Timeouts
+
+Every method takes a `Duration timeoutPeriod`. The request is sent and the reply awaited on a shared
+executor; if no (final) reply arrives in time the pending request is aborted and a `TimeoutException`
+is raised. Bookkeeping for the request is always cleaned up on success, error or timeout.
+
+### Single vs. multi response
+
+- **Single response** — you expect exactly one answer. Use `requestAndAwaitReply*` (blocking) or
+ `requestReplyTo*` (returns a `CompletableFuture`).
+- **Multi response** — you expect zero to N answers. Use `requestReplyTo*Reactive`, which returns a
+ `Flux`. The responder signals completion with a terminal (empty) message and the total number of
+ replies, so the requester knows when the stream is done.
-### Usage
+For large multi‑reply result sets, replies can be **grouped**: instead of one broker message per
+answer, the responder packs several answers into a single message. Grouping is enabled automatically
+for non‑`Message` requests, or explicitly via the `groupedMessages` header
+(`SpringHeaderParser.GROUPED_MESSAGES`). A group is flushed when any of these is reached:
-#### For requester
+- the grouped message would exceed **1 MB**,
+- the group reaches **10 000** individual messages, or
+- the first message in the group is older than the responder's group timeout (default **200 ms**).
-If you want to send a request.
-You have to define a topic pattern that matches the topic where you send your requests to,
-to define the binding that should be used for these requests.
-With the binding, you define the binder, contentType, ...
-and the response address where the replier should send the response to.
+### Reply deduplication
-`spring.cloud.stream.requestreply.bindingMapping[n].binding` have to:
-- match an entry in: `spring.cloud.function.definition`
-- match `spring.cloud.stream.bindings.XX-in-0` where you defined the binder, contentType, ...
+In some operational scenarios (e.g. in‑place broker updates, short disconnects/reconnects) the same
+request may be delivered twice and the responder may therefore emit **duplicate replies**. The
+requester deduplicates incoming replies by `replyIndex`:
-`spring.cloud.stream.requestreply.bindingMapping[n].topicPatterns[m]`:
-- Is a list of RegEx patterns that will match against the destination of your requests.
-- If there are no matching patterns when execute `requestAndAwaitReplyToTopic()`/`requestReplyToTopic()` and `IllegalArgumentException` will be thrown.
-- If you only want to use `requestAndAwaitReplyToBinding()`/`requestReplyToBinding()` you don't need to give this configuration.
+- duplicate `replyIndex` values are processed only once (including range indices such as `"0-45"` used
+ for grouped replies),
+- terminal messages (finish/error) are always processed, even when they share a `replyIndex`.
-Please remember to define this binding as well in the `spring.cloud.function.definition`,
-otherwise you will not receive responses.
-But you don't need to create a bean for that, this will be generated by the library.
+When `totalReplies` is not yet known (streaming/unknown‑size patterns), numeric `replyIndex` values are
+still deduplicated up to a bounded bitmap size (see [Configuration](#configuration)).
+
+## Configuration
+
+A complete requester + responder configuration ties four things together for each binding:
+
+1. the binding name appears in `spring.cloud.function.definition`,
+2. `spring.cloud.stream.requestreply.bindingMapping[]` defines the `replyTopic` (and optional
+ `topicPatterns`),
+3. `spring.cloud.stream.bindings.-in-0` defines where replies are consumed, and
+4. `spring.cloud.stream.bindings.-out-0` defines the binder used to send.
+
+### Dynamic topics (send to a topic chosen at call time)
-##### Using dynamic topics
```yaml
spring:
cloud:
@@ -83,28 +240,32 @@ spring:
binder: solace
```
-###### single response
+Single response:
+
```java
- SensorReading response = requestReplyService.requestAndAwaitReplyToTopic(
- reading,
- "requestReply/request/last_value/temperature/celsius/" + location,
- SensorReading.class,
- Duration.ofSeconds(30)
- );
+SensorReading response = requestReplyService.requestAndAwaitReplyToTopic(
+ reading,
+ "requestReply/request/last_value/temperature/celsius/" + location,
+ SensorReading.class,
+ Duration.ofSeconds(30)
+);
```
-###### multiple responses
-Introduction into [Flux/project reactor](https://www.baeldung.com/reactor-core)
+Multiple responses (reactive):
+
```java
- Flux responses = requestReplyService.requestAndAwaitReplyToTopicReactive(
- reading,
- "requestReply/request/last_value/temperature/celsius/" + location,
- SensorReading.class,
- Duration.ofSeconds(30)
- );
+Flux responses = requestReplyService.requestReplyToTopicReactive(
+ reading,
+ "requestReply/request/last_value/temperature/celsius/" + location,
+ SensorReading.class,
+ Duration.ofSeconds(30)
+);
```
-##### Using static topics
+### Static topics (send to the binding's configured destination)
+
+Omit `topicPatterns` and configure `…-out-0.destination`, then address the request by **binding name**:
+
```yaml
spring:
cloud:
@@ -121,246 +282,218 @@ spring:
contentType: "application/json"
binder: solace
requestReplyRepliesDemo-out-0:
- destination: requestReply/request/last_value/temperature/celsius/livingroom,
+ destination: requestReply/request/last_value/temperature/celsius/livingroom
binder: solace
```
-###### single response
```java
- SensorReading response = requestReplyService.requestAndAwaitReplyToBinding(
- request,
- "requestReplyRepliesDemo",
- SensorReading.class,
- Duration.ofSeconds(30)
- );
+SensorReading response = requestReplyService.requestAndAwaitReplyToBinding(
+ request,
+ "requestReplyRepliesDemo",
+ SensorReading.class,
+ Duration.ofSeconds(30)
+);
```
-###### multiple responses
-Introduction into [Flux/project reactor](https://www.baeldung.com/reactor-core)
-```java
- Flux responses = requestReplyService.requestReplyToBindingReactive(
- request,
- "requestReplyRepliesDemo",
- SensorReading.class,
- Duration.ofSeconds(30)
- );
-```
+### Property reference
-[Full example](examples/request_reply_sending/src/main/java/community/solace/spring/cloud/requestreply/examples/sending/controller/RequestReplyController.java)
+All properties live under `spring.cloud.stream.requestreply`:
-##### How everything is related
+| Property | Type | Description |
+|----------|------|-------------|
+| `bindingMapping[].binding` | String | Binding name. Must appear in `spring.cloud.function.definition` and match `spring.cloud.stream.bindings.-in-0`/`-out-0`. |
+| `bindingMapping[].replyTopic` | String | Reply‑to topic placed on outgoing requests. Should be unique per process (host + process‑stable UUID). Required. |
+| `bindingMapping[].topicPatterns` | List<RegEx> | Patterns matched against the request destination in `requestReplyTo*Topic*` calls. First match wins. Not needed if you only use the `…ToBinding` methods. |
+| `variableReplacements` | Map<String,String> | `{key}` placeholders replaced with the mapped value in request and reply topics. |
+| `copyHeadersOnWrap` | List<String> | Additional request headers to copy onto the reply when using the `wrap*` helpers. |
-When using `requestAndAwaitReplyToBinding` / `requestReplyToTopic`
-The topic use in the code will be matched against all `spring.cloud.stream.requestreply.bindingMapping[].topicPatterns`,
-the fist hit will be used.
-
+Reply topic placeholders contributed by this starter (usable anywhere in the environment):
-The `spring.cloud.stream.requestreply.bindingMapping[].binding` of the matching section will be matched against
-`spring.cloud.stream.bindings[]` and always out-1 will be used.
-This is required to let the requestReply service know what `binder`, `contentType`, ... is to use.
-
+| Placeholder | Meaning |
+|-------------|---------|
+| `${replyTopicWithWildcards\|uuid}` | A UUID generated **once** at process start (unlike `${random.uuid}`). |
+| `${replyTopicWithWildcards\|\|}` | The named binding's `replyTopic` with every `{placeholder}` replaced by `` (`*` for Solace). Use this for `…-in-0.destination`. |
-The `spring.cloud.stream.requestreply.bindingMapping[].replyTopic` of the matching section will be placed in the outbound
-message to let the foreign service know where you expect the answer.
-In general, you want to use here a topic that is unique.
-Best practice is to put:
-- HOSTNAME into the topic to make debugging a little easier.
-- To ensure a unique inbox for your topic, include a UUID directly within it.
- Avoid using `${random.uuid}` from Spring, as this would generate a new UUID each time it is called.
- Instead, use `${replyTopicWithWildcards|uuid}` to get a static UUID that is generated at the start of the process.
+Dedup bitmap bound for unknown/streaming reply counts is read as a **JVM system property** (default
+`100000`), for example:
-
+```
+-Dspring.cloud.stream.requestreply.dedup.maxBitsWhenUnknown=100000
+```
-The requestReply service will iterate over `spring.cloud.stream.requestreply.bindingMapping`
-And create a Bean to consume messages received on `-in-0.destination` using the given binder.
-
+Any `replyIndex` (or range end) above this bound is not deduplicated.
-As soon as you use the `{StagePlaceholder}` feature, you cannot listen on:
-`requestReply/response/solace/{StagePlaceholder}/pub_sub_sending_K353456_315fd96b-b981-417b-be99-3be065c6611d`
-because the foreign side will replace `{StagePlaceholder}` against `p-pineapple`for example before responding.
-Therefore, you need to listen to:
-`requestReply/response/solace/*/pub_sub_sending_K353456_315fd96b-b981-417b-be99-3be065c6611d`
-This task will be done by: `${replyTopicWithWildcards|requestReplyRepliesDemo|*}`
-It will take `spring.cloud.stream.requestreply.bindingMapping[].replyTopic` of the section, matching the first parameter.
-And replace all `{someThing}` to the wildcard (second parameter) `*`
-
+## API reference
-#### For replier
-In general, if you want to respond to a message, you do not need this library.
-Instead,
-you can send the response to the topic specified in the reply-to header
-and replicate all headers from the request to the response.
+### `RequestReplyService`
+Autowire this bean to send requests. All methods are generic in the request type `Q` and response type
+`A`, take an `expectedClass` the reply is mapped to and a `Duration timeoutPeriod`, and have an overload
+that accepts a `Map additionalHeaders`.
-However, the methods `RequestReplyMessageHeaderSupportService.wrap`, `RequestReplyMessageHeaderSupportService.wrapList`
-and `RequestReplyMessageHeaderSupportService.wrapFlux`
-from this library can support in creating the response with properly setting the message headers,
-as well as with substituting variables in dynamic topics.
+**Single response**
-By default, the wrapping function will set the correlationId and reply destination headers of the message.
-In the case of multi responses, the header for totalReplies and replyIndex will be set as well.
-Additional headers can be configured to be copied from the request by
-setting `spring.cloud.stream.requestReply.copyHeadersOnWrap` accordingly, e.g.:
+| Method | Returns | Notes |
+|--------|---------|-------|
+| `requestAndAwaitReplyToTopic(request, requestDestination, expectedClass, timeout)` | `A` | Blocks. `requestDestination` is matched against `topicPatterns`. Any `-out-0.destination` is ignored. |
+| `requestAndAwaitReplyToBinding(request, bindingName, expectedClass, timeout)` | `A` | Blocks. Sends to the destination configured for the binding's `-out-0`. |
+| `requestReplyToTopic(request, requestDestination, expectedClass, timeout)` | `CompletableFuture` | Non‑blocking. Use only when you need parallel request/reply on the same thread. |
+| `requestReplyToBinding(request, bindingName, expectedClass, timeout)` | `CompletableFuture` | Non‑blocking. Use only when you need parallel request/reply on the same thread. |
-```properties
-spring.cloud.stream.requestReply.copyHeadersOnWrap=encoding,yetAnotherHeader
-```
+**Multi response (zero to N answers)**
+
+| Method | Returns | Notes |
+|--------|---------|-------|
+| `requestReplyToTopicReactive(request, requestDestination, expectedClass, timeout)` | `Flux` | Request destination matched against `topicPatterns`. |
+| `requestReplyToBindingReactive(request, bindingName, expectedClass, timeout)` | `Flux` | Sends to the binding's `-out-0` destination. |
+
+The blocking methods declare `InterruptedException`, `TimeoutException` and `RemoteErrorException`;
+`RemoteErrorException` is thrown when the responder forwarded an application error (see below).
-The actual Wrapping can be used as in the example below:
+Blocking example that collects a list of answers:
-##### single response
```java
-public class PingPongConfig {
- @Bean
- public Function, Message> responseToRequest(
- RequestReplyMessageHeaderSupportService headerSupport
- ) {
- return headerSupport.wrap((request) -> {
- SensorReading response = new SensorReading();
- response.setFoo(1337);
-
- return response;
- });
- }
+@GetMapping(value = "/temperature/last_hour/{location}")
+public List requestMultiReplySample(@PathVariable("location") final String location) {
+ MyRequest request = new MyRequest();
+ request.setLocation(location);
+
+ return requestReplyService.requestReplyToTopicReactive(
+ request,
+ "last_hour/temperature/celsius/" + location,
+ SensorReading.class,
+ Duration.ofSeconds(30)
+ )
+ .collectList()
+ .block();
}
```
-[Full example](examples/request_reply_response/src/main/java/community/solace/spring/cloud/requestreply/examples/response/config/PingPongConfig.java)
-##### multiple responses, functional
+Non‑blocking example:
+
```java
-public class PingPongConfig {
- @Bean
- public Function, List>> responseMultiToRequestKnownSizeSolace(
- RequestReplyMessageHeaderSupportService headerSupport
- ) {
- return headerSupport.wrapList((request) -> {
- List responses = new ArrayList<>();
- responses.add(new SensorReading());
- // ....
-
- return responses;
- }, "responseMultiToRequestKnownSizeSolace-out-0");
- }
-}
+requestReplyService.requestReplyToTopicReactive(
+ request,
+ "last_hour/temperature/celsius/" + location,
+ SensorReading.class,
+ Duration.ofSeconds(30)
+ )
+ .subscribe(
+ sensorReading -> log.info("Got an answer: " + sensorReading),
+ throwable -> log.error("The request finished with error", throwable),
+ () -> log.info("The request finished")
+ );
```
-[Full example](examples/request_reply_response/src/main/java/community/solace/spring/cloud/requestreply/examples/response/config/PingMultiPongConfig.java)
-##### multiple responses, reactive
+### `RequestReplyMessageHeaderSupportService`
+
+For a pure responder you do not strictly need this library — you could copy the reply‑to header and
+correlation id onto your response yourself. The `wrap*` helpers do this for you: they set the
+correlation id and reply destination header, substitute `{placeholder}` variables, and (for multi
+responses) set the `totalReplies` and `replyIndex` headers. Additional request headers can be copied
+onto the reply via `spring.cloud.stream.requestreply.copyHeadersOnWrap`.
+
+**Single response** — return `null` from the wrapped function to drop the message (no reply sent):
+
```java
-public class PingPongConfig {
- @Bean
- public Function>, Flux>> responseMultiToRequestRandomSizeSolace(
- RequestReplyMessageHeaderSupportService headerSupport
- ) {
- return headerSupport.wrapFlux((request, responseSink) -> {
- try {
- while (yourBusinessLogic) { // Your business logic can submit 0 to N responses.
- responseSink.next(response);
- }
- responseSink.complete();
- } catch (Exception e) {
- responseSink.error(new IllegalArgumentException("Business error message", e));
- }
- }, "responseMultiToRequestRandomSizeSolace-out-0");
- }
+@Bean
+public Function, Message> responseToRequest(
+ RequestReplyMessageHeaderSupportService headerSupport
+) {
+ return headerSupport.wrap(request -> {
+ SensorReading response = new SensorReading();
+ response.setTemperature(21.5);
+ return response;
+ });
}
```
-[Full example](examples/request_reply_response/src/main/java/community/solace/spring/cloud/requestreply/examples/response/config/PingMultiPongConfig.java)
-##### error handling
-You might want to forward errors to requester.
-To forward errors, you only need to define 1 to N exception classes that should be forwarded to the requestor.
+**Multiple responses, known size** (`wrapList`) — pass the output binding name so grouping and content
+type can be resolved:
```java
-public class PingPongConfig {
- @Bean
- public Function, Message> responseToRequest(
- RequestReplyMessageHeaderSupportService headerSupport
- ) {
- return headerSupport.wrap((request) -> {
- SensorReading response = new SensorReading();
- response.setFoo(1337);
-
- return response;
- }, MyBusinessException.class, SomeOtherException.class);
- }
+@Bean
+public Function, List>> responseMultiToRequestKnownSize(
+ RequestReplyMessageHeaderSupportService headerSupport
+) {
+ return headerSupport.wrapList(request -> {
+ List responses = new ArrayList<>();
+ // ... add responses ...
+ return responses;
+ }, "responseMultiToRequestKnownSize-out-0");
}
```
-Input class validation and JSON parsing cannot be returned to the requester out of the box.
-Therefore, you need to do the validation on your own like:
+**Multiple responses, streaming/unknown size** (`wrapFlux`) — emit 0 to N responses through the sink:
```java
-public class PingPongConfig {
- @Qualifier("mvcValidator")
- private final Validator validator;
-
- private final ObjectMapper objectMapper;
-
- @Bean
- public Function, Message> responseToRequest(
- RequestReplyMessageHeaderSupportService headerSupport
- ) {
- return headerSupport.wrap((rawRequest) -> {
- SensorReading request = objectMapper.readValue(rawRequest.getPayload(), SensorReading.class);
- final DataBinder db = new DataBinder(request);
- db.setValidator(validator);
- db.validate();
-
-
- SensorReading response = new SensorReading();
- response.setFoo(1337);
-
- return response;
- }, MyBusinessException.class, SomeOtherException.class);
- }
+@Bean
+public Function>, Flux>> responseMultiToRequestRandomSize(
+ RequestReplyMessageHeaderSupportService headerSupport
+) {
+ return headerSupport.wrapFlux((request, responseSink) -> {
+ try {
+ while (moreData) { // your business logic can submit 0..N responses
+ responseSink.next(response);
+ }
+ responseSink.complete();
+ } catch (Exception e) {
+ responseSink.error(new IllegalArgumentException("Business error message", e));
+ }
+ }, "responseMultiToRequestRandomSize-out-0");
}
```
-##### Variable replacement
+**Forwarding errors to the requester.** Pass one or more exception classes to the `wrap*` helpers; if
+the wrapped function throws a matching exception, the error message is sent back to the requester, which
+then throws a `RemoteErrorException` (for streaming replies the error terminates the `Flux`):
+
+```java
+return headerSupport.wrap(request -> {
+ // ...
+ return response;
+}, MyBusinessException.class, SomeOtherException.class);
+```
-The requestor may include placeholders in the reply destination.
-These placeholders must be replaced before sending the response.
+> Input validation and JSON parsing errors cannot be forwarded automatically — perform validation
+> inside the wrapped function (e.g. with a `DataBinder`/`Validator`) and throw one of your forwarded
+> exception types.
-A possible use case for this feature is
-when there are multiple instances available to process the response for load-balancing or data center redundancy reasons.
-In such scenarios, it can be helpful for debugging purposes to know how messages are processed.
+## Advanced usage
-For instance, the request might contain the placeholder `{StagePlaceholder}` in the reply destination header.
+### Variable replacement
-Here’s an example configuration to replace the `{StagePlaceholder}` term with a unique identifying string:
+A requester may embed `{placeholder}` segments in the reply destination (useful, for example, to encode
+which instance/data center should process a load‑balanced reply). The responder substitutes them before
+answering:
```yaml
spring:
cloud:
- function:
- definition: requestReplyRepliesDemo
stream:
requestreply:
variableReplacements:
"{StagePlaceholder}": ${RCS_ENV_ROLE}-${RCS_CLUSTER}
```
-Those `variableReplacements` will be applied to request and reply topics.
+`variableReplacements` are applied to both request and reply topics.
-##### Configure custom logging
-In case the standard logging is not matching your expectations,
-you can define your own logging spring bean to customize logging behavior.
+### Custom logging
+
+Provide a `RequestReplyLogger` bean to override the default logging:
```java
@Configuration
public class CustomizedLoggerConfig {
-
@Bean
public RequestReplyLogger requestReplyLogger() {
return new CustomizedLogger();
}
}
```
-Example logger interface implementation:
+
```java
public class CustomizedLogger implements RequestReplyLogger {
-
@Override
public void logRequest(Logger logger, Level suggestedLevel, String suggestedLogMessage, Message message) {
logger.atLevel(Level.DEBUG).log("<<< {} {}", message.getPayload(), message.getHeaders());
@@ -368,8 +501,7 @@ public class CustomizedLogger implements RequestReplyLogger {
@Override
public void logReply(Logger logger, Level suggestedLevel, String suggestedLogMessage, long remainingReplies, Message message) {
- String payloadString = new String((byte[])message.getPayload());
- logger.atLevel(Level.DEBUG).log(">>> {} {} remaining replies: {}", payloadString, message.getHeaders(), remainingReplies);
+ logger.atLevel(Level.DEBUG).log(">>> {} remaining replies: {}", message.getPayload(), remainingReplies);
}
@Override
@@ -379,233 +511,74 @@ public class CustomizedLogger implements RequestReplyLogger {
}
```
-You find an entire example-application for the purpose of showing how to configure this under examples/customized_logging.
-
-##### Configure custom message interception
-If you need to modify a request-message before it is sent by the request side, you can define an interceptor bean
-for the interface "RequestSendingInterceptor".
-You can find an entire example tho this under "examples/customized_reply_to_header_sending".
-
-On the responder side, if you need to change a message while it is being wrapped,
-you can define an interceptor bean for the interface "ReplyWrappingInterceptor".
-You can find an entire example tho this under "examples/customized_reply_to_header_response".
-
-### API
-
-#### `RequestReplyService`
-
-The request/reply functionality
-provided by this starter can be used by autowiring the `RequestReplyService` which offers the following methods:
-
-#### for single response
+See [`examples/customized_logging`](examples/customized_logging) for a full application.
-If you expect exactly one response.
+### Message interceptors
-- The method `A requestAndAwaitReplyToTopic(Q request, String requestDestination, Class expectedResponseClass, Duration timeoutPeriod)`
- sends the specified request to the designated request destination,
- waits for the response, and maps it to the provided class as the return value.
- If a `-out-0.destination` is configured, it will be ignored.
+- Implement `RequestSendingInterceptor` (bean name `requestSendingInterceptor`) to modify a request
+ message before it is sent — for example to add or rewrite headers. See
+ [`examples/customized_reply_to_header_sending`](examples/customized_reply_to_header_sending).
+- Implement `ReplyWrappingInterceptor` (bean name `replyWrappingInterceptor`) to modify a reply while it
+ is being wrapped. Implement all three callbacks (payload, finishing/empty and error messages). See
+ [`examples/customized_reply_to_header_response`](examples/customized_reply_to_header_response).
-- The method `A requestAndAwaitReplyToBinding(Q request, String bindingName, Class expectedResponseClass, Duration timeoutPeriod)`
- sends the specified request to the destination configured for the `-out-0` of this binding,
- waits for the response, and maps it to the provided class as the return value.
- If a `-out-0.destination` is configured, it will be ignored.
+Both interfaces receive the binding name so you can behave differently per binding. If you do not
+provide your own, no‑op implementations are auto‑configured.
-- The method `CompletableFuture requestReplyToTopic(Q request, String requestDestination, Class expectedClass, Duration timeoutPeriod)`
- sends the specified request to the designated request destination.
- It returns a future that maps the received response to the provided class.
- Use this method only in rare edge cases
- where you need to execute multiple request-reply operations in parallel within the same thread.
+### Extending to other binders
-- The method `CompletableFuture requestReplyToBinding(Q request, String bindingName, Class expectedClass, Duration timeoutPeriod)`
- sends the specified request to the destination configured for the `-out-0` of this binding.
- It returns a future that maps the received response to the provided class.
- Use this method only in rare edge cases
- where you need to execute multiple request-reply operations in parallel within the same thread.
+The starter works out of the box with the
+[Solace binder](https://github.com/SolaceProducts/solace-spring-cloud) and the
+[TestSupportBinder](https://github.com/spring-cloud/spring-cloud-stream), and can be extended to other
+binders by providing message / header parser beans.
-#### for multi response
+When receiving a message the library must be able to determine the `correlationId`, `destination`,
+`replyTo`, `totalReplies` and `replyIndex`. Unless a binder adheres to Spring messaging standards, add
+parser beans and order them with `@Order` (lower value = higher priority).
-If you expect zero to N responses.
+Parser interfaces (root interface parses a `Message`, the `…Header…` variant parses `MessageHeaders`):
-- The method `Flux requestReplyToTopicReactive(Q request, String requestDestination, Class expectedClass, Duration timeoutPeriod)`
- sends the specified request to the designated request destination.
- It returns a reactive stream that maps the received responses to the provided class.
- Use this method when you expect multiple responses for a single question.
- If your response type is an array,
- it is advisable to send the elements as individual messages to avoid exceeding the message size limit.
+- `MessageCorrelationIdParser` / `MessageHeaderCorrelationIdParser`
+- `MessageDestinationParser` / `MessageHeaderDestinationParser`
+- `MessageReplyToParser` / `MessageHeaderReplyToParser`
+- `MessageTotalRepliesParser` / `MessageHeaderTotalRepliesParser`
+- `MessageReplyIndexParser` / `MessageHeaderReplyIndexParser`
+- `MessageErrorMessageParser` / `MessageHeaderErrorMessageParser`
-- The method `Flux requestReplyToBindingReactive(Q request, String bindingName, Class expectedClass, Duration timeoutPeriod)`
- sends the specified request to the destination configured for the `-out-0` of this binding.
- It returns a reactive stream that maps the received responses to the provided class.
- Use this method when you expect multiple responses for a single query.
- If your response type is an array,
- it is recommended to send the elements as individual messages to prevent exceeding the message size limit.
+Bundled implementations, in priority order:
-###### Example for blocking
+| Parser | Order | Provides |
+|--------|-------|----------|
+| `SolaceHeaderParser` | 200 | correlationId, destination, replyTo for the Solace binder |
+| `SpringCloudStreamHeaderParser` | 10000 | destination, totalReplies for Spring Cloud Stream headers |
+| `SpringIntegrationHeaderParser` | 20000 | correlationId for Spring Integration headers |
+| `BinderHeaderParser` | 30000 | destination for Spring Cloud Stream binder headers |
+| `SpringHeaderParser` | 40000 | replyTo, totalReplies, replyIndex, errorMessage for Spring Framework headers |
+| `HttpHeaderParser` | `LOWEST_PRECEDENCE` | correlationId per the HTTP header standard |
-A blocking request reply where you receive a list of answers.
-
-```java
- @GetMapping(value = "/temperature/last_hour/{location}")
- public List requestMultiReplySample(
- @PathVariable("location") final String location
- ) {
- MyRequest request = new MyRequest();
- request.setLocation(location);
-
- return requestReplyService.requestReplyToTopicReactive(
- request,
- "last_hour/temperature/celsius/" + location,
- SensorReading.class,
- Duration.ofSeconds(30)
- )
- .collectList()
- .block();
- }
-```
-
-###### Example for non-blocking
-
-A blocking request reply where you receive a list of answers.
-
-```java
- @GetMapping(value = "/temperature/last_hour/{location}")
- public void requestMultiReplySample(
- @PathVariable("location") final String location
- ) {
- MyRequest request = new MyRequest();
- request.setLocation(location);
-
- requestReplyService.requestReplyToTopicReactive(
- request,
- "last_hour/temperature/celsius/" + location,
- SensorReading.class,
- Duration.ofSeconds(30)
- )
- .subscribe(
- sensorReading -> log.info("Got an answer: " + sensorReading),
- throwable -> log.error("The request was finished with error", throwable),
- () -> log.info("The request was finished")
- );
- }
-```
+### Tracing and context propagation
-###### receive many answers
+The library forwards the Micrometer trace id from requester to responder so all spans share one trace.
+No special configuration is required beyond your normal tracing setup, e.g.:
-If your request is a `org.springframework.messaging.Message`, you can decide if every response should be
-transported as a single message or if a couple of messages should be grouped together.
-
-Message grouping can be enabled via:
-`requestMsg.setHeader(SpringHeaderParser.GROUPED_MESSAGES, true);`
-
-In case you request something else, will the request/reply lib add `GROUPED_MESSAGES=true` automatically.
-
-Until you require a separate header for each reply,
-it is recommended to use grouped messages instead of sending individual messages.
-Grouped messages improve reply speed by reducing the message header overhead and conserving broker resources.
-
-Messages will be grouped together until one of the following conditions is met:
-- The grouped message exceeds the 1MB limit
-- The group contains more than 10_000 individual messages
-- The first message in the group was sent more than 0.5 seconds ago (or a different threshold configured by the replier).
-
-
-#### `RequestReplyMessageHeaderSupportService`
-
-If a service only responds to requests,
-this library can still be used as it provides helper methods for receiving services.
-These methods wrap response functions using the `RequestReplyMessageHeaderSupportService`,
-which appropriately sets the Message's destination header.
-This header can then be accessed through Spring Cloud Functions.
-For example:
-
-```java
- @Bean
- @Autowired
- public Function, Message> reverse(RequestReplyMessageHeaderSupportService headerSupport) {
- return headerSupport.wrap((value) -> new StringBuilder(value).reverse().toString());
- }
-```
-
-Return `null` from the wrapped function to indicate that the message should be dropped.
-
-## Extensibility
-
-The Request/Reply Spring Boot Starter has been designed
-to work with both the [Solace Binder](https://github.com/SolaceProducts/solace-spring-cloud) and the [TestSupportBinder](https://github.com/spring-cloud/spring-cloud-stream/blob/main/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java),
-but can be extended to work with other binders as well.
-
-For that, the following beans must be provided to be able to adapt to another binder.
-
-### Message and MessageHeader parsers
-
-When receiving messages, the application must be able to determine correlationId, destination and replyTo properties attached to a message. Unless a particular binder adheres to spring messaging standards or requires differing headers for performance optimizations, additional message parsers (or related message header parsers) must be present. For proper prioritization the bean should be annotated with the `@Order` annotation. (see [@Order in Spring @Baeldung])
-
-
-
-The Starter includes the following message parser interfaces:
-
-- `MessageCorrelationIdParser` - root interface for parsing the correlation id from an incoming message
- - `MessageHeaderCorrelationIdParser` - extended interface for parsing the correlation id from an incoming message's MessageHeaders
-- `MessageDestinationParser` - root interface for parsing the destination from an incoming message
- - `MessageHeaderDestinationParser`- extended interface for parsing the destination from an incoming message's MessageHeaders
-- `MessageReplyToParser` - root interface for parsing the reply destination from an incoming message
- - `MessageHeaderReplyToParser` - extended interface for parsing the reply destination from an incoming message's MessageHeaders
-- `MessageTotalRepliesParser` - root interface for parsing the total replies for from an incoming multi response message
- - `MessageHeaderTotalRepliesParser` - extended interface for parsing the total replies from an incoming multi response message's MessageHeaders
-
-
-
-The starter also includes the following message parser implementations:
-
-- `SolaceHeaderParser` _Order: 200_
- implements `MessageHeaderCorrelationIdParser`, `MessageHeaderDestinationParser`, `MessageHeaderReplyToParser` for the Solace binder
-- `SpringCloudStreamHeaderParser` _Order 10000_
- implements `MessageHeaderDestinationParser`, `MessageTotalRepliesParser` for standard Spring Cloud Stream headers
-- `SpringIntegrationHeaderParser` _Order 20000_
- implements `MessageHeaderCorrelationIdParser` for standard Spring Integration headers
-- `BinderHeaderParser` _Order 30000_
- implements `MessageHeaderDestinationParser` for standard Spring Cloud Stream Binder headers
-- `SpringHeaderParser` _Order 40000_
- implements `MessageHeaderReplyToParser` for Spring Framework message header standards
-- `HttpHeaderParser` _Order LOWEST_PRECEDENCE_
- implements `MessageHeaderCorrelationIdParser` according to the HTTP header standard
-
-## Compatibility
-
-### Tracing
-
-The request reply lib will forward the traceId from micrometer to have all spans of requester and replier in the same tracing.
-
-No special configuration is required, only set the default configuration required for tracing.
```yaml
spring:
application:
name: the-name-of-your-micro-service
management:
- zipkin:
- tracing:
- endpoint: https://demo-zipkin.xxxx.net/api/v2/spans
- export:
- enabled: true
- tracing:
- sampling:
- probability: 1.0
+ tracing:
+ sampling:
+ probability: 1.0
logging:
- pattern: correlation=[${spring.application.name:},%X{traceId:-},%X{spanId:-}]
+ pattern: correlation=[${spring.application.name:},%X{traceId:-},%X{spanId:-}]
```
-#### Context propagation across asynchronous stages
-
-A request/reply call is processed on a dedicated executor, so the request is sent and the reply is
-awaited on a different thread than the caller. To keep tracing (and any other thread-local context
-such as the SLF4J `MDC`) consistent, this library propagates the
-[micrometer context](https://docs.micrometer.io/context-propagation/reference/) captured on the
-calling thread to that executor.
-
-Because the executor itself is wrapped (rather than each individual task), the context is restored
-for **every** stage of the internal pipeline as well as for stages that your application chains onto
-the returned `CompletableFuture`:
+A request/reply call is processed on a dedicated executor, so the request is sent and the reply awaited
+on a different thread than the caller. To keep tracing (and any other thread‑local context such as the
+SLF4J `MDC`) consistent, the library propagates the
+[Micrometer context](https://docs.micrometer.io/context-propagation/reference/) captured on the calling
+thread. Because the executor itself is wrapped, the context is restored for **every** stage of the
+internal pipeline as well as for stages your application chains onto the returned `CompletableFuture`:
```java
MDC.put("traceId", "abc");
@@ -618,17 +591,15 @@ requestReplyService
});
```
-You do not need to capture a `ContextSnapshot` yourself. As usual for micrometer context
-propagation, the relevant `ThreadLocalAccessor` (e.g. the one for the `MDC`, normally registered by
-your observability/tracing setup) must be present on the `ContextRegistry`.
+You do not need to capture a `ContextSnapshot` yourself; the relevant `ThreadLocalAccessor` (e.g. the
+one for the `MDC`) must be registered on the `ContextRegistry`, as is usual for Micrometer context
+propagation.
### Excluding the starter in tests
-Tests that do not need the request/reply functionality (for example a `@JsonTest` or a `@WebMvcTest`
-slice) do not load this starter's auto-configuration, so it is inactive there automatically.
-
-If you use a broader test that does pick up auto-configuration but you still want to switch the
-request/reply machinery off, exclude the auto-configuration like any other:
+Sliced tests (e.g. `@JsonTest`, `@WebMvcTest`) do not load this starter's auto‑configuration, so it is
+inactive there automatically. If a broader test picks up auto‑configuration but you want request/reply
+switched off, exclude it like any other auto‑configuration:
```java
@SpringBootTest
@@ -646,44 +617,51 @@ spring:
exclude: community.solace.spring.cloud.requestreply.service.RequestReplyAutoConfiguration
```
-When excluded, neither the request/reply service nor the per-binding reply consumers are registered.
+When excluded, neither the request/reply service nor the per‑binding reply consumers are registered.
-## Known issues and Open Points
+## Known issues and limitations
### Statefulness
-Since message relations are kept in memory, this starter is neither fail-safe nor scalable.
-More precisely:
-- If one instance of the service sends a message and another one receives the response, these cannot be related.
-- If a service dies, any relations are forgotten and replies can no longer be related to request,
- potentially resulting in message loss.
+Request/reply relations are kept **in memory**, so this starter is neither fail‑safe nor horizontally
+scalable for a single request:
-### Reply duplication & requester-side deduplication
+- if one instance sends a request and another receives the reply, they cannot be correlated;
+- if an instance dies, its in‑flight relations are lost and the corresponding replies can no longer be
+ matched, potentially resulting in message loss.
-In some operational scenarios (e.g. Solace in-place broker updates / short disconnects / reconnects) the same request can be sent twice and the replier may therefore produce **duplicate replies**.
+### Duplicate replies
-To make request-reply robust against such duplicate delivery, the requester keeps per-request bookkeeping and **deduplicates incoming reply messages by `replyIndex`**:
+Duplicate delivery (e.g. after a broker reconnect during an in‑place update) can cause duplicate replies.
+The requester mitigates this by deduplicating on `replyIndex` (see
+[Reply deduplication](#reply-deduplication)); the bound for unknown/streaming reply counts is
+configurable via the `spring.cloud.stream.requestreply.dedup.maxBitsWhenUnknown` JVM system property.
-- If the requester receives multiple messages with the same `replyIndex`, only the first one is processed; later duplicates are ignored.
-- This also supports range indices such as `replyIndex="0-45"` (used when replies are grouped into an SDTStream). The entire grouped message will be consumed only once.
-- Terminal messages (finish/error) are always processed, even if they share a `replyIndex` with another message.
+## Building and testing
-#### Dedup bitmap size limit (unknown / streaming totalReplies)
+The project builds with Maven (Java 17+):
-When `totalReplies` is not known yet (e.g. streaming / unknown-size reply patterns), the requester still deduplicates numeric `replyIndex` values, but it must place an upper bound on how large the internal bitmap can grow.
+```sh
+# compile and run the unit + integration tests
+mvn verify
-You can configure this limit via:
+# build without signing artifacts
+mvn -Dgpg.skip verify
+```
-```properties
-spring.cloud.stream.requestreply.dedup.maxBitsWhenUnknown=100000
+An optional OWASP dependency check is available via the `owasp-dependency-check` profile:
+
+```sh
+mvn -Powasp-dependency-check verify
```
-- Default: **100000** bits
-- Effect: any `replyIndex` (or range end) above this limit will not be deduplicated.
+## Contributing
-#### Example log message
+Contributions are welcome. Please read [CONTRIBUTING.md](CONTRIBUTING.md) and our
+[Code of Conduct](CODE_OF_CONDUCT.md), and see the [CHANGELOG](CHANGELOG.md) for release history.
+Questions about the code or Solace technologies are welcome in the
+[Solace community](https://solace.community).
-```
-2023-10-04 10:00:00.000 INFO 12345 --- [nio-8080-exec-1] c.s.s.requestreply.examples.sending : <<< MyRequest(location=livingroom) [correlationId=12345, replyTo=requestReply/response/solace/*/pub_sub_sending_K353456_315fd96b-b981-417b-be99-3be065c6611d, ...]
-2023-10-04 10:00:00.000 INFO 12345 --- [nio-8080-exec-1] c.s.s.requestreply.examples.sending : >>> SensorReading(foo=1337) [correlationId=12345, replyTo=requestReply/response/solace/*/pub_sub_sending_K353456_315fd96b-b981-417b-be99-3be065c6611d, remainingReplies=0, ...]
-```
+## License
+
+This project is licensed under the MIT License — see the [LICENSE](LICENSE) file for details.