From 26816a3eef5d7b6ca69136bb777e49a5646dcc22 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 26 Sep 2022 17:43:37 -0400 Subject: [PATCH] GH-3626: RabbitMQ Stream Support (#3895) Resolves https://github.com/spring-projects/spring-integration/issues/3626 - spec for stream listener container - move message handler from scst to here; add spec * Fix test. * Polishing and Docs. * Fix anchors in doc. * Don't Extend `AbstractMessageProducingHandler` --- build.gradle | 6 +- .../AbstractMessageListenerContainerSpec.java | 8 +- .../integration/amqp/dsl/Amqp.java | 2 +- .../dsl/AmqpInboundChannelAdapterSpec.java | 10 +- .../dsl/MessageListenerContainerSpec.java | 47 ++++ .../integration/amqp/dsl/RabbitStream.java | 89 +++++++ ...RabbitStreamInboundChannelAdapterSpec.java | 54 ++++ .../dsl/RabbitStreamMessageHandlerSpec.java | 143 ++++++++++ ...bitStreamMessageListenerContainerSpec.java | 110 ++++++++ .../outbound/RabbitStreamMessageHandler.java | 249 ++++++++++++++++++ .../amqp/support/MappingUtils.java | 33 ++- .../integration/amqp/dsl/AmqpTests.java | 19 ++ .../RabbitStreamMessageHandlerTests.java | 120 +++++++++ .../amqp/support/RabbitTestContainer.java | 57 ++++ .../IntegrationManagementConfigurer.java | 1 - src/reference/asciidoc/amqp.adoc | 54 +++- src/reference/asciidoc/whats-new.adoc | 5 + 17 files changed, 988 insertions(+), 19 deletions(-) create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/MessageListenerContainerSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStream.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamInboundChannelAdapterSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageHandlerSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageListenerContainerSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandler.java create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandlerTests.java create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/RabbitTestContainer.java diff --git a/build.gradle b/build.gradle index 516e780429..03e9aca6dd 100644 --- a/build.gradle +++ b/build.gradle @@ -99,7 +99,7 @@ ext { rsocketVersion = '1.1.3' servletApiVersion = '5.0.0' smackVersion = '4.4.6' - springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '3.0.0-M4' + springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '3.0.0-SNAPSHOT' springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2022.0.0-M6' springGraphqlVersion = '1.1.0-M1' springKafkaVersion = '3.0.0-M6' @@ -470,12 +470,16 @@ project('spring-integration-amqp') { api("org.springframework.amqp:spring-rabbit:$springAmqpVersion") { exclude group: 'org.springframework' } + optionalApi("org.springframework.amqp:spring-rabbit-stream:$springAmqpVersion") { + exclude group: 'org.springframework' + } testImplementation("org.springframework.amqp:spring-rabbit-junit:$springAmqpVersion") { exclude group: 'org.springframework' } testImplementation project(':spring-integration-stream') testImplementation 'org.springframework:spring-web' + testImplementation 'org.testcontainers:rabbitmq' } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java index 2e60c34f59..5e5eb31215 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2021 the original author or authors. + * Copyright 2017-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,14 +28,14 @@ import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.support.ConditionalExceptionLogger; import org.springframework.amqp.support.ConsumerTagStrategy; -import org.springframework.integration.dsl.IntegrationComponentSpec; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.TransactionAttribute; import org.springframework.util.ErrorHandler; import org.springframework.util.backoff.BackOff; /** - * Base class for container specs. + * Base class for container specs for containers that extend + * {@link AbstractMessageListenerContainer}. * * @param the current spec extension type * @param the listener container type @@ -48,7 +48,7 @@ import org.springframework.util.backoff.BackOff; */ public abstract class AbstractMessageListenerContainerSpec, C extends AbstractMessageListenerContainer> - extends IntegrationComponentSpec { + extends MessageListenerContainerSpec { public AbstractMessageListenerContainerSpec(C listenerContainer) { this.target = listenerContainer; diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java index 33b7a70466..38d2cc45d6 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2021 the original author or authors. + * Copyright 2014-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java index 057be05453..c3042ad6c2 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,7 +19,7 @@ package org.springframework.integration.amqp.dsl; import java.util.Collections; import java.util.Map; -import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; import org.springframework.integration.dsl.ComponentsRegistration; @@ -36,13 +36,13 @@ import org.springframework.integration.dsl.ComponentsRegistration; * @since 5.0 */ public abstract class AmqpInboundChannelAdapterSpec - , C extends AbstractMessageListenerContainer> + , C extends MessageListenerContainer> extends AmqpBaseInboundChannelAdapterSpec implements ComponentsRegistration { - protected final AbstractMessageListenerContainerSpec listenerContainerSpec; // NOSONAR final + protected final MessageListenerContainerSpec listenerContainerSpec; // NOSONAR final - protected AmqpInboundChannelAdapterSpec(AbstractMessageListenerContainerSpec listenerContainerSpec) { + protected AmqpInboundChannelAdapterSpec(MessageListenerContainerSpec listenerContainerSpec) { super(new AmqpInboundChannelAdapter(listenerContainerSpec.get())); this.listenerContainerSpec = listenerContainerSpec; } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/MessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/MessageListenerContainerSpec.java new file mode 100644 index 0000000000..0c748671df --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/MessageListenerContainerSpec.java @@ -0,0 +1,47 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.dsl; + +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; +import org.springframework.integration.dsl.IntegrationComponentSpec; + +/** + * Base class for container specs. + * + * @param the current spec extension type + * @param the listener container type + * + * @author Gary Russell + * + * @since 6.0 + * + */ +public abstract class MessageListenerContainerSpec, + C extends MessageListenerContainer> + extends IntegrationComponentSpec { + + /** + * Set the queue names. + * @param queueNames the queue names. + * @return this spec. + */ + public S queueName(String... queueNames) { + this.target.setQueueNames(queueNames); + return _this(); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStream.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStream.java new file mode 100644 index 0000000000..dd0c760738 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStream.java @@ -0,0 +1,89 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.dsl; + +import org.springframework.rabbit.stream.listener.StreamListenerContainer; +import org.springframework.rabbit.stream.producer.RabbitStreamTemplate; + +import com.rabbitmq.stream.Codec; +import com.rabbitmq.stream.Environment; + +/** + * Factory class for RabbitMQ components. + * + * @author Gary Russell + * @since 6.0 + * + */ +public final class RabbitStream { + + private RabbitStream() { + } + + /** + * Create an initial {@link RabbitStreamInboundChannelAdapterSpec} + * with the provided {@link StreamListenerContainer}. + * Note: only endpoint options are available from spec. + * The {@code listenerContainer} options should be specified + * on the provided {@link StreamListenerContainer} using + * {@link RabbitStreamInboundChannelAdapterSpec#configureContainer(java.util.function.Consumer)}. + * @param listenerContainer the listenerContainer. + * @return the RabbitInboundChannelAdapterSLCSpec. + */ + public static RabbitStreamInboundChannelAdapterSpec inboundAdapter(StreamListenerContainer listenerContainer) { + return new RabbitStreamInboundChannelAdapterSpec(listenerContainer); + } + + /** + * Create an initial {@link RabbitStreamInboundChannelAdapterSpec} + * with the provided {@link Environment}. + * Note: only endpoint options are available from spec. + * The {@code listenerContainer} options should be specified + * on the provided {@link StreamListenerContainer} using + * {@link RabbitStreamInboundChannelAdapterSpec#configureContainer(java.util.function.Consumer)}. + * @param environment the environment. + * @return the RabbitInboundChannelAdapterSLCSpec. + */ + public static RabbitStreamInboundChannelAdapterSpec inboundAdapter(Environment environment) { + return new RabbitStreamInboundChannelAdapterSpec(environment, null); + } + + /** + * Create an initial {@link RabbitStreamInboundChannelAdapterSpec} + * with the provided {@link Environment}. + * Note: only endpoint options are available from spec. + * The {@code listenerContainer} options should be specified + * on the provided {@link StreamListenerContainer} using + * {@link RabbitStreamInboundChannelAdapterSpec#configureContainer(java.util.function.Consumer)}. + * @param environment the environment. + * @param codec the codec. + * @return the RabbitInboundChannelAdapterSLCSpec. + */ + public static RabbitStreamInboundChannelAdapterSpec inboundAdapter(Environment environment, Codec codec) { + return new RabbitStreamInboundChannelAdapterSpec(environment, codec); + } + + /** + * Create an initial {@link RabbitStreamMessageHandlerSpec} (adapter). + * @param template the amqpTemplate. + * @return the RabbitStreamMessageHandlerSpec. + */ + public static RabbitStreamMessageHandlerSpec outboundStreamAdapter(RabbitStreamTemplate template) { + return new RabbitStreamMessageHandlerSpec(template); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamInboundChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamInboundChannelAdapterSpec.java new file mode 100644 index 0000000000..09c9f54a23 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamInboundChannelAdapterSpec.java @@ -0,0 +1,54 @@ +/* + * Copyright 2017-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.dsl; + +import java.util.function.Consumer; + +import org.springframework.lang.Nullable; +import org.springframework.rabbit.stream.listener.StreamListenerContainer; + +import com.rabbitmq.stream.Codec; +import com.rabbitmq.stream.Environment; + +/** + * Spec for an inbound channel adapter with a {@link StreamListenerContainer}. + * + * @author Gary Russell + * @author Artem Bilan + * + * @since 6.0 + * + */ +public class RabbitStreamInboundChannelAdapterSpec + extends AmqpInboundChannelAdapterSpec { + + protected RabbitStreamInboundChannelAdapterSpec(StreamListenerContainer listenerContainer) { + super(new RabbitStreamMessageListenerContainerSpec(listenerContainer)); + } + + protected RabbitStreamInboundChannelAdapterSpec(Environment environment, @Nullable Codec codec) { + super(new RabbitStreamMessageListenerContainerSpec(environment, codec)); + } + + public RabbitStreamInboundChannelAdapterSpec configureContainer( + Consumer configurer) { + + configurer.accept((RabbitStreamMessageListenerContainerSpec) this.listenerContainerSpec); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageHandlerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageHandlerSpec.java new file mode 100644 index 0000000000..fde1aef521 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageHandlerSpec.java @@ -0,0 +1,143 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.dsl; + +import org.springframework.integration.amqp.outbound.RabbitStreamMessageHandler; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.messaging.MessageChannel; +import org.springframework.rabbit.stream.producer.RabbitStreamOperations; + +/** + * The base {@link MessageHandlerSpec} for {@link RabbitStreamMessageHandler}s. + * + * @author Gary Russell + * + * @since 6.0 + */ +public class RabbitStreamMessageHandlerSpec + extends MessageHandlerSpec { + + private final DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); + + RabbitStreamMessageHandlerSpec(RabbitStreamOperations operations) { + this.target = new RabbitStreamMessageHandler(operations); + } + + /** + * Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers. + * @param headerMapper the {@link AmqpHeaderMapper} to use. + * @return this spec. + */ + public RabbitStreamMessageHandlerSpec headerMapper(AmqpHeaderMapper headerMapper) { + this.target.setHeaderMapper(headerMapper); + return this; + } + + /** + * Provide the header names that should be mapped from a request to a + * {@link org.springframework.messaging.MessageHeaders}. + * @param headers The request header names. + * @return this spec. + */ + public RabbitStreamMessageHandlerSpec mappedRequestHeaders(String... headers) { + this.headerMapper.setRequestHeaderNames(headers); + return this; + } + + /** + * Determine whether the headers are + * mapped before the message is converted, or afterwards. + * @param headersLast true to map headers last. + * @return this spec. + * @see RabbitStreamMessageHandler#setHeadersMappedLast(boolean) + */ + public RabbitStreamMessageHandlerSpec headersMappedLast(boolean headersLast) { + this.target.setHeadersMappedLast(headersLast); + return this; + } + + /** + * Set the success channel. + * @param channel the channel. + * @return this spec. + */ + public RabbitStreamMessageHandlerSpec sendSuccessChannel(MessageChannel channel) { + this.target.setSendSuccessChannel(channel); + return this; + } + + /** + * Set the failure channel. After a send failure, an + * {@link org.springframework.messaging.support.ErrorMessage} will be sent + * to this channel with a payload of the exception with the + * failed message. + * @param channel the channel. + * @return this spec. + */ + public RabbitStreamMessageHandlerSpec sendFailureChannel(MessageChannel channel) { + this.target.setSendFailureChannel(channel); + return this; + } + + /** + * Set the success channel. + * @param channel the channel. + * @return this spec. + */ + public RabbitStreamMessageHandlerSpec sendSuccessChannel(String channel) { + this.target.setSendSuccessChannelName(channel); + return this; + } + + /** + * Set the failure channel. After a send failure, an + * {@link org.springframework.messaging.support.ErrorMessage} will be sent + * to this channel with a payload of the exception with the + * failed message. + * @param channel the channel. + * @return this spec. + */ + public RabbitStreamMessageHandlerSpec sendFailureChannel(String channel) { + this.target.setSendFailureChannelName(channel); + return this; + } + + /** + * Set to true to wait for a confirmation. + * @param sync true to wait. + * @return this spec. + * @see #setConfirmTimeout(long) + */ + public RabbitStreamMessageHandlerSpec sync(boolean sync) { + this.target.setSync(sync); + return this; + } + + /** + * Set a timeout for the confirm result. + * @param timeout the approximate timeout. + * @return this spec. + * @see #sync(boolean) + */ + public RabbitStreamMessageHandlerSpec confirmTimeout(long timeout) { + this.target.setConfirmTimeout(timeout); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageListenerContainerSpec.java new file mode 100644 index 0000000000..dff3de9e19 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageListenerContainerSpec.java @@ -0,0 +1,110 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.dsl; + +import java.util.function.Consumer; + +import org.aopalliance.aop.Advice; + +import org.springframework.lang.Nullable; +import org.springframework.rabbit.stream.listener.ConsumerCustomizer; +import org.springframework.rabbit.stream.listener.StreamListenerContainer; +import org.springframework.rabbit.stream.support.converter.StreamMessageConverter; + +import com.rabbitmq.stream.Codec; +import com.rabbitmq.stream.Environment; + +/** + * Spec for {@link StreamListenerContainer}. + * + * @author Gary Russell + * @since 6.0 + * + */ +public class RabbitStreamMessageListenerContainerSpec extends + MessageListenerContainerSpec { + + RabbitStreamMessageListenerContainerSpec(StreamListenerContainer container) { + this.target = container; + } + + RabbitStreamMessageListenerContainerSpec(Environment environment, @Nullable Codec codec) { + this.target = new StreamListenerContainer(environment, codec); + } + + /** + * Set the Stream queue name; + * Mutually exclusive with {@link #superStream(String, String)}. + * @return this spec. + */ + public RabbitStreamMessageListenerContainerSpec queueName(String queueName) { + return super.queueName(queueName); + } + + /** + * Enable Single Active Consumer on a Super Stream. + * Mutually exclusive with {@link #setQueueName(String...)}. + * @param superStream the stream. + * @param name the consumer name. + * @return this spec. + */ + public RabbitStreamMessageListenerContainerSpec superStream(String superStream, String name) { + this.target.superStream(superStream, name); + return this; + } + + /** + * Set a stream message converter. + * @param converter the converter. + * @return this spec. + */ + public RabbitStreamMessageListenerContainerSpec streamConverter(StreamMessageConverter converter) { + this.target.setStreamConverter(converter); + return this; + } + + /** + * Set a consumer customizer. + * @param customizer the customizer. + * @return this spec. + */ + public RabbitStreamMessageListenerContainerSpec consumerCustomizer(ConsumerCustomizer customizer) { + this.target.setConsumerCustomizer(customizer); + return this; + } + + /** + * @param adviceChain the adviceChain. + * @return the spec. + * @see StreamListenerContainer#setAdviceChain(Advice[]) + */ + public RabbitStreamMessageListenerContainerSpec adviceChain(Advice... adviceChain) { + this.target.setAdviceChain(adviceChain); + return this; + } + + /** + * Perform additional configuration of the container. + * @param consumer a consumer for the container. + * @return this spec. + */ + public RabbitStreamMessageListenerContainerSpec configure(Consumer consumer) { + consumer.accept(this.target); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandler.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandler.java new file mode 100644 index 0000000000..5abe2c3862 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandler.java @@ -0,0 +1,249 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.outbound; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.amqp.support.MappingUtils; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.rabbit.stream.producer.RabbitStreamOperations; +import org.springframework.rabbit.stream.support.StreamMessageProperties; +import org.springframework.util.Assert; + +/** + * {@link MessageHandler} based on {@link RabbitStreamOperations}. + * + * @author Gary Russell + * @author Chris Bono + * @since 6.0 + * + */ +public class RabbitStreamMessageHandler extends AbstractMessageHandler { + + private static final int DEFAULT_CONFIRM_TIMEOUT = 10_000; + + private final RabbitStreamOperations streamOperations; + + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + + private boolean sync; + + private long confirmTimeout = DEFAULT_CONFIRM_TIMEOUT; + + private MessageChannel sendFailureChannel; + + private String sendFailureChannelName; + + private MessageChannel sendSuccessChannel; + + private String sendSuccessChannelName; + + private AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); + + private boolean headersMappedLast; + + /** + * Create an instance with the provided {@link RabbitStreamOperations}. + * @param streamOperations the operations. + */ + public RabbitStreamMessageHandler(RabbitStreamOperations streamOperations) { + Assert.notNull(streamOperations, "'streamOperations' cannot be null"); + this.streamOperations = streamOperations; + } + + /** + * Set the failure channel. After a send failure, an + * {@link org.springframework.messaging.support.ErrorMessage} will be sent + * to this channel with a payload of the exception with the + * failed message. + * @param sendFailureChannel the failure channel. + */ + public void setSendFailureChannel(MessageChannel sendFailureChannel) { + this.sendFailureChannel = sendFailureChannel; + } + + /** + * Set the failure channel name. After a send failure, an + * {@link org.springframework.messaging.support.ErrorMessage} will be sent + * to this channel with a payload of the exception with the + * failed message. + * @param sendFailureChannelName the failure channel name. + */ + public void setSendFailureChannelName(String sendFailureChannelName) { + this.sendFailureChannelName = sendFailureChannelName; + } + + /** + * Set the success channel. + * @param sendSuccessChannel the success channel. + */ + public void setSendSuccessChannel(MessageChannel sendSuccessChannel) { + this.sendSuccessChannel = sendSuccessChannel; + } + + /** + * Set the Success channel name. + * @param sendSuccessChannelName the success channel name. + */ + public void setSendSuccessChannelName(String sendSuccessChannelName) { + this.sendSuccessChannelName = sendSuccessChannelName; + } + + /** + * Set to true to wait for a confirmation. + * @param sync true to wait. + * @see #setConfirmTimeout(long) + */ + public void setSync(boolean sync) { + this.sync = sync; + } + + /** + * Set the confirm timeout. + * @param confirmTimeout the timeout. + * @see #setSync(boolean) + */ + public void setConfirmTimeout(long confirmTimeout) { + this.confirmTimeout = confirmTimeout; + } + + /** + * Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers. + * Defaults to {@link DefaultAmqpHeaderMapper#outboundMapper()}. + * @param headerMapper the {@link AmqpHeaderMapper} to use. + */ + public void setHeaderMapper(AmqpHeaderMapper headerMapper) { + Assert.notNull(headerMapper, "headerMapper must not be null"); + this.headerMapper = headerMapper; + } + + /** + * When mapping headers for the outbound message, determine whether the headers are + * mapped before the message is converted, or afterwards. This only affects headers + * that might be added by the message converter. When false, the converter's headers + * win; when true, any headers added by the converter will be overridden (if the + * source message has a header that maps to those headers). You might wish to set this + * to true, for example, when using a + * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} with a + * String payload that contains json; the converter will set the content type to + * {@code text/plain} which can be overridden to {@code application/json} by setting + * the {@link AmqpHeaders#CONTENT_TYPE} message header. Default: false. + * @param headersMappedLast true if headers are mapped after conversion. + */ + public void setHeadersMappedLast(boolean headersMappedLast) { + this.headersMappedLast = headersMappedLast; + } + + /** + * Return the {@link RabbitStreamOperations}. + * @return the operations. + */ + public RabbitStreamOperations getStreamOperations() { + return this.streamOperations; + } + + protected MessageChannel getSendFailureChannel() { + if (this.sendFailureChannel != null) { + return this.sendFailureChannel; + } + else if (this.sendFailureChannelName != null) { + this.sendFailureChannel = getChannelResolver().resolveDestination(this.sendFailureChannelName); + return this.sendFailureChannel; + } + return null; + } + + protected MessageChannel getSendSuccessChannel() { + if (this.sendSuccessChannel != null) { + return this.sendSuccessChannel; + } + else if (this.sendSuccessChannelName != null) { + this.sendSuccessChannel = getChannelResolver().resolveDestination(this.sendSuccessChannelName); + return this.sendSuccessChannel; + } + return null; + } + + @Override + protected void handleMessageInternal(Message requestMessage) { + CompletableFuture future; + com.rabbitmq.stream.Message streamMessage; + if (requestMessage.getPayload() instanceof com.rabbitmq.stream.Message) { + streamMessage = (com.rabbitmq.stream.Message) requestMessage.getPayload(); + } + else { + MessageConverter converter = this.streamOperations.messageConverter(); + org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter, + this.headerMapper, this.headersMappedLast); + streamMessage = this.streamOperations.streamMessageConverter().fromMessage(amqpMessage); + } + future = this.streamOperations.send(streamMessage); + handleConfirms(requestMessage, future); + } + + private void handleConfirms(Message message, CompletableFuture future) { + future.whenComplete((bool, ex) -> { + if (ex != null) { + MessageChannel failures = getSendFailureChannel(); + if (failures != null) { + this.messagingTemplate.send(failures, new ErrorMessage(ex, message)); + } + } + else { + MessageChannel successes = getSendSuccessChannel(); + if (successes != null) { + this.messagingTemplate.send(successes, message); + } + } + }); + if (this.sync) { + try { + future.get(this.confirmTimeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new MessageHandlingException(message, ex); + } + catch (ExecutionException | TimeoutException ex) { + throw new MessageHandlingException(message, ex); + } + } + } + + private static org.springframework.amqp.core.Message mapMessage(Message message, + MessageConverter converter, AmqpHeaderMapper headerMapper, boolean headersMappedLast) { + + MessageProperties amqpMessageProperties = new StreamMessageProperties(); + return MappingUtils.mapMessage(message, converter, headerMapper, headersMappedLast, headersMappedLast, + amqpMessageProperties); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java index 9259440971..361d57fbb6 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -73,17 +73,39 @@ public final class MappingUtils { * @since 5.1.9 */ public static org.springframework.amqp.core.Message mapReplyMessage(Message replyMessage, - MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode, - boolean headersMappedLast) { + MessageConverter converter, AmqpHeaderMapper headerMapper, + @Nullable MessageDeliveryMode defaultDeliveryMode, boolean headersMappedLast) { return doMapMessage(replyMessage, converter, headerMapper, defaultDeliveryMode, headersMappedLast, true); } private static org.springframework.amqp.core.Message doMapMessage(Message message, - MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode, - boolean headersMappedLast, boolean reply) { + MessageConverter converter, AmqpHeaderMapper headerMapper, + @Nullable MessageDeliveryMode defaultDeliveryMode, boolean headersMappedLast, boolean reply) { MessageProperties amqpMessageProperties = new MessageProperties(); + org.springframework.amqp.core.Message amqpMessage = mapMessage(message, converter, headerMapper, + headersMappedLast, reply, amqpMessageProperties); + checkDeliveryMode(message, amqpMessageProperties, defaultDeliveryMode); + return amqpMessage; + } + + /** + * Map a reply o.s.m.Message to an o.s.a.core.Message. When using a + * {@link ContentTypeDelegatingMessageConverter}, {@link AmqpHeaders#CONTENT_TYPE} and + * {@link MessageHeaders#CONTENT_TYPE} will be used for the selection, with the AMQP + * header taking precedence. + * @param replyMessage the reply message. + * @param converter the message converter to use. + * @param headerMapper the header mapper to use. + * @param headersMappedLast true if headers are mapped after conversion. + * @return the mapped Message. + * @since 6.0 + */ + public static org.springframework.amqp.core.Message mapMessage(Message message, MessageConverter converter, + AmqpHeaderMapper headerMapper, boolean headersMappedLast, boolean reply, + MessageProperties amqpMessageProperties) { + org.springframework.amqp.core.Message amqpMessage; if (!headersMappedLast) { mapHeaders(message.getHeaders(), amqpMessageProperties, headerMapper, reply); @@ -98,7 +120,6 @@ public final class MappingUtils { if (headersMappedLast) { mapHeaders(message.getHeaders(), amqpMessageProperties, headerMapper, reply); } - checkDeliveryMode(message, amqpMessageProperties, defaultDeliveryMode); return amqpMessage; } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java index 816acd3fe4..5d305e5c57 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java @@ -17,6 +17,10 @@ package org.springframework.integration.amqp.dsl; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import java.util.Collections; import java.util.HashMap; @@ -67,9 +71,13 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; +import org.springframework.rabbit.stream.listener.ConsumerCustomizer; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import com.rabbitmq.stream.ConsumerBuilder; +import com.rabbitmq.stream.Environment; + /** * @author Artem Bilan * @author Gary Russell @@ -263,6 +271,17 @@ public class AmqpTests { registration.destroy(); } + @Test + void streamContainer() { + Environment env = mock(Environment.class); + given(env.consumerBuilder()).willReturn(mock(ConsumerBuilder.class)); + RabbitStreamInboundChannelAdapterSpec inboundAdapter = RabbitStream.inboundAdapter(env); + ConsumerCustomizer customizer = mock(ConsumerCustomizer.class); + inboundAdapter.configureContainer(container -> container.consumerCustomizer(customizer)); + inboundAdapter.start(); + verify(customizer).accept(any(), any()); + } + @Configuration @EnableIntegration public static class ContextConfiguration { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandlerTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandlerTests.java new file mode 100644 index 0000000000..711d5eb996 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandlerTests.java @@ -0,0 +1,120 @@ +/* + * Copyright 2021-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.outbound; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; + +import org.springframework.integration.amqp.dsl.RabbitStream; +import org.springframework.integration.amqp.support.RabbitTestContainer; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.rabbit.stream.producer.RabbitStreamTemplate; + +import com.rabbitmq.stream.Address; +import com.rabbitmq.stream.Consumer; +import com.rabbitmq.stream.Environment; +import com.rabbitmq.stream.OffsetSpecification; + +/** + * @author Gary Russell + * @author Chris Bono + * @since 6.0 + */ +public class RabbitStreamMessageHandlerTests implements RabbitTestContainer { + + @Test + void convertAndSend() throws InterruptedException { + Environment env = Environment.builder() + .lazyInitialization(true) + .addressResolver(add -> new Address("localhost", RabbitTestContainer.streamPort())) + .build(); + try { + env.deleteStream("stream.stream"); + } + catch (Exception e) { + } + env.streamCreator().stream("stream.stream").create(); + RabbitStreamTemplate streamTemplate = new RabbitStreamTemplate(env, "stream.stream"); + + RabbitStreamMessageHandler handler = RabbitStream.outboundStreamAdapter(streamTemplate) + .sync(true) + .get(); + + handler.handleMessage(MessageBuilder.withPayload("foo") + .setHeader("bar", "baz") + .build()); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + Consumer consumer = env.consumerBuilder().stream("stream.stream") + .offset(OffsetSpecification.first()) + .messageHandler((context, msg) -> { + received.set(msg); + latch.countDown(); + }) + .build(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(received.get()).isNotNull(); + assertThat(received.get().getBodyAsBinary()).isEqualTo("foo".getBytes()); + assertThat((String) received.get().getApplicationProperties().get("bar")).isEqualTo("baz"); + consumer.close(); + streamTemplate.close(); + } + + @Test + void sendNative() throws InterruptedException { + Environment env = Environment.builder() + .addressResolver(add -> new Address("localhost", RabbitTestContainer.streamPort())) + .lazyInitialization(true) + .build(); + try { + env.deleteStream("stream.stream"); + } + catch (Exception e) { + } + env.streamCreator().stream("stream.stream").create(); + RabbitStreamTemplate streamTemplate = new RabbitStreamTemplate(env, "stream.stream"); + RabbitStreamMessageHandler handler = new RabbitStreamMessageHandler(streamTemplate); + handler.setSync(true); + handler.handleMessage(MessageBuilder.withPayload(streamTemplate.messageBuilder() + .addData("foo".getBytes()) + .applicationProperties().entry("bar", "baz") + .messageBuilder() + .build()) + .build()); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference received = new AtomicReference<>(); + Consumer consumer = env.consumerBuilder().stream("stream.stream") + .offset(OffsetSpecification.first()) + .messageHandler((context, msg) -> { + received.set(msg); + latch.countDown(); + }) + .build(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(received.get()).isNotNull(); + assertThat(received.get().getBodyAsBinary()).isEqualTo("foo".getBytes()); + assertThat((String) received.get().getApplicationProperties().get("bar")).isEqualTo("baz"); + consumer.close(); + streamTemplate.close(); + } + +} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/RabbitTestContainer.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/RabbitTestContainer.java new file mode 100644 index 0000000000..2c9de59d12 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/RabbitTestContainer.java @@ -0,0 +1,57 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.support; + +import java.time.Duration; + +import org.junit.jupiter.api.BeforeAll; +import org.testcontainers.containers.RabbitMQContainer; +import org.testcontainers.junit.jupiter.Testcontainers; + +/** + * Provides a static {@link RabbitMQContainer} that can be shared across test classes. + * + * @author Chris Bono + * @author Gary Russell + */ +@Testcontainers(disabledWithoutDocker = true) +public interface RabbitTestContainer { + + RabbitMQContainer RABBITMQ = new RabbitMQContainer("rabbitmq:management") + .withExposedPorts(5672, 15672, 5552) + .withEnv("RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS", "-rabbitmq_stream advertised_host localhost") + .withPluginsEnabled("rabbitmq_stream") + .withStartupTimeout(Duration.ofMinutes(2)); + + @BeforeAll + static void startContainer() { + RABBITMQ.start(); + } + + static int amqpPort() { + return RABBITMQ.getMappedPort(5672); + } + + static int managementPort() { + return RABBITMQ.getMappedPort(15672); + } + + static int streamPort() { + return RABBITMQ.getMappedPort(5552); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java index aaffea57d1..208e414b9f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java @@ -162,7 +162,6 @@ public class IntegrationManagementConfigurer return this.metricsCaptor; } - @Nullable private void setupObservationRegistry() { if (this.observationRegistry == null && this.observationRegistryProvider != null) { this.observationRegistry = this.observationRegistryProvider.getIfUnique(); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index ddd53a104a..21a67fd485 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1,5 +1,5 @@ [[amqp]] -== AMQP Support +== AMQP (RabbitMQ) Support Spring Integration provides channel adapters for receiving and sending messages by using the Advanced Message Queuing Protocol (AMQP). @@ -29,6 +29,8 @@ The following adapters are available: * <> * <> * <> +* <> +* <> Spring Integration also provides a point-to-point message channel and a publish-subscribe message channel backed by AMQP Exchanges and Queues. @@ -1417,3 +1419,53 @@ In return, that message is retrieved by Spring Integration and printed to the co The following image illustrates the basic set of Spring Integration components used in this sample. .The Spring Integration graph of the AMQP sample image::images/spring-integration-amqp-sample-graph.png[] + +[[rmq-streams]] +=== RabbitMQ Stream Queue Support + +Version 6.0 introduced support for RabbitMQ Stream Queues. + +The DSL factory class for these endpoints is `Rabbit`. + +[[rmq-stream-inbound-channel-adapter]] +==== RabbitMQ Stream Inbound Channel Adapter + +==== +[source, java] +---- +@Bean +IntegrationFlow flow(Environment env) { + @Bean + IntegrationFlow simpleStream(Environment env) { + return IntegrationFlow.from(RabbitStream.inboundAdapter(env) + .configureContainer(container -> container.queueName("my.stream"))) + // ... + .get(); + } + + @Bean + IntegrationFlow superStream(Environment env) { + return IntegrationFlow.from(RabbitStream.inboundAdapter(env) + .configureContainer(container -> container.superStream("my.stream", "my.consumer"))) + // ... + .get(); + } +} +---- +==== + +[[rmq-stream-outbound-channel-adapter]] +==== RabbitMQ Stream Outbound Channel Adapter + +==== +[source, java] +---- +@Bean +IntegrationFlow outbound(RabbitStreamTemplate template) { + return f -> f + // ... + .handle(RabbitStream.outboundStreamAdapter(template)); + +} +---- +==== diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e7f3db9f73..18ddc403b4 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -44,6 +44,11 @@ A `PostgresSubscribableChannel` allows to receive push notifications via `Postgr See <<./jdbc.adoc#postgresql-push,PostgreSQL: Receiving Push Notifications>> for more information. +[[x6.0-rmq]] +==== RabbitMQ Stream Queue Support + +The AMQP module has been enhanced to provide support for inbound and outbound channel adapters using RabbitMQ Stream Queues. +See <<./amqp.adoc#rmq-streams,RabbitMQ Stream Queue Support>> for more information. [[x6.0-general]] === General Changes