AWS Compute Blog
Implementing enterprise integration patterns with AWS messaging services: publish-subscribe channels
This post is courtesy of Christian Mueller, Sr. Solutions Architect, AWS and Dirk Fröhner, Sr. Solutions Architect, AWS
In this blog, we look at the second part of some fundamental enterprise integration patterns and how you can implement them with AWS messaging services. If you missed the first part, we encourage you to start there.
Read Part 1: Point-to-Point Messaging
Integration patterns
Messaging channels: publish-subscribe
As mentioned in the first blog, we continue with the second major messaging channel pattern: publish-subscribe.
A publish-subscribe channel is usually implemented using message topics. In this model, any message published to a topic is immediately received by all of the subscribers of the topic (unless you have applied the message filter pattern). However, if there is no subscriber, messages are usually discarded. The durable subscriber pattern describes an exception where messages are kept for a while in case the subscriber is offline. Publish-subscribe is used when multiple parties are interested in certain messages. Sometimes, this pattern is also referred to as fan-out.
Let’s apply this pattern to the different AWS messaging services and get our hands dirty. To follow our examples, sign in to your AWS account (or create an account as described in How do I create and activate a new Amazon Web Services account?).
Integration scenarios
Publish-subscribe channels: one-way messaging
Publish-subscribe one-way patterns are often involved in notification style use cases, where the publisher sends out an event and doesn’t care who is interested in this event. For example, Amazon CloudWatch Events publishes state changes in the environment, and you can subscribe and act accordingly.
The diagrams in the following subsections show the principles of one-way messaging for publish-subscribe channels, using both Amazon MQ and Amazon SNS topics. A publisher produces a message and sends it into a topic, and subscribers consume the message from the topic for processing.
For traditional messaging, senders and consumers can use API protocols such JMS or AMQP. For cloud-native messaging, they can use the Amazon SNS API.
Traditional messaging
In this example, we reuse the Amazon MQ broker we set up in part one of this blog. As we can see in the following diagram, messages as published into an Amazon MQ topic and multiple subscribers can consume messages from it.
This example is similar to the point-to-point one-way traditional example using the Apache Active MQ client library, but we use topics instead of queues, as shown in the following code.
public class PublishSubscribeOneWayTraditional {
public static void main(String... args) throws Exception {
ActiveMQSslConnectionFactory connFact = new ActiveMQSslConnectionFactory("failover:(ssl://<broker-1>.amazonaws.com:61617,ssl://<broker-2>.amazonaws.com:61617)");
connFact.setConnectResponseTimeout(10000);
Connection conn = connFact.createConnection("user", "password");
conn.setClientID("PubSubOneWayTraditional");
conn.start();
new Thread(new Subscriber(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Topic.PubSub.OneWay.Traditional")).start();
new Thread(new Publisher(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Topic.PubSub.OneWay.Traditional")).start();
}
public static class Publisher implements Runnable {
private Session session;
private String destination;
public Sender(Session session, String destination) {
this.session = session;
this.destination = destination;
}
public void run() {
try {
MessageProducer messageProducer = session.createProducer(session.createTopic(destination));
long counter = 0;
while (true) {
TextMessage message = session.createTextMessage("Message " + ++counter);
message.setJMSMessageID(UUID.randomUUID().toString());
messageProducer.send(message);
}
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
public static class Subscriber implements Runnable, MessageListener {
private Session session;
private String destination;
public Receiver(Session session, String destination) {
this.session = session;
this.destination = destination;
}
public void run() {
try {
MessageConsumer consumer = session.createDurableSubscriber(session.createTopic(destination), "subscriber-1");
consumer.setMessageListener(this);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public void onMessage(Message message) {
try {
System.out.println(String.format("received message '%s' with message id '%s'", ((TextMessage) message).getText(), message.getJMSMessageID()));
message.acknowledge();
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
}Cloud-native messaging
To follow a similar example using Amazon SNS, open the Amazon SNS console and create an Amazon SNS topic named PubSubOneWayCloudNative. The below diagram illustrates that a publisher sends messages into an Amazon SNS topic which are consumed by subscribers of this topic.
We use the AWS SDK for Java to send messages to our Amazon SNS topic, running in an endless loop. You can run the following code on every Amazon compute service, your on-premises data center, or your personal computer.
public class PublishSubscribeOneWayCloudNative {
public static void main(String... args) throws Exception {
final AmazonSNS sns = AmazonSNSClientBuilder.standard().build();
new Thread(new Publisher(sns, "arn:aws:sns:<region>:<account-number>:PubSubOneWayCloudNative")).start();
}
public static class Publisher implements Runnable {
private AmazonSNS sns;
private String destination;
public Sender(AmazonSNS sns, String destination) {
this.sns = sns;
this.destination = destination;
}
public void run() {
long counter = 0;
while (true) {
sns.publish(
new PublishRequest()
.withTargetArn(destination)
.withSubject("PubSubOneWayCloudNative sample")
.withMessage("Message " + ++counter)
.addMessageAttributesEntry("MessageID", new MessageAttributeValue().withDataType("String").withStringValue(UUID.randomUUID().toString())));
}
}
}
}The subscriber is implemented as an AWS Lambda function, using Amazon SNS as the event source. For more information on how to set this up, see Using Amazon SNS for System-to-System Messaging with a Lambda Function as a Subscriber.
public class Subscriber implements RequestHandler<SNSEvent, Void> {
@Override
public Void handleRequest(SNSEvent request, Context context) {
for (SNSEvent.SNSRecord record: request.getRecords()) {
SNS sns = record.getSNS();
System.out.println(String.format("received message '%s' with message id '%s'", sns.getMessage(), sns.getMessageAttributes().get("MessageID").getValue()));
}
return null;
}
}Publish-subscribe channels: request-response messaging
Publish-subscribe request-response patterns are beneficial in use cases where it’s important to communicate with multiple services that do their work in parallel, but all their responses need to be aggregated afterward. One example is an order service, which needs to enrich the order message with data from multiple backend services.
The diagrams in the following subsections show the principles of request-response messaging for publish-subscribe channels, using both Amazon MQ and Amazon SNS topics. A publisher produces a message and sends it into a topic, and subscribers consume the message from the topic for processing.
Although we use a publish-subscribe channel for the request messages, we would usually use a point-to-point channel for the response messages. This assumes that the requester application or at least a dedicated application is the one entity that works on processing all the responses.
Traditional messaging
As we can see in the following diagram, a Amazon MQ topic is used to send out all the request messages, while all the response messages are sent into an Amazon MQ queue.
In our code sample below, we use two responders.
public class PublishSubscribeRequestResponseTraditional {
public static void main(String... args) throws Exception {
ActiveMQSslConnectionFactory connFact = new ActiveMQSslConnectionFactory("failover:(ssl://<broker-1>.amazonaws.com:61617,ssl://<broker-2>.amazonaws.com:61617)");
connFact.setConnectResponseTimeout(10000);
Connection conn = connFact.createConnection("user", "password");
conn.setClientID("PubSubReqRespTraditional");
conn.start();
new Thread(new Responder(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Topic.PubSub.ReqResp.Traditional", "subscriber-1")).start();
new Thread(new Responder(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Topic.PubSub.ReqResp.Traditional", "subscriber-2")).start();
new Thread(new Requester(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Topic.PubSub.ReqResp.Traditional")).start();
}
public static class Requester implements Runnable {
private Session session;
private String destination;
public Requester(Session session, String destination) {
this.session = session;
this.destination = destination;
}
public void run() {
MessageProducer messageProducer = null;
try {
messageProducer = session.createProducer(session.createTopic(destination));
long counter = 0;
while (true) {
TemporaryQueue replyTo = session.createTemporaryQueue();
String correlationId = UUID.randomUUID().toString();
TextMessage message = session.createTextMessage("Message " + ++counter);
message.setJMSMessageID(UUID.randomUUID().toString());
message.setJMSCorrelationID(correlationId);
message.setJMSReplyTo(replyTo);
messageProducer.send(message);
MessageConsumer consumer = session.createConsumer(replyTo, "JMSCorrelationID='" + correlationId + "'");
try {
Message receivedMessage1 = consumer.receive(5000);
Message receivedMessage2 = consumer.receive(5000);
System.out.println(String.format("received 2 messages '%s' and '%s'", ((TextMessage) receivedMessage1).getText(), ((TextMessage) receivedMessage2).getText()));
receivedMessage2.acknowledge();
} finally {
if (consumer != null) {
consumer.close();
}
}
}
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
public static class Responder implements Runnable, MessageListener {
private Session session;
private String destination;
private String name;
public Responder(Session session, String destination, String name) {
this.session = session;
this.destination = destination;
this.name = name;
}
public void run() {
try {
MessageConsumer consumer = session.createDurableSubscriber(session.createTopic(destination), name);
consumer.setMessageListener(this);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public void onMessage(Message message) {
try {
String correlationId = message.getJMSCorrelationID();
Destination replyTo = message.getJMSReplyTo();
TextMessage responseMessage = session.createTextMessage(((TextMessage) message).getText() + " from responder " + name);
responseMessage.setJMSMessageID(UUID.randomUUID().toString());
responseMessage.setJMSCorrelationID(correlationId);
MessageProducer messageProducer = session.createProducer(replyTo);
try {
messageProducer.send(responseMessage);
message.acknowledge();
} finally {
if (messageProducer != null) {
messageProducer.close();
}
}
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
}Cloud-native messaging
To implement a similar pattern with Amazon SNS, open the Amazon SNS console and create a new SNS topic named PubSubReqRespCloudNative. Then open the Amazon SQS console and create a standard SQS queue named PubSubReqRespCloudNative-Resp. The following diagram illustrates that we now use an Amazon SNS topic for request messages and an Amazon SQS queue for response messages.
This example requester is almost identical to the publish-subscribe one-way cloud-native example sender. The requester also specifies a reply-to address and a correlation ID as message attributes. This way, responders know where to send the responses to, and the receiver of the responses can assign them accordingly.
public class PublishSubscribeReqRespCloudNative {
public static void main(String... args) throws Exception {
final AmazonSNS sns = AmazonSNSClientBuilder.standard().build();
final AmazonSQS sqs = AmazonSQSClientBuilder.standard().build();
new Thread(new Requester(sns, sqs, "arn:aws:sns:<region>:<account-number>:PubSubReqRespCloudNative", "https://sqs.<region>.amazonaws.com/<account-number>/PubSubReqRespCloudNative-Resp")).start();
}
public static class Requester implements Runnable {
private AmazonSNS sns;
private AmazonSQS sqs;
private String destination;
private String replyDestination;
private Map<String, PublishRequest> inflightMessages = new ConcurrentHashMap<>();
public Requester(AmazonSNS sns, AmazonSQS sqs, String destination, String replyDestination) {
this.sns = sns;
this.sqs = sqs;
this.destination = destination;
this.replyDestination = replyDestination;
}
public void run() {
long counter = 0;
while (true) {
String correlationId = UUID.randomUUID().toString();
PublishRequest request = new PublishRequest()
.withTopicArn(destination)
.withMessage("Message " + ++counter)
.addMessageAttributesEntry("CorrelationID", new MessageAttributeValue().withDataType("String").withStringValue(correlationId))
.addMessageAttributesEntry("ReplyTo", new MessageAttributeValue().withDataType("String").withStringValue(replyDestination));
sns.publish(request);
inflightMessages.put(correlationId, request);
ReceiveMessageResult receiveMessageResult = sqs.receiveMessage(
new ReceiveMessageRequest()
.withQueueUrl(replyDestination)
.withMessageAttributeNames("CorrelationID")
.withMaxNumberOfMessages(5)
.withWaitTimeSeconds(2));
for (Message receivedMessage : receiveMessageResult.getMessages()) {
System.out.println(String.format("received message '%s' with message id '%s'", receivedMessage.getBody(), receivedMessage.getMessageId()));
String receivedCorrelationId = receivedMessage.getMessageAttributes().get("CorrelationID").getStringValue();
PublishRequest originalRequest = inflightMessages.remove(receivedCorrelationId);
System.out.println(String.format("Corresponding request message '%s'", originalRequest.getMessage()));
sqs.deleteMessage(
new DeleteMessageRequest()
.withQueueUrl(replyDestination)
.withReceiptHandle(receivedMessage.getReceiptHandle()));
}
}
}
}
}This example responder is almost identical to the publish-subscribe one-way cloud-native example receiver. It also creates a message, enriches it with the correlation ID, and sends it back to the reply-to address provided in the received message.
public class Responder implements RequestHandler<SNSEvent, Void> {
private final AmazonSQS sqs = AmazonSQSClientBuilder.standard().build();
@Override
public Void handleRequest(SNSEvent request, Context context) {
for (SNSEvent.SNSRecord record: request.getRecords()) {
System.out.println(String.format("received record '%s' with message id '%s'", record.getSNS().getMessage(), record.getSNS().getMessageId()));
String correlationId = record.getSNS().getMessageAttributes().get("CorrelationID").getValue();
String replyTo = record.getSNS().getMessageAttributes().get("ReplyTo").getValue();
System.out.println(String.format("sending message with correlation id '%s' to '%s'", correlationId, replyTo));
sqs.sendMessage(
new SendMessageRequest()
.withQueueUrl(replyTo)
.withMessageBody(record.getSNS().getMessage() + " with CorrelationID " + correlationId)
.addMessageAttributesEntry("CorrelationID", new MessageAttributeValue().withDataType("String").withStringValue(correlationId)));
}
return null;
}
}Go Build!
We look forward to hearing about what you build and will continue innovating our services on your behalf.