updated messaging example for cleaner separation of publishing and converting messages

thombergs · thombergs · commit 5e93dcb84b28 · 2018-08-26T22:58:14.000+02:00
diff --git a/pact/pact-message-consumer/src/main/java/io/reflectoring/MessageConsumerConfiguration.java b/pact/pact-message-consumer/src/main/java/io/reflectoring/MessageConsumerConfiguration.java
@@ -47,13 +47,13 @@ public SimpleMessageListenerContainer container(ConnectionFactory connectionFact
 	}
 
 	@Bean
-	public MessageListenerAdapter listenerAdapter(StringMessageConsumer stringMessageConsumer) {
-		return new MessageListenerAdapter(stringMessageConsumer, "consumeStringMessage");
+	public MessageListenerAdapter listenerAdapter(UserCreatedMessageConsumer userCreatedMessageConsumer) {
+		return new MessageListenerAdapter(userCreatedMessageConsumer, "consumeStringMessage");
 	}
 
 	@Bean
-	public StringMessageConsumer eventReceiver(ObjectMapper objectMapper) {
-		return new StringMessageConsumer(objectMapper);
+	public UserCreatedMessageConsumer eventReceiver(ObjectMapper objectMapper) {
+		return new UserCreatedMessageConsumer(objectMapper);
 	}
 
 }
diff --git a/pact/pact-message-consumer/src/main/java/io/reflectoring/UserCreatedMessageConsumer.java b/pact/pact-message-consumer/src/main/java/io/reflectoring/UserCreatedMessageConsumer.java
@@ -13,13 +13,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class StringMessageConsumer {
+public class UserCreatedMessageConsumer {
 
-  private Logger logger = LoggerFactory.getLogger(StringMessageConsumer.class);
+  private Logger logger = LoggerFactory.getLogger(UserCreatedMessageConsumer.class);
 
   private ObjectMapper objectMapper;
 
-  public StringMessageConsumer(ObjectMapper objectMapper) {
+  public UserCreatedMessageConsumer(ObjectMapper objectMapper) {
     this.objectMapper = objectMapper;
   }
 
@@ -31,7 +31,7 @@ public void consumeStringMessage(String messageString) throws IOException {
     if(!violations.isEmpty()){
       throw new ConstraintViolationException(violations);
     }
-    // do something with the message
+    // pass message into business use case
   }
 
 }
diff --git a/pact/pact-message-consumer/src/test/java/io/reflectoring/UserCreatedMessageConsumerTest.java b/pact/pact-message-consumer/src/test/java/io/reflectoring/UserCreatedMessageConsumerTest.java
@@ -17,14 +17,14 @@
 
 @RunWith(SpringRunner.class)
 @SpringBootTest
-public class StringMessageConsumerTest {
+public class UserCreatedMessageConsumerTest {
 
 	@Rule
 	public MessagePactProviderRule mockProvider = new MessagePactProviderRule(this);
 	private byte[] currentMessage;
 
 	@Autowired
-	private StringMessageConsumer stringMessageConsumer;
+	private UserCreatedMessageConsumer userCreatedMessageConsumer;
 
 	@Pact(provider = "userservice", consumer = "userclient")
 	public MessagePact userCreatedMessagePact(MessagePactBuilder builder) {
@@ -46,7 +46,7 @@ public MessagePact userCreatedMessagePact(MessagePactBuilder builder) {
 	@Test
 	@PactVerification("userCreatedMessagePact")
 	public void verifyCreatePersonPact() throws IOException {
-		stringMessageConsumer.consumeStringMessage(new String(this.currentMessage));
+		userCreatedMessageConsumer.consumeStringMessage(new String(this.currentMessage));
 	}
 
 	/**
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/DemoApplication.java b/pact/pact-message-provider/src/main/java/io/reflectoring/DemoApplication.java
@@ -1,9 +1,7 @@
 package io.reflectoring;
 
-import org.springframework.amqp.rabbit.annotation.EnableRabbit;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.web.servlet.config.annotation.EnableWebMvc;
 
 @SpringBootApplication
 public class DemoApplication {
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/MessageProvider.java b/pact/pact-message-provider/src/main/java/io/reflectoring/MessageProvider.java
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/MessageProviderConfiguration.java b/pact/pact-message-provider/src/main/java/io/reflectoring/MessageProviderConfiguration.java
@@ -1,5 +1,6 @@
 package io.reflectoring;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.context.annotation.Bean;
@@ -8,17 +9,28 @@
 
 @Configuration
 @EnableScheduling
-public class MessageProviderConfiguration {
+class MessageProviderConfiguration {
 
 	@Bean
-	public TopicExchange senderTopicExchange() {
+	TopicExchange topicExchange() {
 		return new TopicExchange("myExchange");
 	}
 
 
 	@Bean
-	public MessageProvider eventPublisher(RabbitTemplate rabbitTemplate, TopicExchange senderTopicExchange) {
-		return new MessageProvider(rabbitTemplate, senderTopicExchange);
+	UserCreatedMessageProvider messageProvider(ObjectMapper objectMapper, UserCreatedMessagePublisher publisher) {
+		return new UserCreatedMessageProvider(objectMapper, publisher);
 	}
 
+	@Bean
+	UserCreatedMessagePublisher messagePublisher(RabbitTemplate rabbitTemplate, TopicExchange topicExchange) {
+		return new UserCreatedMessagePublisher(rabbitTemplate, topicExchange);
+	}
+
+	@Bean
+	SendMessageJob job(UserCreatedMessageProvider messageProvider) {
+		return new SendMessageJob(messageProvider);
+	}
+
+
 }
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/SendMessageJob.java b/pact/pact-message-provider/src/main/java/io/reflectoring/SendMessageJob.java
@@ -0,0 +1,37 @@
+package io.reflectoring;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.UUID;
+
+import org.springframework.scheduling.annotation.Scheduled;
+
+class SendMessageJob {
+
+	private Random random = new Random();
+
+	private UserCreatedMessageProvider messageProvider;
+
+	SendMessageJob(UserCreatedMessageProvider messageProvider) {
+		this.messageProvider = messageProvider;
+	}
+
+	/**
+	 * This scheduled job simulates the "real" business logic that should produce messages.
+	 */
+	@Scheduled(fixedDelay = 1000)
+	void sendUserCreatedMessage() {
+		try {
+			UserCreatedMessage userCreatedMessage = UserCreatedMessage.builder()
+							.messageUuid(UUID.randomUUID().toString())
+							.user(User.builder()
+											.id(random.nextLong())
+											.name("Zaphpod Beeblebrox")
+											.build())
+							.build();
+			messageProvider.sendUserCreatedMessage(userCreatedMessage);
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+}
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/User.java b/pact/pact-message-provider/src/main/java/io/reflectoring/User.java
@@ -2,10 +2,12 @@
 
 import javax.validation.constraints.NotNull;
 
+import lombok.Builder;
 import lombok.Data;
 
 @Data
-public class User {
+@Builder
+class User {
 
 	@NotNull
 	private long id;
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/UserCreatedMessage.java b/pact/pact-message-provider/src/main/java/io/reflectoring/UserCreatedMessage.java
@@ -2,10 +2,12 @@
 
 import javax.validation.constraints.NotNull;
 
+import lombok.Builder;
 import lombok.Data;
 
 @Data
-public class UserCreatedMessage {
+@Builder
+class UserCreatedMessage {
 
 	@NotNull
 	private String messageUuid;
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/UserCreatedMessageProvider.java b/pact/pact-message-provider/src/main/java/io/reflectoring/UserCreatedMessageProvider.java
@@ -0,0 +1,31 @@
+package io.reflectoring;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Takes a {@link UserCreatedMessage}, converts it to a {@link String} and sends it to be published.
+ */
+class UserCreatedMessageProvider {
+
+	private Logger logger = LoggerFactory.getLogger(UserCreatedMessageProvider.class);
+
+	private ObjectMapper objectMapper;
+
+	private UserCreatedMessagePublisher userCreatedMessagePublisher;
+
+	UserCreatedMessageProvider(ObjectMapper objectMapper, UserCreatedMessagePublisher userCreatedMessagePublisher) {
+		this.objectMapper = objectMapper;
+		this.userCreatedMessagePublisher = userCreatedMessagePublisher;
+	}
+
+	void sendUserCreatedMessage(UserCreatedMessage message) throws IOException {
+		String stringMessage = objectMapper.writeValueAsString(message);
+		userCreatedMessagePublisher.publishMessage(stringMessage, "user.created");
+		logger.info("Published message '{}'", stringMessage);
+	}
+
+}
diff --git a/pact/pact-message-provider/src/main/java/io/reflectoring/UserCreatedMessagePublisher.java b/pact/pact-message-provider/src/main/java/io/reflectoring/UserCreatedMessagePublisher.java
@@ -0,0 +1,24 @@
+package io.reflectoring;
+
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+
+/**
+ * Publishes a String message to RabbitMQ.
+ */
+class UserCreatedMessagePublisher {
+
+	private RabbitTemplate rabbitTemplate;
+
+	private TopicExchange topicExchange;
+
+	UserCreatedMessagePublisher(RabbitTemplate rabbitTemplate, TopicExchange topicExchange) {
+		this.rabbitTemplate = rabbitTemplate;
+		this.topicExchange = topicExchange;
+	}
+
+	void publishMessage(String message, String routingKey) {
+		rabbitTemplate.convertAndSend(topicExchange.getName(), routingKey, message);
+	}
+
+}
diff --git a/pact/pact-message-provider/src/test/java/io/reflectoring/MessageProviderTest.java b/pact/pact-message-provider/src/test/java/io/reflectoring/MessageProviderTest.java
diff --git a/pact/pact-message-provider/src/test/java/io/reflectoring/UserCreatedMessageProviderTest.java b/pact/pact-message-provider/src/test/java/io/reflectoring/UserCreatedMessageProviderTest.java
@@ -0,0 +1,58 @@
+package io.reflectoring;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.UUID;
+
+import au.com.dius.pact.provider.PactVerifyProvider;
+import au.com.dius.pact.provider.junit.PactRunner;
+import au.com.dius.pact.provider.junit.Provider;
+import au.com.dius.pact.provider.junit.loader.PactFolder;
+import au.com.dius.pact.provider.junit.target.AmqpTarget;
+import au.com.dius.pact.provider.junit.target.Target;
+import au.com.dius.pact.provider.junit.target.TestTarget;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import static org.mockito.Mockito.*;
+
+@RunWith(PactRunner.class)
+@Provider("userservice")
+@PactFolder("../pact-message-consumer/target/pacts")
+@SpringBootTest
+public class UserCreatedMessageProviderTest {
+
+	@TestTarget
+	public final Target target = new AmqpTarget(Collections.singletonList("io.reflectoring"));
+
+	@Autowired
+	private UserCreatedMessageProvider messageProvider;
+
+	@MockBean
+	private UserCreatedMessagePublisher publisher;
+
+	@PactVerifyProvider("a user created message")
+	public String verifyUserCreatedMessage() throws IOException {
+		// given
+		doNothing().when(publisher).publishMessage(any(String.class), eq("user.created"));
+
+		// when
+		UserCreatedMessage message = UserCreatedMessage.builder()
+						.messageUuid(UUID.randomUUID().toString())
+						.user(User.builder()
+										.id(42L)
+										.name("Zaphod Beeblebrox")
+										.build())
+						.build();
+		messageProvider.sendUserCreatedMessage(message);
+
+		// then
+		ArgumentCaptor<String> messageCapture = ArgumentCaptor.forClass(String.class);
+		verify(publisher, times(1)).publishMessage(messageCapture.capture(), eq("user.created"));
+
+		// returning the message
+		return messageCapture.getValue();
+	}
+}

Original file line number	Diff line number	Diff line change
`@@ -47,13 +47,13 @@ public SimpleMessageListenerContainer container(ConnectionFactory connectionFact`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`@Bean`
`50`		`- public MessageListenerAdapter listenerAdapter(StringMessageConsumer stringMessageConsumer) {`
`51`		`- return new MessageListenerAdapter(stringMessageConsumer, "consumeStringMessage");`
	`50`	`+ public MessageListenerAdapter listenerAdapter(UserCreatedMessageConsumer userCreatedMessageConsumer) {`
	`51`	`+ return new MessageListenerAdapter(userCreatedMessageConsumer, "consumeStringMessage");`
`52`	`52`	`}`
`53`	`53`
`54`	`54`	`@Bean`
`55`		`- public StringMessageConsumer eventReceiver(ObjectMapper objectMapper) {`
`56`		`- return new StringMessageConsumer(objectMapper);`
	`55`	`+ public UserCreatedMessageConsumer eventReceiver(ObjectMapper objectMapper) {`
	`56`	`+ return new UserCreatedMessageConsumer(objectMapper);`
`57`	`57`	`}`
`58`	`58`
`59`	`59`	`}`
Original file line number	Diff line number	Diff line change
`@@ -13,13 +13,13 @@`
`13`	`13`	`import org.slf4j.Logger;`
`14`	`14`	`import org.slf4j.LoggerFactory;`
`15`	`15`
`16`		`-public class StringMessageConsumer {`
	`16`	`+public class UserCreatedMessageConsumer {`
`17`	`17`
`18`		`- private Logger logger = LoggerFactory.getLogger(StringMessageConsumer.class);`
	`18`	`+ private Logger logger = LoggerFactory.getLogger(UserCreatedMessageConsumer.class);`
`19`	`19`
`20`	`20`	`private ObjectMapper objectMapper;`
`21`	`21`
`22`		`- public StringMessageConsumer(ObjectMapper objectMapper) {`
	`22`	`+ public UserCreatedMessageConsumer(ObjectMapper objectMapper) {`
`23`	`23`	`this.objectMapper = objectMapper;`
`24`	`24`	`}`
`25`	`25`
`@@ -31,7 +31,7 @@ public void consumeStringMessage(String messageString) throws IOException {`
`31`	`31`	`if(!violations.isEmpty()){`
`32`	`32`	`throw new ConstraintViolationException(violations);`
`33`	`33`	`}`
`34`		`- // do something with the message`
	`34`	`+ // pass message into business use case`
`35`	`35`	`}`
`36`	`36`
`37`	`37`	`}`