Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2931: AMQP De-Batching as List<?> Payload

Browse Source 
Resolves https://github.com/spring-projects/spring-integration/issues/2931
This commit is contained in:
Gary Russell
2019-05-14 15:11:31 -04:00
committed by Artem Bilan
parent e05b6f75f4
commit 9f53b80e18
7 changed files with 202 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -16,11 +16,15 @@
package org.springframework.integration.amqp.inbound;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.AmqpHeaders;
@@ -69,6 +73,8 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
private RecoveryCallback<? extends Object> recoveryCallback;
private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
public AmqpInboundChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
Assert.notNull(listenerContainer, "listenerContainer must not be null");
Assert.isNull(listenerContainer.getMessageListener(),
@@ -115,6 +121,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
this.recoveryCallback = recoveryCallback;
}
/**
* Set a batching strategy to use when de-batching messages.
* Default is {@link SimpleBatchingStrategy}.
* @param batchingStrategy the strategy.
* @since 5.2
*/
public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
Assert.notNull(batchingStrategy, "'batchingStrategy' cannot be null");
this.batchingStrategy = batchingStrategy;
}
@Override
public String getComponentType() {
@@ -239,7 +255,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
}
private org.springframework.messaging.Message<Object> createMessage(Message message, Channel channel) {
Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
Object payload;
if (AmqpInboundChannelAdapter.this.batchingStrategy.canDebatch(message.getMessageProperties())) {
List<Object> payloads = new ArrayList<>();
AmqpInboundChannelAdapter.this.batchingStrategy.deBatch(message, fragment -> payloads
.add(AmqpInboundChannelAdapter.this.messageConverter.fromMessage(fragment)));
payload = payloads;
}
else {
payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
}
Map<String, Object> headers = AmqpInboundChannelAdapter.this.headerMapper
.toHeadersFromRequest(message.getMessageProperties());
if (isManualAck()) {

27
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.amqp.inbound;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
@@ -25,6 +27,8 @@ import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
@@ -81,6 +85,8 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
private RecoveryCallback<? extends Object> recoveryCallback;
private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer) {
this(listenerContainer, new RabbitTemplate(listenerContainer.getConnectionFactory()), false);
}
@@ -175,6 +181,17 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
this.recoveryCallback = recoveryCallback;
}
/**
* Set a batching strategy to use when de-batching messages.
* Default is {@link SimpleBatchingStrategy}.
* @param batchingStrategy the strategy.
* @since 5.2
*/
public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
Assert.notNull(batchingStrategy, "'batchingStrategy' cannot be null");
this.batchingStrategy = batchingStrategy;
}
@Override
public String getComponentType() {
return "amqp:inbound-gateway";
@@ -286,7 +303,15 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
boolean isManualAck = AmqpInboundGateway.this.messageListenerContainer
.getAcknowledgeMode() == AcknowledgeMode.MANUAL;
try {
payload = AmqpInboundGateway.this.amqpMessageConverter.fromMessage(message);
if (AmqpInboundGateway.this.batchingStrategy.canDebatch(message.getMessageProperties())) {
List<Object> payloads = new ArrayList<>();
AmqpInboundGateway.this.batchingStrategy.deBatch(message, fragment -> payloads
.add(AmqpInboundGateway.this.amqpMessageConverter.fromMessage(fragment)));
payload = payloads;
}
else {
payload = AmqpInboundGateway.this.amqpMessageConverter.fromMessage(message);
}
headers = AmqpInboundGateway.this.headerMapper.toHeadersFromRequest(message.getMessageProperties());
if (isManualAck) {
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());

32
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java
View File

@@ -18,12 +18,16 @@ package org.springframework.integration.amqp.inbound;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
@@ -71,6 +75,8 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
private boolean rawMessageHeader;
private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
public AmqpMessageSource(ConnectionFactory connectionFactory, String queue) {
this(connectionFactory, new AmqpAckCallbackFactory(), queue);
}
@@ -151,6 +157,21 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
this.rawMessageHeader = rawMessageHeader;
}
protected BatchingStrategy getBatchingStrategy() {
return this.batchingStrategy;
}
/**
* Set a batching strategy to use when de-batching messages.
* Default is {@link SimpleBatchingStrategy}.
* @param batchingStrategy the strategy.
* @since 5.2
*/
public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
Assert.notNull(batchingStrategy, "'batchingStrategy' cannot be null");
this.batchingStrategy = batchingStrategy;
}
@Override
public String getComponentType() {
return "amqp:message-source";
@@ -174,7 +195,16 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
messageProperties.setConsumerQueue(this.queue);
Map<String, Object> headers = this.headerMapper.toHeadersFromRequest(messageProperties);
org.springframework.amqp.core.Message amqpMessage = new org.springframework.amqp.core.Message(resp.getBody(), messageProperties);
Object payload = this.messageConverter.fromMessage(amqpMessage);
Object payload;
if (this.batchingStrategy.canDebatch(messageProperties)) {
List<Object> payloads = new ArrayList<>();
this.batchingStrategy.deBatch(amqpMessage, fragment -> payloads
.add(this.messageConverter.fromMessage(fragment)));
payload = payloads;
}
else {
payload = this.messageConverter.fromMessage(amqpMessage);
}
AbstractIntegrationMessageBuilder<Object> builder = getMessageBuilderFactory().withPayload(payload)
.copyHeaders(headers)
.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, callback);

39
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java
View File

@@ -24,10 +24,14 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.List;
import java.util.concurrent.ExecutorService;
import org.junit.Test;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.batch.MessageBatch;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.integration.StaticMessageHeaderAccessor;
@@ -123,4 +127,39 @@ public class AmqpMessageSourceTests {
verify(connection).close(30000);
}
@SuppressWarnings({ "unchecked" })
@Test
public void testBatch() throws Exception {
SimpleBatchingStrategy bs = new SimpleBatchingStrategy(2, 10_000, 10_000L);
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType("text/plain");
org.springframework.amqp.core.Message message =
new org.springframework.amqp.core.Message("test1".getBytes(), messageProperties);
bs.addToBatch("foo", "bar", message);
message = new org.springframework.amqp.core.Message("test2".getBytes(), messageProperties);
MessageBatch batched = bs.addToBatch("foo", "bar", message);
Channel channel = mock(Channel.class);
willReturn(true).given(channel).isOpen();
Envelope envelope = new Envelope(123L, false, "ex", "rk");
BasicProperties props = new BasicProperties.Builder()
.headers(batched.getMessage().getMessageProperties().getHeaders())
.contentType("text/plain")
.build();
GetResponse getResponse = new GetResponse(envelope, props, batched.getMessage().getBody(), 0);
willReturn(getResponse).given(channel).basicGet("foo", false);
Connection connection = mock(Connection.class);
willReturn(true).given(connection).isOpen();
willReturn(channel).given(connection).createChannel();
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
willReturn(connection).given(connectionFactory).newConnection((ExecutorService) isNull(), anyString());
CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
AmqpMessageSource source = new AmqpMessageSource(ccf, "foo");
Message<?> received = source.receive();
assertThat(received).isNotNull();
assertThat(((List<String>) received.getPayload())).contains("test1", "test2");
}
}

51
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -27,6 +27,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -36,6 +37,8 @@ import org.mockito.Mockito;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.batch.MessageBatch;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -377,6 +380,54 @@ public class InboundEndpointTests {
assertThat(errors.receive(0)).isNull();
}
@SuppressWarnings({ "unchecked" })
@Test
public void testBatchdAdapter() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(mock(ConnectionFactory.class));
container.setDeBatchingEnabled(false);
AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
QueueChannel out = new QueueChannel();
adapter.setOutputChannel(out);
adapter.afterPropertiesSet();
ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
SimpleBatchingStrategy bs = new SimpleBatchingStrategy(2, 10_000, 10_000L);
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType("text/plain");
org.springframework.amqp.core.Message message =
new org.springframework.amqp.core.Message("test1".getBytes(), messageProperties);
bs.addToBatch("foo", "bar", message);
message = new org.springframework.amqp.core.Message("test2".getBytes(), messageProperties);
MessageBatch batched = bs.addToBatch("foo", "bar", message);
listener.onMessage(batched.getMessage(), null);
Message<?> received = out.receive();
assertThat(received).isNotNull();
assertThat(((List<String>) received.getPayload())).contains("test1", "test2");
}
@SuppressWarnings({ "unchecked" })
@Test
public void testBatchGateway() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(mock(ConnectionFactory.class));
container.setDeBatchingEnabled(false);
AmqpInboundGateway adapter = new AmqpInboundGateway(container);
QueueChannel out = new QueueChannel();
adapter.setRequestChannel(out);
adapter.afterPropertiesSet();
ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
SimpleBatchingStrategy bs = new SimpleBatchingStrategy(2, 10_000, 10_000L);
MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType("text/plain");
org.springframework.amqp.core.Message message =
new org.springframework.amqp.core.Message("test1".getBytes(), messageProperties);
bs.addToBatch("foo", "bar", message);
message = new org.springframework.amqp.core.Message("test2".getBytes(), messageProperties);
MessageBatch batched = bs.addToBatch("foo", "bar", message);
listener.onMessage(batched.getMessage(), null);
Message<?> received = out.receive();
assertThat(received).isNotNull();
assertThat(((List<String>) received.getPayload())).contains("test1", "test2");
}
public static class Foo {
private String bar;

26
src/reference/asciidoc/amqp.adoc
View File

@@ -277,8 +277,22 @@ public class AmqpJavaApplication {
----
====
[[amqp-debatching]]
==== Batched Messages
See https://docs.spring.io/spring-amqp/docs/current/reference/html/#template-batching[the Spring AMQP Documentation] for more information about batched messages.
To produce batched messages with Spring Integration, simply configure the outbound endpoint with a `BatchingRabbitTemplate`.
When receiving batched messages, by default, the listener containers extract each fragment message and the adapter will produce a `Message<?>` for each fragment.
Starting with version 5.2, if the container's `deBatchingEnabled` property is set to `false`, the de-batching is performed by the adapter instead, and a single `Message<List<?>>` is produced with the payload being a list of the fragment payloads (after conversion if appropriate).
The default `BatchingStrategy` is the `SimpleBatchingStrategy`, but this can be overridden on the adapter.
=== Polled Inbound Channel Adapter
==== Overview
Version 5.0.1 introduced a polled channel adapter, letting you fetch individual messages on demand -- for example, with a `MessageSourcePollingTemplate` or a poller.
See <<deferred-acks-message-source>> for more information.
@@ -315,6 +329,13 @@ public IntegrationFlow flow() {
----
====
[[amqp-polled-debatching]]
==== Batched Messages
See <<amqp-debatching>>.
For the polled adapter, there is no listener container, batched messages are always debatched (if the `BatchingStrategy` supports doing so).
[[amqp-inbound-gateway]]
=== Inbound Gateway
@@ -453,6 +474,11 @@ public class AmqpJavaApplication {
----
====
[[amqp-gatewway-debatching]]
==== Batched Messages
See <<amqp-debatching>>.
[[amqp-inbound-ack]]
=== Inbound Endpoint Acknowledge Mode

3
src/reference/asciidoc/whats-new.adoc
View File

@@ -40,6 +40,9 @@ See <<splitter>> for more information.
The outbound endpoints can now be configured to synthesize a "nack" if no publisher confirm is received within a timeout.
See <<amqp-outbound-endpoints>> for more information.
The inbound channel adapter can now receive batched messages as a `List<?>` payload instead of receiving a discrete message for each batch fragment.
See <<amqp-debatching>> for more information.
[[x5.2-file]]
==== File Changes