In this tutorial, we’ll start by exploring the configuration that’s needed to setup a message channel via Spring Cloud Stream, and we’ll look at how a message is produced and sent. Finally, we’ll create a simple message listener and discuss how enforce idempotent message consumption.
Throughout this tutorial, we’ll specifically focus on the message that is sent when a checkout is successfully completed. That said, the patterns that we’ll look at are generally applicable throughout the Broadleaf Microservices ecosystem.
Much of building a message producer involves setting up simple Spring Cloud Stream configuration with the following components:
@Output binding targetimport org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
public interface CheckoutCompletionProducer {
String CHANNEL = "checkoutCompletionOutput";
@Output(CHANNEL)
MessageChannel checkoutCompletionOutput();
}@EnableBinding...
import org.springframework.cloud.stream.annotation.EnableBinding;
...
@Configuration
@EnableBinding(CheckoutCompletionProducer.class)
public static class CartCheckoutMessagingAutoConfiguration {}spring:
cloud:
stream:
bindings:
checkoutCompletionOutput:
destination: checkoutCompletion|
Note
|
|
With the output binding in place, the CheckoutCompletionProducer interface can now be injected into any Spring component that needs to send a message via this channel.
Let’s take a look at the DefaultCheckoutService to understand how messages are built and sent.
package com.broadleafcommerce.cartoperation.service.checkout;
import static com.broadleafcommerce.common.messaging.service.DefaultMessageLockService.MESSAGE_IDEMPOTENCY_KEY;
import org.springframework.integration.support.MessageBuilder;
...
@RequiredArgsConstructor
public class DefaultCheckoutService implements CheckoutService {
...
@Getter(AccessLevel.PROTECTED)
private final CheckoutCompletionProducer checkoutCompletionProducer; // (1)
...
/**
* Send a message to notify external services (and internal listeners) of the completed checkout
* so that they can react accordingly.
*
* @param cart The cart that has completed checkout. This should be the primary content of the
* out-going message.
* @param requestId The id representing this request to checkout
* @param contextInfo Context information around sandbox and multi-tenant state.
*/
protected void sendCheckoutCompletionMessage(@lombok.NonNull Cart cart,
@lombok.NonNull String requestId,
@Nullable ContextInfo contextInfo) {
try {
Assert.notNull(checkoutCompletionProducer,
"CheckoutCompletionProducer was not available for sending a " +
"message for checkout completion. Likely a configuration issue in " +
"which messaging for this component is disabled.");
CheckoutCompletionEvent completionEvent =
typeFactory.get(CheckoutCompletionEvent.class);
completionEvent.setCart(cart);
completionEvent.setRequestId(requestId);
completionEvent.setContextInfo(contextInfo);
checkoutCompletionProducer.checkoutCompletionOutput()
.send(MessageBuilder.withPayload(completionEvent) // (2) (3)
.setHeaderIfAbsent(MESSAGE_IDEMPOTENCY_KEY, cart.getId()) // (4)
.build());
} catch (Exception e) {
String errorMessage = String.format(
"Failed to send checkout completion message for cart (order number: %s).",
cart.getOrderNumber());
throw new CheckoutCompletionMessageException(errorMessage, e, cart);
}
}
}Inject the CheckoutCompletionProducer
Leverage the MessageBuilder to produce a Message (org.springframework.messaging.Message) with a CheckoutCompletionEvent payload and an idempotency header
Send the message by calling checkoutCompletionProducer.checkoutCompletionOutput().send(…)
Include the MESSAGE_IDEMPOTENCY_KEY header with the cart id, doing so ensures the message for this cart only gets processed once. See Idempotent Message Consumption in Detail.
Similar to the creation of a message producer, creating a message listener also involves setting up a few similar components:
@Input binding targetimport org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
public interface MyCheckoutCompletionConsumer {
String CHANNEL = "myCheckoutCompletionInput";
@Input(CHANNEL)
SubscribableChannel myCheckoutCompletionInput();
}@EnableBinding@Configuration
@EnableBinding(OfferCheckoutCompletionConsumer.class)
public class MyMessagingConfiguration {
...
}spring:
cloud:
stream:
bindings:
myCheckoutCompletionInput:
group: my-checkout-completion-group
destination: checkoutCompletion|
Note
|
|
Listening for messages on the input channel can be achieved using the following pattern:
import org.springframework.cloud.stream.annotation.StreamListener;
...
@RequiredArgsConstructor
@DataRouteByKey("MY_SERVICE_DATA_ROUTE_KEY") // (1)
public class MyCheckoutCompletionListener {
@Getter(AccessLevel.PROTECTED)
private final IdempotentMessageConsumptionService idempotentConsumptionService;
@StreamListener(MyCheckoutCompletionConsumer.CHANNEL) // (2)
public void listen(Message<CheckoutCompletionEvent> message) { // (3)
idempotentConsumptionService.consumeMessage(message, // (4)
CheckoutCompletionListener.class.getSimpleName(), this::processMessage); // (5)
}
protected void processMessage(Message<CheckoutCompletionEvent> message) {
CheckoutCompletionEvent event = message.getPayload();
// Your logic goes here
}
}Leverage @DataRouteByKey or @DataRouteByExample to declare the data route that should be used by the IdempotentMessageConsumptionService.
Use the @StreamListener annotation to declare the channel that you’d like to listen to.
Declare that you expect to receive a message with the CheckoutCompletionEvent payload. This will cause the payload to be deserialized into the CheckoutCompletionEvent object.
Leverage the IdempotentMessageConsumptionService to ensure that you don’t process the same message twice. Spring Cloud Stream will deliver messages at least once, so you need to be sure that the message is only processed once.
If the idempotency check passes, then defer to another method for processing of the message.
|
Note
|
Throwing an exception while processing the message and allowing it to bubble up to the @StreamListener proxy will engage Spring Cloud Stream’s retry and dead letter queue mechanisms.
|
From there, we just need to register a bean for this class.
@Bean
public MyCheckoutCompletionListener myCheckoutCompletionEventListener(
IdempotentMessageConsumptionService idempotentMessageConsumptionService) {
return new MyCheckoutCompletionListener(idempotentMessageConsumptionService);
}As previously mentioned, Spring Cloud Stream will deliver messages at least once, so you need to be mindful of not processing the same message twice. The primary mechanisms that we use to avoid duplicate processing are the IdempotentMessageConsumptionService and the MESSAGE_IDEMPOTENCY_KEY message header.
When each message is sent, it should include the MESSAGE_IDEMPOTENCY_KEY header with a value representing a unique identifier for the set of work that is to be completed. For example, the checkout completion messages use the cart id. For other messages a random UUID is sufficient.
When a message is passed to the IdempotentMessageConsumptionService, we gather the MESSAGE_IDEMPOTENCY_KEY value and check to see if the key has been persisted to the blc_resource_lock table (schema identified by the @DataRouteByKey annotation on the listener). If a lock is not present, then the idempotency check will pass and the message will be processed. Otherwise, the message will be ignored.
|
Note
|
If the same message payload is sent multiple times with different MESSAGE_IDEMPOTENCY_KEY values, then the message will be processed multiple times.
|