diff --git a/microservices-messaging/README.md b/microservices-messaging/README.md new file mode 100644 index 000000000000..29497c7229c5 --- /dev/null +++ b/microservices-messaging/README.md @@ -0,0 +1,219 @@ +--- +title: "Microservices Messaging Pattern in Java: Enabling Asynchronous Communication Between Services" +shortTitle: Microservices Messaging +description: "Learn about the Microservices Messaging pattern, a method for enabling asynchronous communication between services through message brokers to enhance decoupling, scalability, and fault tolerance in distributed systems." +category: Integration +language: en +tag: + - API design + - Asynchronous + - Cloud distributed + - Decoupling + - Enterprise patterns + - Event-driven + - Messaging + - Microservices + - Scalability +--- +## Also known as + +* Asynchronous Messaging +* Event-Driven Communication +* Message-Oriented Middleware (MOM) + +## Intent of Microservices Messaging Design Pattern + +The Microservices Messaging pattern enables asynchronous communication between microservices through message passing, allowing for better decoupling, scalability, and fault tolerance. Services communicate by exchanging messages over messaging channels managed by a message broker. + +## Detailed Explanation of Microservices Messaging Pattern with Real-World Examples + +Real-world example + +> Imagine an e-commerce platform where a customer places an order. The Order Service publishes an "Order Created" message to a message broker. Multiple services listen to this message: the Inventory Service updates stock levels, the Payment Service processes payment, and the Notification Service sends confirmation emails. Each service operates independently, processing messages at its own pace without blocking others. If the Payment Service is temporarily down, the message broker holds the message until it recovers, ensuring no data is lost. + +In plain words + +> The Microservices Messaging pattern allows services to communicate asynchronously through a message broker, enabling them to work independently without waiting for each other. + +Wikipedia says + +> Message-oriented middleware is software or hardware infrastructure supporting sending and receiving messages between distributed systems. MOM allows application modules to be distributed over heterogeneous platforms and reduces the complexity of developing applications that span multiple operating systems and network protocols. + +Flowchart + +![Microservices Messaging flowchart](./etc/microservices-messaging-flowchart.png) + + + +## Programmatic Example of Microservices Messaging Pattern in Java + + +The Microservices Messaging pattern demonstrates how services communicate through a message broker without direct coupling. In this example, we show an order processing system where services exchange messages asynchronously. + +The `Message` class represents the data exchanged between services. + +```java +public class Message { + private final String id; + private final String content; + private final LocalDateTime timestamp; + + public Message(String content) { + this.id = UUID.randomUUID().toString(); + this.content = content; + this.timestamp = LocalDateTime.now(); + } + + // Getters +} +``` + +The `MessageBroker` acts as the intermediary that routes messages between producers and consumers. + +```java +public class MessageBroker { + private final Map subscribers = new ConcurrentHashMap<>(); + + public void subscribe(String topic, Consumer handler) { + subscribers.computeIfAbsent(topic, k -> new ArrayList<>()).add(handler); + } + + public void publish(String topic, Message message) { + List handlers = subscribers.get(topic); + if (handlers != null) { + handlers.forEach(handler -> handler.accept(message)); + } + } +} +``` + +The `OrderService` is a message producer that publishes order messages. + +```java +public class OrderService { + private static final Logger LOGGER = LoggerFactory.getLogger(OrderService.class); + private final MessageBroker broker; + + public OrderService(MessageBroker broker) { + this.broker = broker; + } + + public void createOrder(String orderId) { + Message message = new Message("Order Created: " + orderId); + broker.publish("order-topic", message); + LOGGER.info("Published order message: {}", orderId); + } +} +``` + +The `InventoryService` is a message consumer that processes inventory updates. + +```java +public class InventoryService { + private static final Logger LOGGER = LoggerFactory.getLogger(InventoryService.class); + + public void handleMessage(Message message) { + LOGGER.info("Inventory Service received: {}", message.getContent()); + LOGGER.info("Updating inventory..."); + } +} +``` + +The `PaymentService` handles payment processing messages. + +```java +public class PaymentService { + private static final Logger LOGGER = LoggerFactory.getLogger(PaymentService.class); + + public void handleMessage(Message message) { + LOGGER.info("Payment Service received: {}", message.getContent()); + LOGGER.info("Processing payment..."); + } +} +``` + +The `main` application demonstrates the messaging pattern in action. + +```java +public class App { + private static final Logger LOGGER = LoggerFactory.getLogger(App.class); + + public static void main(String[] args) throws InterruptedException { + final MessageBroker broker = new MessageBroker(); + + final InventoryService inventoryService = new InventoryService(); + final PaymentService paymentService = new PaymentService(); + + broker.subscribe("order-topic", inventoryService::handleMessage); + broker.subscribe("order-topic", paymentService::handleMessage); + + final OrderService orderService = new OrderService(broker); + + orderService.createOrder("ORDER-123"); + + Thread.sleep(1000); + } +} +``` + +Console output: + +``` +Published order message: ORDER-123 +Inventory Service received: Order Created: ORDER-123 +Updating inventory... +Payment Service received: Order Created: ORDER-123 +Processing payment... +``` + +Sequence Diagram + +![Microservices Messaging sequence_diagram](./etc/microservices-messaging-sequence-diagram.png) + + + +## When to Use the Microservices Messaging Pattern in Java + +* When services need to communicate without blocking each other. +* In systems requiring loose coupling between components. +* For event-driven architectures where multiple services react to events. +* When you need to handle traffic spikes by buffering messages. +* In distributed systems where services may be temporarily unavailable. + +## Real-World Applications of Microservices Messaging Pattern in Java + +* Java applications using Apache Kafka, RabbitMQ, or ActiveMQ for service communication. +* E-commerce platforms for order processing and inventory management. +* Financial services for transaction processing and notifications. +* IoT systems for sensor data processing and event handling. + +## Benefits and Trade-offs of Microservices Messaging Pattern + +* Services are loosely coupled and can be developed and deployed independently. +* Message buffering improves system resilience when services are temporarily unavailable. +* Supports multiple communication patterns like publish/subscribe and request/reply. +* Enhances scalability by allowing parallel message processing. +* Natural support for event-driven architectures. + +Trade-offs: + +* Introduces additional complexity with the message broker infrastructure. +* Requires high availability setup for the message broker. +* Eventual consistency instead of immediate consistency. +* Debugging asynchronous flows is more complex than synchronous calls. +* Need to handle message duplication and ensure idempotent consumers. + +## Related Java Design Patterns + +* [Saga Pattern](https://java-design-patterns.com/patterns/saga/): Uses messaging to coordinate distributed transactions. +* [CQRS Pattern](https://java-design-patterns.com/patterns/cqrs/): Often uses messaging to separate read and write operations. +* [Event Sourcing](https://java-design-patterns.com/patterns/event-sourcing/): Stores state changes as messages. +* [API Gateway](https://java-design-patterns.com/patterns/microservices-api-gateway/): Complements messaging for synchronous requests. + +## References and Credits + +* [Enterprise Integration Patterns: Designing, Building, and Deploying Messaging Solutions](https://amzn.to/3vLKqET) +* [Microservices Patterns: With examples in Java](https://amzn.to/3UyWD5O) +* [Building Event-Driven Microservices: Leveraging Organizational Data at Scale](https://amzn.to/3PihS9R) +* [Pattern: Messaging (microservices.io)](https://microservices.io/patterns/communication-style/messaging.html) +* [Apache Kafka Documentation](https://kafka.apache.org/documentation/) \ No newline at end of file diff --git a/microservices-messaging/etc/microservices-messaging-flowchart.png b/microservices-messaging/etc/microservices-messaging-flowchart.png new file mode 100644 index 000000000000..dd26a2e2ce4a Binary files /dev/null and b/microservices-messaging/etc/microservices-messaging-flowchart.png differ diff --git a/microservices-messaging/etc/microservices-messaging-sequence-diagram.png b/microservices-messaging/etc/microservices-messaging-sequence-diagram.png new file mode 100644 index 000000000000..d22be5acbb5d Binary files /dev/null and b/microservices-messaging/etc/microservices-messaging-sequence-diagram.png differ diff --git a/microservices-messaging/pom.xml b/microservices-messaging/pom.xml new file mode 100644 index 000000000000..ce4267165e62 --- /dev/null +++ b/microservices-messaging/pom.xml @@ -0,0 +1,116 @@ + + + + 4.0.0 + + + com.iluwatar + java-design-patterns + 1.26.0-SNAPSHOT + + + microservices-messaging + 1.26.0-SNAPSHOT + + + 3.6.1 + + + + + + org.apache.kafka + kafka-clients + ${kafka.version} + + + + + org.slf4j + slf4j-api + + + ch.qos.logback + logback-classic + + + + + com.fasterxml.jackson.core + jackson-databind + 2.16.1 + + + + + org.junit.jupiter + junit-jupiter-engine + test + + + org.mockito + mockito-core + test + + + com.fasterxml.jackson.datatype + jackson-datatype-jsr310 + 2.19.2 + compile + + + org.junit.jupiter + junit-jupiter-api + 5.12.2 + test + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + + + + + + com.iluwatar.messaging.App + + + + + + + + + \ No newline at end of file diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/App.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/App.java new file mode 100644 index 000000000000..27387d8b4676 --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/App.java @@ -0,0 +1,137 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +/** + * The Microservices Messaging pattern enables asynchronous communication between services through + * Apache Kafka. This example demonstrates how services can communicate without tight coupling. + * + *

In this example: + *

+ * + *

Key benefits demonstrated: + *

+ * + *

Prerequisites: This example requires a running Kafka instance. + * Start Kafka locally: + * 

+ * # Start Zookeeper
+ * bin/zookeeper-server-start.sh config/zookeeper.properties
+ *
+ * # Start Kafka
+ * bin/kafka-server-start.sh config/server.properties
+ *
+ * # Create topic
+ * bin/kafka-topics.sh --create --topic order-topic --bootstrap-server localhost:9092
+ *
+ */ +public class App { + private static final Logger LOGGER = LoggerFactory.getLogger(App.class); + private static final String BOOTSTRAP_SERVERS = "localhost:9092"; + + /** + * Program entry point. + * + * @param args command line arguments + */ + public static void main(String[] args) throws InterruptedException { + LOGGER.info("Starting Microservices Messaging Pattern with Apache Kafka"); + + // Create Kafka producer + KafkaMessageProducer producer = new KafkaMessageProducer(BOOTSTRAP_SERVERS); + + // Create consumer services + InventoryService inventoryService = new InventoryService(); + PaymentService paymentService = new PaymentService(); + NotificationService notificationService = new NotificationService(); + + // Create Kafka consumers + KafkaMessageConsumer inventoryConsumer = new KafkaMessageConsumer( + BOOTSTRAP_SERVERS, "inventory-group", "order-topic", inventoryService::handleMessage); + + KafkaMessageConsumer paymentConsumer = new KafkaMessageConsumer( + BOOTSTRAP_SERVERS, "payment-group", "order-topic", paymentService::handleMessage); + + KafkaMessageConsumer notificationConsumer = new KafkaMessageConsumer( + BOOTSTRAP_SERVERS, "notification-group", "order-topic", + notificationService::handleMessage); + + // Start consumers in separate threads + ExecutorService executor = Executors.newFixedThreadPool(3); + executor.submit(inventoryConsumer); + executor.submit(paymentConsumer); + executor.submit(notificationConsumer); + + // Give consumers time to subscribe + Thread.sleep(2000); + + // Create producer service + OrderService orderService = new OrderService(producer); + + // Demonstrate the messaging pattern + LOGGER.info("\n=== Creating Order ==="); + orderService.createOrder("ORDER-001"); + + Thread.sleep(2000); + + LOGGER.info("\n=== Updating Order ==="); + orderService.updateOrder("ORDER-001"); + + Thread.sleep(2000); + + LOGGER.info("\n=== Cancelling Order ==="); + orderService.cancelOrder("ORDER-001"); + + Thread.sleep(2000); + + // Cleanup + LOGGER.info("\nShutting down..."); + inventoryConsumer.stop(); + paymentConsumer.stop(); + notificationConsumer.stop(); + producer.close(); + + executor.shutdown(); + executor.awaitTermination(5, TimeUnit.SECONDS); + + LOGGER.info("Microservices Messaging Pattern demonstration completed"); + } +} \ No newline at end of file diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/InventoryService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/InventoryService.java new file mode 100644 index 000000000000..5419a1229f04 --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/InventoryService.java @@ -0,0 +1,85 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * InventoryService is a message consumer that processes inventory-related messages from Kafka. + * It listens to order events and updates inventory accordingly. + * + *

This service runs in its own Kafka consumer group (inventory-group) which allows it to: + *

+ */ +public class InventoryService { + private static final Logger LOGGER = LoggerFactory.getLogger(InventoryService.class); + + /** + * Handles incoming messages related to orders from Kafka. + * + * @param message the message to process + */ + public void handleMessage(Message message) { + LOGGER.info("Inventory Service received message [{}]: {}", + message.getId(), message.getContent()); + + if (message.getContent().contains("Order Created")) { + updateInventory(message); + } else if (message.getContent().contains("Order Cancelled")) { + restoreInventory(message); + } else { + LOGGER.debug("No inventory action needed for: {}", message.getContent()); + } + } + + private void updateInventory(Message message) { + LOGGER.info("Updating inventory for message: {}", message.getId()); + // Simulate inventory update - reserve stock for the order + try { + Thread.sleep(100); + LOGGER.info("Inventory updated successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Inventory update interrupted", e); + } + } + + private void restoreInventory(Message message) { + LOGGER.info("Restoring inventory for message: {}", message.getId()); + // Simulate inventory restoration - release reserved stock + try { + Thread.sleep(100); + LOGGER.info("Inventory restored successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Inventory restore interrupted", e); + } + } +} \ No newline at end of file diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageConsumer.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageConsumer.java new file mode 100644 index 000000000000..779a3ee4eace --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageConsumer.java @@ -0,0 +1,114 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.time.Duration; +import java.util.Collections; +import java.util.Properties; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; + +/** + * Kafka message consumer that subscribes to topics and processes messages. + */ +public class KafkaMessageConsumer implements AutoCloseable, Runnable { + private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageConsumer.class); + private final KafkaConsumer consumer; + private final ObjectMapper objectMapper; + private final String topic; + private final Consumer messageHandler; + private final AtomicBoolean running = new AtomicBoolean(true); + + /** + * Creates a new Kafka message consumer. + * + * @param bootstrapServers Kafka bootstrap servers + * @param groupId consumer group ID + * @param topic topic to subscribe to + * @param messageHandler handler for received messages + */ + public KafkaMessageConsumer(String bootstrapServers, String groupId, String topic, + Consumer messageHandler) { + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); + + this.consumer = new KafkaConsumer<>(props); + this.objectMapper = new ObjectMapper(); + this.objectMapper.registerModule(new JavaTimeModule()); + this.topic = topic; + this.messageHandler = messageHandler; + } + + @Override + public void run() { + try { + consumer.subscribe(Collections.singletonList(topic)); + LOGGER.info("Consumer subscribed to topic: {}", topic); + + while (running.get()) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + records.forEach(record -> { + try { + Message message = objectMapper.readValue(record.value(), Message.class); + LOGGER.info("Received message from topic '{}': {}", topic, message.getId()); + messageHandler.accept(message); + } catch (Exception e) { + LOGGER.error("Error processing message: {}", e.getMessage(), e); + } + }); + } + } catch (Exception e) { + LOGGER.error("Consumer error: {}", e.getMessage(), e); + } finally { + consumer.close(); + LOGGER.info("Consumer closed for topic: {}", topic); + } + } + + /** + * Stops the consumer. + */ + public void stop() { + running.set(false); + } + + @Override + public void close() { + stop(); + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageProducer.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageProducer.java new file mode 100644 index 000000000000..94e75ef0d098 --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageProducer.java @@ -0,0 +1,96 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.Properties; + +/** + * Kafka message producer that publishes messages to Kafka topics. + */ +public class KafkaMessageProducer implements AutoCloseable { + private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageProducer.class); + private final KafkaProducer producer; + private final ObjectMapper objectMapper; + + /** + * Creates a new Kafka message producer. + * + * @param bootstrapServers Kafka bootstrap servers + */ + public KafkaMessageProducer(String bootstrapServers) { + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.ACKS_CONFIG, "all"); + props.put(ProducerConfig.RETRIES_CONFIG, 3); + + this.producer = new KafkaProducer<>(props); + this.objectMapper = new ObjectMapper(); + this.objectMapper.registerModule(new JavaTimeModule()); + } + + /** + * Publishes a message to a Kafka topic. + * + * @param topic the topic to publish to + * @param message the message to publish + */ + public void publish(String topic, Message message) { + try { + String json = objectMapper.writeValueAsString(message); + ProducerRecord record = new ProducerRecord<>(topic, message.getId(), json); + + producer.send(record, (metadata, exception) -> { + if (exception != null) { + LOGGER.error("Failed to publish message to topic {}: {}", topic, exception.getMessage()); + } else { + LOGGER.info("Published message to topic '{}' [partition={}, offset={}]", + topic, metadata.partition(), metadata.offset()); + } + }); + + } catch (Exception e) { + LOGGER.error("Error serializing message: {}", e.getMessage(), e); + } + } + + @Override + public void close() { + if (producer != null) { + producer.flush(); + producer.close(); + LOGGER.info("Kafka producer closed"); + } + } +} diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/Message.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/Message.java new file mode 100644 index 000000000000..ce1c37f9bbbd --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/Message.java @@ -0,0 +1,74 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.Getter; +import java.time.LocalDateTime; +import java.util.UUID; + +/** + * Represents a message exchanged between services. + */ +@Getter +public class Message { + private final String id; + private final String content; + private final LocalDateTime timestamp; + + /** + * Creates a new message with the given content. + * + * @param content the message content + */ + public Message(String content) { + this.id = UUID.randomUUID().toString(); + this.content = content; + this.timestamp = LocalDateTime.now(); + } + + /** + * JSON constructor for deserialization. + */ + @JsonCreator + public Message( + @JsonProperty("id") String id, + @JsonProperty("content") String content, + @JsonProperty("timestamp") LocalDateTime timestamp) { + this.id = id; + this.content = content; + this.timestamp = timestamp; + } + + @Override + public String toString() { + return "Message{" + + "id='" + id + '\'' + + ", content='" + content + '\'' + + ", timestamp=" + timestamp + + '}'; + } +} \ No newline at end of file diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/NotificationService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/NotificationService.java new file mode 100644 index 000000000000..de2d1aca59fb --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/NotificationService.java @@ -0,0 +1,100 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * NotificationService is a message consumer that processes notification-related messages from Kafka. + * It listens to order events and sends notifications to customers. + * + *

This service runs in its own Kafka consumer group (notification-group) which allows it to: + *

    + *
  • Process messages independently from other services
    • + *
  • Scale horizontally by adding more instances to the consumer group
    • + *
  • Resume from last committed offset if the service restarts
    • + *
+ */ +public class NotificationService { + private static final Logger LOGGER = LoggerFactory.getLogger(NotificationService.class); + + /** + * Handles incoming messages related to orders from Kafka. + * + * @param message the message to process + */ + public void handleMessage(Message message) { + LOGGER.info("Notification Service received message [{}]: {}", + message.getId(), message.getContent()); + + if (message.getContent().contains("Order Created")) { + sendOrderConfirmation(message); + } else if (message.getContent().contains("Order Updated")) { + sendOrderUpdate(message); + } else if (message.getContent().contains("Order Cancelled")) { + sendCancellationNotice(message); + } else { + LOGGER.debug("No notification action needed for: {}", message.getContent()); + } + } + + private void sendOrderConfirmation(Message message) { + LOGGER.info("Sending order confirmation for message: {}", message.getId()); + // Simulate sending email/SMS notification to customer + try { + Thread.sleep(50); + LOGGER.info("Order confirmation sent successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Notification send interrupted", e); + } + } + + private void sendOrderUpdate(Message message) { + LOGGER.info("Sending order update notification for message: {}", message.getId()); + // Simulate sending update notification + try { + Thread.sleep(50); + LOGGER.info("Order update notification sent successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Notification send interrupted", e); + } + } + + private void sendCancellationNotice(Message message) { + LOGGER.info("Sending cancellation notice for message: {}", message.getId()); + // Simulate sending cancellation notification + try { + Thread.sleep(50); + LOGGER.info("Cancellation notice sent successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Notification send interrupted", e); + } + } +} \ No newline at end of file diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/OrderService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/OrderService.java new file mode 100644 index 000000000000..5e8d58e0765c --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/OrderService.java @@ -0,0 +1,78 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * OrderService is a message producer that publishes order-related messages using Kafka. + */ +public class OrderService { + private static final Logger LOGGER = LoggerFactory.getLogger(OrderService.class); + private static final String ORDER_TOPIC = "order-topic"; + + private final KafkaMessageProducer producer; + + public OrderService(KafkaMessageProducer producer) { + this.producer = producer; + } + + /** + * Creates an order and publishes a message to notify other services. + * + * @param orderId the ID of the order to create + */ + public void createOrder(String orderId) { + LOGGER.info("Creating order: {}", orderId); + Message message = new Message("Order Created: " + orderId); + producer.publish(ORDER_TOPIC, message); + LOGGER.info("Order creation message published for: {}", orderId); + } + + /** + * Updates an order and publishes a message to notify other services. + * + * @param orderId the ID of the order to update + */ + public void updateOrder(String orderId) { + LOGGER.info("Updating order: {}", orderId); + Message message = new Message("Order Updated: " + orderId); + producer.publish(ORDER_TOPIC, message); + LOGGER.info("Order update message published for: {}", orderId); + } + + /** + * Cancels an order and publishes a message to notify other services. + * + * @param orderId the ID of the order to cancel + */ + public void cancelOrder(String orderId) { + LOGGER.info("Cancelling order: {}", orderId); + Message message = new Message("Order Cancelled: " + orderId); + producer.publish(ORDER_TOPIC, message); + LOGGER.info("Order cancellation message published for: {}", orderId); + } +} \ No newline at end of file diff --git a/microservices-messaging/src/main/java/com/iluwatar/messaging/PaymentService.java b/microservices-messaging/src/main/java/com/iluwatar/messaging/PaymentService.java new file mode 100644 index 000000000000..34d752d48339 --- /dev/null +++ b/microservices-messaging/src/main/java/com/iluwatar/messaging/PaymentService.java @@ -0,0 +1,86 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * PaymentService is a message consumer that processes payment-related messages from Kafka. + * It listens to order events and handles payment processing. + * + *

This service runs in its own Kafka consumer group (payment-group) which allows it to: + *

    + *
  • Process messages independently from other services
    • + *
  • Scale horizontally by adding more instances to the consumer group
    • + *
  • Resume from last committed offset if the service restarts
    • + *
+ */ +public class PaymentService { + private static final Logger LOGGER = LoggerFactory.getLogger(PaymentService.class); + + /** + * Handles incoming messages related to orders from Kafka. + * + * @param message the message to process + */ + public void handleMessage(Message message) { + LOGGER.info("Payment Service received message [{}]: {}", + message.getId(), message.getContent()); + + if (message.getContent().contains("Order Created")) { + processPayment(message); + } else if (message.getContent().contains("Order Cancelled")) { + refundPayment(message); + } else { + LOGGER.debug("No payment action needed for: {}", message.getContent()); + } + } + + private void processPayment(Message message) { + LOGGER.info("Processing payment for message: {}", message.getId()); + // Simulate payment processing - charge the customer + try { + Thread.sleep(150); + LOGGER.info("Payment processed successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Payment processing interrupted", e); + } + } + + private void refundPayment(Message message) { + LOGGER.info("Refunding payment for message: {}", message.getId()); + // Simulate payment refund - return money to customer + try { + Thread.sleep(150); + LOGGER.info("Payment refunded successfully for: {}", message.getContent()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Payment refund interrupted", e); + } + } +} \ No newline at end of file diff --git a/microservices-messaging/src/main/resources/logback.xml b/microservices-messaging/src/main/resources/logback.xml new file mode 100644 index 000000000000..7f40741c0ff1 --- /dev/null +++ b/microservices-messaging/src/main/resources/logback.xml @@ -0,0 +1,40 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + + \ No newline at end of file diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/AppTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/AppTest.java new file mode 100644 index 000000000000..96af311c79ca --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/AppTest.java @@ -0,0 +1,49 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; + +/** + * Unit tests for {@link App}. + * Tests main application entry point. + */ +class AppTest { + + @Test + void testMainMethodDoesNotThrowException() { + // Note: This test requires a running Kafka instance + // In a real scenario, we would use embedded Kafka for testing + // For now, we just verify the method can be called without compilation errors + + // Act & Assert + assertDoesNotThrow(() -> { + // Main method requires Kafka to be running, so we don't actually call it in unit tests + // This is a placeholder to ensure the class structure is correct + }, "App should be instantiable"); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/InventoryServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/InventoryServiceTest.java new file mode 100644 index 000000000000..f1728f853470 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/InventoryServiceTest.java @@ -0,0 +1,101 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Unit tests for {@link InventoryService}. + * Tests service behavior with various message types without Kafka dependencies. + */ +class InventoryServiceTest { + + private InventoryService inventoryService; + + @BeforeEach + void setUp() { + inventoryService = new InventoryService(); + } + + @Test + void testServiceCanBeInstantiated() { + // Arrange & Act & Assert + assertNotNull(inventoryService, "InventoryService should be instantiated"); + } + + @Test + void testHandleOrderCreatedMessage() { + // Arrange + var message = new Message("Order Created: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> inventoryService.handleMessage(message), + "Should handle order created message without error"); + } + + @Test + void testHandleOrderCancelledMessage() { + // Arrange + var message = new Message("Order Cancelled: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> inventoryService.handleMessage(message), + "Should handle order cancelled message without error"); + } + + @Test + void testHandleOrderUpdatedMessage() { + // Arrange + var message = new Message("Order Updated: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> inventoryService.handleMessage(message), + "Should handle order updated message without error"); + } + + @Test + void testHandleUnknownMessage() { + // Arrange + var message = new Message("Unknown Event: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> inventoryService.handleMessage(message), + "Should handle unknown message without error"); + } + + @Test + void testHandleMultipleMessages() { + // Act & Assert + assertDoesNotThrow(() -> { + inventoryService.handleMessage(new Message("Order Created: ORDER-001")); + inventoryService.handleMessage(new Message("Order Updated: ORDER-001")); + inventoryService.handleMessage(new Message("Order Cancelled: ORDER-001")); + }, "Should handle multiple messages without error"); + } +} \ No newline at end of file diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java new file mode 100644 index 000000000000..a5042eb7f776 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java @@ -0,0 +1,81 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Unit tests for {@link KafkaMessageConsumer}. + * Note: These tests verify basic functionality without requiring a Kafka instance. + * For integration tests with Kafka, use embedded Kafka or testcontainers. + */ +class KafkaMessageConsumerTest { + + @Test + void testConsumerCanBeInstantiated() { + // Arrange & Act & Assert + // Note: Don't actually create consumer in unit test as it requires Kafka + assertNotNull(KafkaMessageConsumer.class, + "KafkaMessageConsumer class should exist"); + } + + @Test + void testConsumerImplementsRunnable() { + // Arrange & Act & Assert + var interfaces = KafkaMessageConsumer.class.getInterfaces(); + for (var i : interfaces) { + if (i.equals(Runnable.class)) { + break; + } + } + assertNotNull(interfaces, "Should have interfaces"); + // Note: Runnable is implemented for threading + } + + @Test + void testConsumerImplementsAutoCloseable() { + // Arrange & Act & Assert + var interfaces = KafkaMessageConsumer.class.getInterfaces(); + for (var i : interfaces) { + if (i.equals(AutoCloseable.class)) { + break; + } + } + assertNotNull(interfaces, "Should have interfaces"); + // Note: AutoCloseable is implemented + } + + @Test + void testConsumerClassHasStopMethod() { + // Arrange & Act & Assert + assertDoesNotThrow(() -> { + var method = KafkaMessageConsumer.class.getDeclaredMethod("stop"); + assertNotNull(method, "stop method should exist"); + }, "KafkaMessageConsumer should have stop method"); + } +} \ No newline at end of file diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java new file mode 100644 index 000000000000..ee169595c1a8 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java @@ -0,0 +1,70 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Unit tests for {@link KafkaMessageProducer}. + * Note: These tests verify basic functionality without requiring a Kafka instance. + * For integration tests with Kafka, use embedded Kafka or testcontainers. + */ +class KafkaMessageProducerTest { + + @Test + void testProducerCanBeInstantiated() { + // Arrange & Act & Assert + // Note: Don't actually create producer in unit test as it requires Kafka + // This test just verifies the class structure is correct + assertNotNull(KafkaMessageProducer.class, + "KafkaMessageProducer class should exist"); + } + + @Test + void testProducerClassHasPublishMethod() { + // Arrange & Act & Assert + assertDoesNotThrow(() -> { + var method = KafkaMessageProducer.class.getDeclaredMethod( + "publish", String.class, Message.class); + assertNotNull(method, "publish method should exist"); + }, "KafkaMessageProducer should have publish method"); + } + + @Test + void testProducerImplementsAutoCloseable() { + // Arrange & Act & Assert + var interfaces = KafkaMessageProducer.class.getInterfaces(); + for (var i : interfaces) { + if (i.equals(AutoCloseable.class)) { + break; + } + } + assertNotNull(interfaces, "Should have interfaces"); + // Note: AutoCloseable is implemented + } +} \ No newline at end of file diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/MessageTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/MessageTest.java new file mode 100644 index 000000000000..b640ef147adf --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/MessageTest.java @@ -0,0 +1,166 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.LocalDateTime; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Unit tests for {@link Message}. + * Tests follow FIRST principles: Fast, Isolated, Repeatable, Self-validating, Timely. + */ +class MessageTest { + + private ObjectMapper objectMapper; + + @BeforeEach + void setUp() { + objectMapper = new ObjectMapper(); + objectMapper.registerModule(new JavaTimeModule()); + } + + @Test + void testMessageCreation() { + // Arrange & Act + var message = new Message("Test content"); + + // Assert + assertNotNull(message.getId(), "Message ID should not be null"); + assertEquals("Test content", message.getContent(), "Content should match"); + assertNotNull(message.getTimestamp(), "Timestamp should not be null"); + } + + @Test + void testMessageIdIsUnique() { + // Arrange & Act + var message1 = new Message("Content 1"); + var message2 = new Message("Content 2"); + + // Assert + assertNotEquals(message1.getId(), message2.getId(), "Each message should have unique ID"); + } + + @Test + void testMessageTimestamp() { + // Arrange + var beforeCreation = LocalDateTime.now(); + + // Act + var message = new Message("Test"); + var afterCreation = LocalDateTime.now(); + + // Assert + assertTrue(message.getTimestamp().isAfter(beforeCreation.minusSeconds(1)) + && message.getTimestamp().isBefore(afterCreation.plusSeconds(1)), + "Timestamp should be close to creation time"); + } + + @Test + void testJsonSerialization() throws Exception { + // Arrange + var originalMessage = new Message("Test content"); + + // Act + var json = objectMapper.writeValueAsString(originalMessage); + + // Assert + assertNotNull(json, "JSON should not be null"); + assertTrue(json.contains("Test content"), "JSON should contain content"); + assertTrue(json.contains(originalMessage.getId()), "JSON should contain ID"); + } + + @Test + void testJsonDeserialization() throws Exception { + // Arrange + var originalMessage = new Message("Test content"); + var json = objectMapper.writeValueAsString(originalMessage); + + // Act + var deserializedMessage = objectMapper.readValue(json, Message.class); + + // Assert + assertNotNull(deserializedMessage, "Deserialized message should not be null"); + assertEquals(originalMessage.getId(), deserializedMessage.getId(), "IDs should match"); + assertEquals(originalMessage.getContent(), deserializedMessage.getContent(), + "Content should match"); + } + + @Test + void testToString() { + // Arrange + var message = new Message("Test content"); + + // Act + var result = message.toString(); + + // Assert + assertNotNull(result, "ToString should not return null"); + assertTrue(result.contains("Message{"), "ToString should contain class name"); + assertTrue(result.contains("Test content"), "ToString should contain content"); + assertTrue(result.contains(message.getId()), "ToString should contain ID"); + } + + @Test + void testMessageWithEmptyContent() { + // Arrange & Act + var message = new Message(""); + + // Assert + assertNotNull(message.getId(), "ID should be generated even for empty content"); + assertEquals("", message.getContent(), "Empty content should be preserved"); + } + + @Test + void testMessageWithNullContent() { + // Arrange & Act + var message = new Message(null); + + // Assert + assertNotNull(message.getId(), "ID should be generated even for null content"); + assertEquals(null, message.getContent(), "Null content should be preserved"); + } + + @Test + void testMessageWithSpecialCharacters() { + // Arrange + var specialContent = "Test with special chars: @#$%^&*()"; + + // Act + var message = new Message(specialContent); + + // Assert + assertEquals(specialContent, message.getContent(), + "Special characters should be preserved"); + } +} diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/NotificationServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/NotificationServiceTest.java new file mode 100644 index 000000000000..1e6bc145876b --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/NotificationServiceTest.java @@ -0,0 +1,111 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Unit tests for {@link NotificationService}. + * Tests service behavior with various message types without Kafka dependencies. + */ +class NotificationServiceTest { + + private NotificationService notificationService; + + @BeforeEach + void setUp() { + notificationService = new NotificationService(); + } + + @Test + void testServiceCanBeInstantiated() { + // Arrange & Act & Assert + assertNotNull(notificationService, "NotificationService should be instantiated"); + } + + @Test + void testHandleOrderCreatedMessage() { + // Arrange + var message = new Message("Order Created: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> notificationService.handleMessage(message), + "Should handle order created message without error"); + } + + @Test + void testHandleOrderUpdatedMessage() { + // Arrange + var message = new Message("Order Updated: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> notificationService.handleMessage(message), + "Should handle order updated message without error"); + } + + @Test + void testHandleOrderCancelledMessage() { + // Arrange + var message = new Message("Order Cancelled: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> notificationService.handleMessage(message), + "Should handle order cancelled message without error"); + } + + @Test + void testHandleUnknownMessage() { + // Arrange + var message = new Message("Unknown Event: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> notificationService.handleMessage(message), + "Should handle unknown message without error"); + } + + @Test + void testHandleAllMessageTypes() { + // Act & Assert + assertDoesNotThrow(() -> { + notificationService.handleMessage(new Message("Order Created: ORDER-001")); + notificationService.handleMessage(new Message("Order Updated: ORDER-001")); + notificationService.handleMessage(new Message("Order Cancelled: ORDER-001")); + }, "Should handle all message types without error"); + } + + @Test + void testHandleMultipleOrdersSequentially() { + // Act & Assert + assertDoesNotThrow(() -> { + notificationService.handleMessage(new Message("Order Created: ORDER-001")); + notificationService.handleMessage(new Message("Order Created: ORDER-002")); + notificationService.handleMessage(new Message("Order Created: ORDER-003")); + }, "Should handle multiple orders sequentially without error"); + } +} \ No newline at end of file diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/OrderServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/OrderServiceTest.java new file mode 100644 index 000000000000..2e04e23fd0de --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/OrderServiceTest.java @@ -0,0 +1,113 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * Unit tests for {@link OrderService}. + * Tests follow Arrange-Act-Assert pattern. + */ +class OrderServiceTest { + + private KafkaMessageProducer mockProducer; + private OrderService orderService; + + @BeforeEach + void setUp() { + // Arrange - Create mock producer to avoid Kafka dependency + mockProducer = mock(KafkaMessageProducer.class); + orderService = new OrderService(mockProducer); + } + + @Test + void testCreateOrder() { + // Arrange + var orderId = "ORDER-001"; + + // Act + assertDoesNotThrow(() -> orderService.createOrder(orderId)); + + // Assert + verify(mockProducer, times(1)).publish(eq("order-topic"), any(Message.class)); + } + + @Test + void testUpdateOrder() { + // Arrange + var orderId = "ORDER-002"; + + // Act + assertDoesNotThrow(() -> orderService.updateOrder(orderId)); + + // Assert + verify(mockProducer, times(1)).publish(eq("order-topic"), any(Message.class)); + } + + @Test + void testCancelOrder() { + // Arrange + var orderId = "ORDER-003"; + + // Act + assertDoesNotThrow(() -> orderService.cancelOrder(orderId)); + + // Assert + verify(mockProducer, times(1)).publish(eq("order-topic"), any(Message.class)); + } + + @Test + void testMultipleOrderOperations() { + // Arrange + var orderId = "ORDER-004"; + + // Act + orderService.createOrder(orderId); + orderService.updateOrder(orderId); + orderService.cancelOrder(orderId); + + // Assert + verify(mockProducer, times(3)).publish(eq("order-topic"), any(Message.class)); + } + + @Test + void testCreateOrderWithDifferentIds() { + // Act + orderService.createOrder("ORDER-001"); + orderService.createOrder("ORDER-002"); + orderService.createOrder("ORDER-003"); + + // Assert + verify(mockProducer, times(3)).publish(eq("order-topic"), any(Message.class)); + } +} \ No newline at end of file diff --git a/microservices-messaging/src/test/java/com/iluwatar/messaging/PaymentServiceTest.java b/microservices-messaging/src/test/java/com/iluwatar/messaging/PaymentServiceTest.java new file mode 100644 index 000000000000..77678304c079 --- /dev/null +++ b/microservices-messaging/src/test/java/com/iluwatar/messaging/PaymentServiceTest.java @@ -0,0 +1,100 @@ +/* + * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt). + * + * The MIT License + * Copyright © 2014-2022 Ilkka Seppälä + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.iluwatar.messaging; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Unit tests for {@link PaymentService}. + * Tests service behavior with various message types without Kafka dependencies. + */ +class PaymentServiceTest { + + private PaymentService paymentService; + + @BeforeEach + void setUp() { + paymentService = new PaymentService(); + } + + @Test + void testServiceCanBeInstantiated() { + // Arrange & Act & Assert + assertNotNull(paymentService, "PaymentService should be instantiated"); + } + + @Test + void testHandleOrderCreatedMessage() { + // Arrange + var message = new Message("Order Created: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> paymentService.handleMessage(message), + "Should handle order created message without error"); + } + + @Test + void testHandleOrderCancelledMessage() { + // Arrange + var message = new Message("Order Cancelled: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> paymentService.handleMessage(message), + "Should handle order cancelled message without error"); + } + + @Test + void testHandleOrderUpdatedMessage() { + // Arrange + var message = new Message("Order Updated: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> paymentService.handleMessage(message), + "Should handle order updated message without error"); + } + + @Test + void testHandleUnknownMessage() { + // Arrange + var message = new Message("Unknown Event: ORDER-001"); + + // Act & Assert + assertDoesNotThrow(() -> paymentService.handleMessage(message), + "Should handle unknown message without error"); + } + + @Test + void testHandleMultipleMessages() { + // Act & Assert + assertDoesNotThrow(() -> { + paymentService.handleMessage(new Message("Order Created: ORDER-001")); + paymentService.handleMessage(new Message("Order Cancelled: ORDER-001")); + }, "Should handle multiple messages without error"); + } +} \ No newline at end of file