* added messages publisher

Author: VolhvPorechja

app.iml

@@ -26,6 +26,8 @@
     <orderEntry type="library" name="Maven: com.fasterxml.jackson.core:jackson-databind:2.9.5" level="project" />
     <orderEntry type="library" name="Maven: com.fasterxml.jackson.core:jackson-annotations:2.9.0" level="project" />
     <orderEntry type="library" name="Maven: com.fasterxml.jackson.core:jackson-core:2.9.5" level="project" />
+    <orderEntry type="library" name="Maven: com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.9.5" level="project" />
+    <orderEntry type="library" name="Maven: org.yaml:snakeyaml:1.18" level="project" />
     <orderEntry type="library" scope="PROVIDED" name="Maven: org.projectlombok:lombok:1.16.8" level="project" />
     <orderEntry type="library" name="Maven: com.rabbitmq:amqp-client:5.2.0" level="project" />
     <orderEntry type="library" name="Maven: org.slf4j:slf4j-api:1.7.25" level="project" />

docker-compose.yml

@@ -0,0 +1,9 @@
+version: '3'
+services:
+  queue:
+    image: rabbitmq:3.7-rc-management
+    ports:
+        - "5672:5672"
+        - "15672:15672"
+        - "25672:25672"
+        - "4369:4369"

pom.xml

@@ -8,6 +8,8 @@
 
     <properties>
         <akkastreams.version>2.5.11</akkastreams.version>
+        <jackson.version>2.9.5</jackson.version>
+        <rabbitmq.version>5.2.0</rabbitmq.version>
     </properties>
 
     <dependencies>
@@ -33,7 +35,12 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.9.5</version>
+            <version>${jackson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <version>${jackson.version}</version>
         </dependency>
 
         <!-- LOMBOK -->
@@ -48,7 +55,7 @@
         <dependency>
             <groupId>com.rabbitmq</groupId>
             <artifactId>amqp-client</artifactId>
-            <version>5.2.0</version>
+            <version>${rabbitmq.version}</version>
         </dependency>
     </dependencies>
     <build>

settings.yml

@@ -0,0 +1,5 @@
+host: 192.168.99.100
+port: 5672
+user: guest
+pass: guest
+queue: hello

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/FetchStartMessage.java

@@ -4,6 +4,10 @@
 import lombok.Builder;
 import lombok.Getter;
 
+/**
+ * Сообщение говорящее о том, что пора зачинать поток заданий
+ * содержит все параметры, чтобы начать это делать.
+ */
 @Getter
 @AllArgsConstructor
 @Builder

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/FetcherInitMessage.java

@@ -4,6 +4,10 @@
 import lombok.Builder;
 import lombok.Getter;
 
+/**
+ * Инициализирующее сообщние
+ * содержит параметры в какой начальной конфигурации надо инициализироваться систему
+ */
 @Getter
 @AllArgsConstructor
 @Builder

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/GreetingConfigMessage.java

@@ -4,6 +4,9 @@
 import lombok.Getter;
 import lombok.Setter;
 
+/**
+ * Сообщение по реконфигурации шаблонов печати
+ */
 @Getter
 @Setter
 @AllArgsConstructor

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/GreetingMessage.java

@@ -3,6 +3,9 @@
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 
+/**
+ * Сообщение для печати
+ */
 @Getter
 @AllArgsConstructor
 public class GreetingMessage {

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/IncomingMessage.java

@@ -0,0 +1,6 @@
+package tech.volhvporechja.akka.demo.actors.Contracts;
+
+public class MessagesTypes {
+	public static final String WELCOME = "welcome";
+	public static final String CONFIG = "config";
+}

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/MessagesTypes.java

@@ -0,0 +1,25 @@
+package tech.volhvporechja.akka.demo.actors.Contracts;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Контракт сообщения получаемого из RabbitMQ
+ */
+@Getter
+@Setter
+@AllArgsConstructor
+@Builder
+public class QueueIncomingMessage {
+	private String type;
+	private String load;
+
+	public byte[] marshall() throws JsonProcessingException {
+		ObjectMapper mapper = new ObjectMapper();
+		return mapper.writeValueAsString(this).getBytes();
+	}
+}

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/QueueIncomingMessage.java

@@ -3,6 +3,9 @@
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 
+/**
+ * Сообщение для шаблона и последующей печати
+ */
 @Getter
 @AllArgsConstructor
 public class WelcomeMessage {

src/main/java/tech/volhvporechja/akka/demo/actors/Contracts/WelcomeMessage.java

@@ -22,38 +22,47 @@ public class FetcherActor extends AbstractActor {
 	private Router broadcastRouter;
 	private Router router;
 
+	// Blueprint нашего аткора, то как его следует инстанцировать
 	static public Props props() {
 		return Props.create(FetcherActor.class, FetcherActor::new);
 	}
 
 	/**
-	 * Processor for actors system initialization
+	 * Логика инициалиазции нашей системы
+	 * Передаем сообщение в котором храниться конфигурация нашей системы
 	 *
 	 * @param config initialization config
 	 */
-	private void intializePrinterSystem(FetcherInitMessage config) {
+	private void initializePrinterSystem(FetcherInitMessage config) {
+		// готовим наши "принтеры"
 		List<Routee> printerRoutees = new ArrayList<>();
 		for (int i = 0; i < config.getPrintersCount(); i++) {
 			ActorRef printer = getContext().actorOf(PrinterActor.props(i), "printer-" + i);
 			getContext().watch(printer);
 			printerRoutees.add(new ActorRefRoutee(printer));
 		}
 
+		// готовим наши шаблонизаторы
 		Router printerRouter = new Router(new RoundRobinRoutingLogic(), printerRoutees);
-
 		List<Routee> greeterRoutees = new ArrayList<>();
 		for (int i = 0; i < config.getGreetersCount(); i++) {
 			ActorRef greeter = getContext().actorOf(GreeterActor.props("Aloha, %s!!", printerRouter, i), "greeter-" + i);
 			getContext().watch(greeter);
 			greeterRoutees.add(new ActorRefRoutee(greeter));
 		}
 
+		// Роутер для распространения конфигурационного сообщения
 		broadcastRouter = new Router(new BroadcastRoutingLogic(), greeterRoutees);
+
+		// Роутер для отправки сообщений предназначенных для шаблонизации и печати
 		router = new Router(new RoundRobinRoutingLogic(), greeterRoutees);
 	}
 
 	/**
-	 * Processor for listening for queue
+	 * Метод прослушивания очереди
+	 * !! Disclaimer: используется блокирующее API для доступа к RabbitMQ
+	 * поток, который будет исполнять этот код будет потерян для пула.
+	 * Но так как такой актор только один, для демонстрационных целей - ничего страшного
 	 *
 	 * @param config listening config
 	 */
@@ -78,18 +87,18 @@ public void handleDelivery(String consumerTag, Envelope envelope,
 					String message = new String(body, "UTF-8");
 					log.info(" [x] Received '" + message + "'");
 
-					IncomingMessage request = null;
+					QueueIncomingMessage request = null;
 					try {
 						ObjectMapper mapper = new ObjectMapper();
-						request = mapper.readValue(message, IncomingMessage.class);
+						request = mapper.readValue(message, QueueIncomingMessage.class);
 					} catch (Exception ex) {
 						log.error(ex, "Unable to parse request.");
 					}
 
 					if (request != null)
-						if (request.getType().equals("welcome"))
+						if (request.getType().equals(MessagesTypes.WELCOME))
 							router.route(new WelcomeMessage(request.getLoad()), getSelf());
-						else if (request.getType().equals("config"))
+						else if (request.getType().equals(MessagesTypes.CONFIG))
 							broadcastRouter.route(new GreetingConfigMessage(request.getLoad()), getSelf());
 						else
 							log.warning("Unknown message type");
@@ -103,12 +112,13 @@ else if (request.getType().equals("config"))
 		}
 	}
 
+	// Билдер нашего обраотчика сообщений
 	@Override
 	public Receive createReceive() {
 		return receiveBuilder()
-				.match(FetchStartMessage.class, this::listen)
-				.match(FetcherInitMessage.class, this::intializePrinterSystem)
-				.matchAny(o -> log.info("received unknown message"))
+				.match(FetchStartMessage.class, this::listen) // Обработчик сообщения начала процесса прослушивания заданий на печать
+				.match(FetcherInitMessage.class, this::initializePrinterSystem) // Обработчик сообщения инициализации системы
+				.matchAny(o -> log.info("received unknown message")) // Обработчик неизвестного сообщения
 				.build();
 	}
 }

src/main/java/tech/volhvporechja/akka/demo/actors/FetcherActor.java

@@ -9,16 +9,33 @@
 import tech.volhvporechja.akka.demo.actors.Contracts.GreetingMessage;
 import tech.volhvporechja.akka.demo.actors.Contracts.WelcomeMessage;
 
+/**
+ * Наш шаблонизатор
+ * на борту: шаблон и логика обработки двух типов сообщений.
+ */
 public class GreeterActor extends AbstractActor {
 	private final int id;
 	private LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
 	private String messageTemplate;
 	private Router printer;
 
-	static public Props props(String message, Router printer, int id) {
-		return Props.create(GreeterActor.class, () -> new GreeterActor(message, printer, id));
+	/**
+	 * Blueprint, описывающий как надо собирать нашего Актора
+	 * @param template печатная форма
+	 * @param printersRouter роутер, дающий нам доступ к принтерам
+	 * @param id идентификтор эксземпляра этого шаблонизатора
+	 * @return
+	 */
+	static public Props props(String template, Router printersRouter, int id) {
+		return Props.create(GreeterActor.class, () -> new GreeterActor(template, printersRouter, id));
 	}
 
+	/**
+	 * JUST CTOR
+	 * @param messageTemplate
+	 * @param router
+	 * @param id
+	 */
 	public GreeterActor(String messageTemplate, Router router, int id) {
 		this.messageTemplate = messageTemplate;
 		this.printer = router;
@@ -28,15 +45,15 @@ public GreeterActor(String messageTemplate, Router router, int id) {
 	@Override
 	public Receive createReceive() {
 		return receiveBuilder()
-				.match(GreetingConfigMessage.class, wtg -> { // Reconfiguration
+				.match(GreetingConfigMessage.class, wtg -> { // Реконфигурируем
 					log.info(String.format("GREETER-%s: Reconfiguring greeter [%s]->[%s]", id, messageTemplate, wtg.getFormatString()));
 					messageTemplate = wtg.getFormatString();
 				})
-				.match(WelcomeMessage.class, wtg -> { // Execution
+				.match(WelcomeMessage.class, wtg -> { // Рендерим печатную псевдоформу с шаблоном
 					log.info(String.format("GREETER-%s: Received data [%s]", id, wtg.getWho()));
 					printer.route(new GreetingMessage(String.format(messageTemplate, wtg.getWho())), getSelf());
 				})
-				.matchAny(o -> log.info("received unknown message"))
+				.matchAny(o -> log.info("received unknown message")) // Обработчик неизвестного сообщения
 				.build();
 	}
 }

src/main/java/tech/volhvporechja/akka/demo/actors/GreeterActor.java

@@ -0,0 +1,56 @@
+package tech.volhvporechja.akka.demo.actors;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import tech.volhvporechja.akka.demo.actors.Contracts.MessagesTypes;
+import tech.volhvporechja.akka.demo.actors.Contracts.QueueIncomingMessage;
+import tech.volhvporechja.akka.demo.actors.config.RabbitMQConfig;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
+public class MessagesPublisher {
+
+	private static String[] names = {"Alex", "Ben", "Carrol", "Diana", "Evan", "Martin Fawler"};
+
+	public static void main(String[] args) throws IOException, TimeoutException {
+		RabbitMQConfig config = RabbitMQConfig.Load("settings.yml");
+
+		ConnectionFactory factory = new ConnectionFactory();
+		factory.setHost(config.getHost());
+		factory.setPort(config.getPort());
+		factory.setUsername(config.getUser());
+		factory.setPassword(config.getPass());
+
+		try (Connection conn = factory.newConnection()) {
+			try (Channel channel = conn.createChannel()) {
+				channel.queueDeclare(config.getQueue(), false, false, false, null);
+
+				QueueIncomingMessage.QueueIncomingMessageBuilder welcomeMessageBuilder = QueueIncomingMessage.builder().
+						type(MessagesTypes.WELCOME);
+				for (int i = 0; i < 5e6; i++) {
+					final byte[] message = welcomeMessageBuilder
+							.load(names[i % names.length])
+							.build().marshall();
+
+					channel.basicPublish("", config.getQueue(), null, message);
+				}
+
+				byte[] reconfig = QueueIncomingMessage.builder()
+						.type(MessagesTypes.CONFIG)
+						.load("Heyyo, %s")
+						.build().marshall();
+				channel.basicPublish("", config.getQueue(), null, reconfig);
+
+				for (int i = 0; i < 5e6; i++) {
+					final byte[] message = welcomeMessageBuilder
+							.load(names[i % names.length])
+							.build().marshall();
+
+					channel.basicPublish("", config.getQueue(), null, message);
+				}
+			}
+		}
+	}
+}

src/main/java/tech/volhvporechja/akka/demo/actors/MessagesPublisher.java

@@ -31,10 +31,11 @@ public Receive createReceive() {
 				})
 				.match(ReceiveTimeout.class, r -> {
 					log.info("Soooo boooring!!");
-					// Switch off
+					// Switch off - просто висячий код, который позволяет выключить таймер
+					// для этой адской машины скуки
 					// getContext().setReceiveTimeout(Duration.Undefined());
 				})
-				.matchAny(o -> log.info("received unknown message"))
+				.matchAny(o -> log.info("received unknown message")) // Обработчик неизвестного сообщения
 				.build();
 	}
 }