Skip to content

Commit 663fb8f

Browse files
ozangunalpozangunalp
committed
Add RabbitMQRequestReply implementation (#3357)
Port of 2bb415e
1 parent 655c71c commit 663fb8f

24 files changed

Lines changed: 1152 additions & 111 deletions

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/ConnectionHolder.java‎

Lines changed: 61 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,9 @@ public class ConnectionHolder {
3636
private final Set<String> channels = ConcurrentHashMap.newKeySet();
3737

3838
private volatile Connection connection;
39+
private final ConcurrentHashMap<String, Channel> sharedChannels = new ConcurrentHashMap<>();
40+
private final ConcurrentHashMap<String, Context> sharedChannelContexts = new ConcurrentHashMap<>();
41+
private volatile Uni<Connection> connectionUni;
3942
private Consumer<Connection> onConnectionEstablished;
4043

4144
public ConnectionHolder(
@@ -74,35 +77,44 @@ public void onConnectionEstablished(Consumer<Connection> callback) {
7477
* Retries connection based on reconnect-attempts and reconnect-interval configuration.
7578
*/
7679
public Uni<Connection> connect() {
77-
Uni<Connection> connectionUni = Uni.createFrom().item(() -> {
78-
try {
79-
log.establishingConnection(channelName);
80+
if (connectionUni != null) {
81+
return connectionUni;
82+
}
83+
synchronized (this) {
84+
if (connectionUni != null) {
85+
return connectionUni;
86+
}
87+
Uni<Connection> uni = Uni.createFrom().item(() -> {
88+
try {
89+
log.establishingConnection(channelName);
8090

81-
Connection conn = factory.newConnection(connectionName);
91+
Connection conn = factory.newConnection(connectionName);
8292

83-
// Set connection before recovery listeners so createChannel() works in recovery callbacks
84-
connection = conn;
85-
connected.set(true);
93+
// Set connection before recovery listeners so createChannel() works in recovery callbacks
94+
connection = conn;
95+
connected.set(true);
8696

87-
setupRecoveryListeners(conn);
97+
setupRecoveryListeners(conn);
8898

89-
log.connectionEstablished(channelName);
99+
log.connectionEstablished(channelName);
90100

91-
return conn;
92-
} catch (IOException | TimeoutException e) {
93-
log.unableToConnectToBroker(e);
94-
throw ex.illegalStateUnableToCreateClient(e);
101+
return conn;
102+
} catch (IOException | TimeoutException e) {
103+
log.unableToConnectToBroker(e);
104+
throw ex.illegalStateUnableToCreateClient(e);
105+
}
106+
}).runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
107+
108+
if (reconnectAttempts > 0) {
109+
uni = uni
110+
.onFailure().retry()
111+
.withBackOff(Duration.ofSeconds(reconnectInterval))
112+
.atMost(reconnectAttempts);
95113
}
96-
}).runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
97114

98-
if (reconnectAttempts > 0) {
99-
connectionUni = connectionUni
100-
.onFailure().retry()
101-
.withBackOff(Duration.ofSeconds(reconnectInterval))
102-
.atMost(reconnectAttempts);
115+
connectionUni = uni.memoize().until(() -> connection != null && !connection.isOpen());
116+
return connectionUni;
103117
}
104-
105-
return connectionUni;
106118
}
107119

108120
private void setupRecoveryListeners(Connection conn) {
@@ -135,7 +147,34 @@ public void handleRecovery(Recoverable recoverable) {
135147
}
136148

137149
/**
138-
* Creates a new channel.
150+
* Returns the Vert.x context for the given shared channel name, creating it if needed.
151+
* All access to the corresponding shared AMQP channel must be serialized through this context.
152+
*/
153+
public Context getOrCreateSharedChannelContext(String name) {
154+
return sharedChannelContexts.computeIfAbsent(name,
155+
k -> ((VertxInternal) vertx).createEventLoopContext());
156+
}
157+
158+
/**
159+
* Returns the shared AMQP channel for the given name, creating it if needed.
160+
* An outgoing publisher and its reply-to consumer must use the same named channel
161+
* for RabbitMQ direct reply-to to work (it is channel-scoped).
162+
* All access must be serialized through {@link #getOrCreateSharedChannelContext(String)}.
163+
*/
164+
public Channel getOrCreateSharedChannel(String name) throws IOException {
165+
Channel ch = sharedChannels.get(name);
166+
if (ch == null || !ch.isOpen()) {
167+
if (connection == null || !connection.isOpen()) {
168+
throw ex.illegalStateConnectionClosed();
169+
}
170+
ch = connection.createChannel();
171+
sharedChannels.put(name, ch);
172+
}
173+
return ch;
174+
}
175+
176+
/**
177+
* Creates a new private channel.
139178
* Note: Channels are NOT thread-safe and should be used from a single thread or synchronized.
140179
*/
141180
public Channel createChannel() throws IOException {

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/IncomingRabbitMQMessage.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,10 @@ public java.util.Optional<String> getCorrelationId() {
141141
return java.util.Optional.ofNullable(rabbitMQMetadata.getCorrelationId());
142142
}
143143

144+
public java.util.Map<String, Object> getHeaders() {
145+
return rabbitMQMetadata.getHeaders();
146+
}
147+
144148
@Override
145149
public Supplier<CompletionStage<Void>> getAck() {
146150
return () -> ackHandler.handle(this);

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/IncomingRabbitMQMetadata.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
import com.rabbitmq.client.AMQP;
66
import com.rabbitmq.client.Envelope;
7+
import com.rabbitmq.client.LongString;
78

89
/**
910
* Metadata for incoming RabbitMQ messages.
@@ -63,7 +64,16 @@ public String getContentEncoding() {
6364
}
6465

6566
public Map<String, Object> getHeaders() {
66-
return properties.getHeaders() != null ? properties.getHeaders() : Collections.emptyMap();
67+
Map<String, Object> raw = properties.getHeaders();
68+
if (raw == null) {
69+
return Collections.emptyMap();
70+
}
71+
Map<String, Object> result = new java.util.HashMap<>();
72+
for (Map.Entry<String, Object> e : raw.entrySet()) {
73+
Object v = e.getValue();
74+
result.put(e.getKey(), v instanceof LongString ? v.toString() : v);
75+
}
76+
return result;
6777
}
6878

6979
public Integer getDeliveryMode() {

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/RabbitMQConnector.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@
117117
@ConnectorAttribute(name = "failure-strategy", direction = INCOMING, description = "The failure strategy to apply when a RabbitMQ message is nacked. Accepted values are `fail`, `accept`, `reject` (default), `requeue` or name of a bean", type = "string", defaultValue = "reject")
118118
@ConnectorAttribute(name = "broadcast", direction = INCOMING, description = "Whether the received RabbitMQ messages must be dispatched to multiple _subscribers_", type = "boolean", defaultValue = "false")
119119
@ConnectorAttribute(name = "auto-acknowledgement", direction = INCOMING, description = "Whether the received RabbitMQ messages must be acknowledged when received; if true then delivery constitutes acknowledgement", type = "boolean", defaultValue = "false")
120-
@ConnectorAttribute(name = "routing-keys", direction = INCOMING, description = "A comma-separated list of routing keys to bind the queue to the exchange. Relevant only if 'exchange.type' is topic or direct", type = "string", defaultValue = "#")
120+
@ConnectorAttribute(name = "routing-keys", direction = INCOMING, description = "A comma-separated list of routing keys to bind the queue to the exchange. Relevant only if 'exchange.type' is topic or direct", type = "string")
121121
@ConnectorAttribute(name = "arguments", direction = INCOMING, description = "A comma-separated list of arguments [key1:value1,key2:value2,...] to bind the queue to the exchange. Relevant only if 'exchange.type' is headers", type = "string")
122122
@ConnectorAttribute(name = "consumer-arguments", direction = INCOMING, description = "A comma-separated list of arguments [key1:value1,key2:value2,...] for created consumer", type = "string")
123123
@ConnectorAttribute(name = "consumer-tag", direction = INCOMING, description = "The consumer-tag option for created consumer, if not provided the consumer gets assigned a tag generated by the broker", type = "string")

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/RabbitMQMessageConverter.java‎

Lines changed: 40 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -54,22 +54,50 @@ public static OutgoingRabbitMQMessage convert(
5454
return convertFromIncoming((IncomingRabbitMQMessage<?>) message, defaultRoutingKey);
5555
}
5656

57-
// Get routing key from metadata or use default
58-
String routingKey = message.getMetadata(OutgoingRabbitMQMetadata.class)
59-
.map(OutgoingRabbitMQMetadata::getRoutingKey)
60-
.orElse(defaultRoutingKey);
61-
62-
// Get exchange from metadata (optional override)
63-
Optional<String> exchange = message.getMetadata(OutgoingRabbitMQMetadata.class)
64-
.flatMap(OutgoingRabbitMQMetadata::getExchange);
65-
6657
// Convert payload to bytes
6758
byte[] body = getBodyFromPayload(message.getPayload());
59+
String defaultContentType = getDefaultContentTypeForPayload(message.getPayload());
60+
61+
Optional<OutgoingRabbitMQMetadata> outgoing = message.getMetadata(OutgoingRabbitMQMetadata.class);
62+
OutgoingRabbitMQMetadata.Builder builder = outgoing.map(out -> {
63+
OutgoingRabbitMQMetadata.Builder b = OutgoingRabbitMQMetadata.from(out);
64+
if (out.getProperties().getContentType() == null) {
65+
b.withContentType(defaultContentType);
66+
}
67+
if (out.getProperties().getDeliveryMode() == null) {
68+
b.withDeliveryMode(2);
69+
}
70+
return b;
71+
}).orElseGet(() -> OutgoingRabbitMQMetadata.builder()
72+
.withContentType(defaultContentType)
73+
.withDeliveryMode(2)
74+
.withExpiration(defaultTtl.map(String::valueOf).orElse(null)));
75+
76+
Optional<IncomingRabbitMQMetadata> incoming = message.getMetadata(IncomingRabbitMQMetadata.class);
77+
incoming.ifPresent(in -> {
78+
if (outgoing.map(m -> m.getProperties().getCorrelationId()).isEmpty()) {
79+
String cid = in.getCorrelationId();
80+
if (cid != null) {
81+
builder.withCorrelationId(cid);
82+
}
83+
}
84+
});
85+
86+
OutgoingRabbitMQMetadata metadata = builder.build();
87+
88+
// Get routing key: outgoing metadata > replyTo from incoming > default
89+
String routingKey = metadata.getRoutingKey();
90+
if (routingKey == null) {
91+
routingKey = incoming.map(IncomingRabbitMQMetadata::getReplyTo).orElse(null);
92+
}
93+
if (routingKey == null) {
94+
routingKey = defaultRoutingKey;
95+
}
6896

69-
// Build properties
70-
AMQP.BasicProperties properties = buildProperties(message, defaultTtl);
97+
// Get exchange from metadata (optional override)
98+
Optional<String> exchange = metadata.getExchange();
7199

72-
return new OutgoingRabbitMQMessage(routingKey, exchange, body, properties);
100+
return new OutgoingRabbitMQMessage(routingKey, exchange, body, metadata.getProperties());
73101
}
74102

75103
/**
@@ -118,44 +146,6 @@ private static OutgoingRabbitMQMessage convertFromIncoming(
118146
return new OutgoingRabbitMQMessage(routingKey, exchange, body, properties);
119147
}
120148

121-
/**
122-
* Build AMQP properties from message metadata.
123-
*/
124-
private static AMQP.BasicProperties buildProperties(Message<?> message, Optional<Long> defaultTtl) {
125-
Optional<OutgoingRabbitMQMetadata> metadata = message.getMetadata(OutgoingRabbitMQMetadata.class);
126-
127-
if (metadata.isPresent() && metadata.get().getProperties() != null) {
128-
// Use properties from metadata if provided
129-
return metadata.get().getProperties();
130-
}
131-
132-
// Build default properties
133-
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
134-
135-
// Set content type based on payload
136-
String contentType = getDefaultContentTypeForPayload(message.getPayload());
137-
builder.contentType(contentType);
138-
139-
// Set delivery mode to persistent by default
140-
builder.deliveryMode(2);
141-
142-
// Set TTL if provided
143-
if (defaultTtl.isPresent()) {
144-
builder.expiration(String.valueOf(defaultTtl.get()));
145-
}
146-
147-
// Apply metadata if present
148-
if (metadata.isPresent()) {
149-
OutgoingRabbitMQMetadata meta = metadata.get();
150-
151-
// Note: If properties were set via builder in metadata, they were already returned above
152-
// This handles the case where individual fields might be set (though current implementation
153-
// always builds complete properties)
154-
}
155-
156-
return builder.build();
157-
}
158-
159149
/**
160150
* Convert payload to byte array.
161151
*/

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/i18n/RabbitMQLogging.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -218,4 +218,12 @@ public interface RabbitMQLogging extends BasicLogger {
218218
@LogMessage(level = Logger.Level.ERROR)
219219
@Message(id = 18062, value = "Wait for confirms failed for channel `%s`")
220220
void waitForConfirmsFailed(String channel, @Cause Throwable e);
221+
222+
@LogMessage(level = Logger.Level.DEBUG)
223+
@Message(id = 18063, value = "Request-reply message ignored on channel `%s`, correlation ID `%s`")
224+
void requestReplyMessageIgnored(String channel, String correlationId);
225+
226+
@LogMessage(level = Logger.Level.ERROR)
227+
@Message(id = 18064, value = "Request-reply consumer failure on channel `%s`")
228+
void requestReplyConsumerFailure(String channel, @Cause Throwable t);
221229
}

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/internals/IncomingRabbitMQChannel.java‎

Lines changed: 2 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,6 @@
3737
import io.smallrye.reactive.messaging.rabbitmq.og.tracing.RabbitMQOpenTelemetryInstrumenter;
3838
import io.smallrye.reactive.messaging.rabbitmq.og.tracing.RabbitMQTrace;
3939
import io.vertx.core.Context;
40-
import io.vertx.core.internal.VertxInternal;
4140

4241
/**
4342
* Incoming RabbitMQ channel that consumes messages from a queue.
@@ -70,7 +69,7 @@ public IncomingRabbitMQChannel(
7069
this.connectionHolder = connectionHolder;
7170
this.configuration = configuration;
7271
this.configMaps = configMaps;
73-
this.incomingContext = ((VertxInternal) connectionHolder.getVertx()).createEventLoopContext();
72+
this.incomingContext = connectionHolder.getOrCreateSharedChannelContext(configuration.getChannel());
7473

7574
// Initialize tracing if enabled
7675
if (configuration.getTracingEnabled()) {
@@ -144,23 +143,8 @@ private void setupConsumer(
144143
java.util.function.Consumer<Throwable> onError,
145144
Runnable onComplete) {
146145

147-
// Register recovery callback to re-register consumer after connection recovery.
148-
// With topologyRecoveryEnabled=false, consumers are not automatically recovered
149-
// by the RabbitMQ client, so we must re-setup manually.
150146
connectionHolder.onConnectionEstablished(conn -> {
151147
try {
152-
// Close old channel (it has been recovered but has no consumer registered)
153-
Channel oldChannel = channelRef.get();
154-
if (oldChannel != null) {
155-
try {
156-
if (oldChannel.isOpen()) {
157-
oldChannel.close();
158-
}
159-
} catch (Exception e) {
160-
// Ignore - channel may already be closed
161-
}
162-
}
163-
// Re-setup consumer on recovered connection
164148
setupConsumerOnConnection(onMessage, onError, onComplete);
165149
} catch (Exception e) {
166150
log.unableToCreateConsumer(configuration.getChannel(), e);
@@ -189,7 +173,7 @@ private void setupConsumerOnConnection(
189173
Runnable onComplete) throws Exception {
190174

191175
// Create channel for consuming
192-
Channel channel = connectionHolder.createChannel();
176+
Channel channel = connectionHolder.getOrCreateSharedChannel(configuration.getChannel());
193177
channelRef.set(channel);
194178

195179
Context context = incomingContext;
@@ -501,14 +485,6 @@ private void cleanup() {
501485
}
502486
}
503487

504-
if (channel != null && channel.isOpen()) {
505-
try {
506-
channel.close();
507-
} catch (Exception e) {
508-
log.unableToCloseChannel(configuration.getChannel(), e);
509-
}
510-
}
511-
512488
subscribed.set(false);
513489
} catch (Exception e) {
514490
log.cleanupFailed(configuration.getChannel(), e);

‎smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/internals/OutgoingRabbitMQChannel.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
import io.smallrye.reactive.messaging.rabbitmq.og.tracing.RabbitMQOpenTelemetryInstrumenter;
3030
import io.smallrye.reactive.messaging.rabbitmq.og.tracing.RabbitMQTrace;
3131
import io.vertx.core.Context;
32-
import io.vertx.core.internal.VertxInternal;
3332

3433
/**
3534
* Outgoing RabbitMQ channel that publishes messages to an exchange.
@@ -69,7 +68,7 @@ public OutgoingRabbitMQChannel(
6968
this.connectionHolder = connectionHolder;
7069
this.configuration = configuration;
7170
this.configMaps = configMaps;
72-
this.outgoingContext = ((VertxInternal) connectionHolder.getVertx()).createEventLoopContext();
71+
this.outgoingContext = connectionHolder.getOrCreateSharedChannelContext(configuration.getChannel());
7372

7473
// Initialize tracing if enabled
7574
if (configuration.getTracingEnabled()) {
@@ -92,8 +91,7 @@ private void initialize() {
9291
.subscribe().with(
9392
conn -> {
9493
try {
95-
// Create channel for publishing
96-
channel = connectionHolder.createChannel();
94+
channel = connectionHolder.getOrCreateSharedChannel(configuration.getChannel());
9795

9896
// Set up topology
9997
setupTopology();

0 commit comments

Comments
 (0)