mirror of
https://github.com/tiennm99/java-design-patterns.git
synced 2026-09-03 04:18:19 +00:00
feat: microservice messaging pattern (#3564)
* Initialize microservices-messaging Spring Boot project Add initial project structure for microservices-messaging using Spring Boot. Includes Maven configuration with dependencies for Kafka, Lombok, and testing, as well as main application, test class, and application properties. * Initialize microservices messaging pattern 1. Added initial project structure for demonstrating the microservices messaging pattern. 2. Introduced service stubs (OrderService, InventoryService, PaymentService, NotificationService), a Message and MessageBroker class, and a main App entry point. 3. Added README and logging configuration. * Implement Message and MessageBroker classes 1. Added the Message class with unique ID, content, and timestamp fields, and a toString method. 2. Implemented the MessageBroker class to support topic-based publish-subscribe messaging, including subscriber management, message publishing, and logging. * Implement messaging pattern for microservices 1. Added message handling logic to InventoryService, PaymentService, and NotificationService. 2. OrderService now publishes order events to a MessageBroker, and App demonstrates the messaging workflow. 3. Each service processes relevant order events and logs actions for demonstration purposes. * Expand microservices messaging docs and add diagrams 1. Enhanced the README with detailed explanations, real-world examples, Java code samples, and references for the Microservices Messaging pattern. 2. Added flowchart and sequence diagram images to illustrate the pattern. * Refactor to use Kafka for microservices messaging 1. Added Apache Kafka for asynchronous communication between services. 2. Added KafkaMessageProducer and KafkaMessageConsumer classes, updated service implementations and main application logic to use Kafka, and adjusted the Maven configuration to include Kafka and Jackson dependencies. 3. Updated and moved all classes to the com.iluwatar.messaging package, improved documentation, and updated diagrams to reflect the new architecture. * Add unit tests and license headers to messaging module 1. Added comprehensive unit tests for App, InventoryService, KafkaMessageConsumer, KafkaMessageProducer, Message, NotificationService, OrderService, and PaymentService. 2. Added MIT license headers to all main source files and logback.xml. 3. Updated pom.xml to include JUnit Jupiter as a test dependency. * Refactor and simplify service and Kafka test classes 1. Simplified unit tests for InventoryService, NotificationService, and PaymentService by removing null content tests and adding instantiation checks. 2. Refactored KafkaMessageConsumerTest and KafkaMessageProducerTest to avoid requiring a real Kafka instance, focusing on class structure and method existence instead of integration behavior. * Add microservices-messaging module and run scripts Introduce the microservices-messaging module and register it in the root pom.xml. Add docker-compose.yml to run a local Kafka (confluentinc/cp-kafka) with a healthcheck, plus run-app.ps1 and run-app.sh helper scripts that start Kafka if needed and then launch the application. Update the module README with usage instructions. Also remove a duplicated junit-jupiter-api test dependency from the module pom to rely on project defaults. * microservices-messaging: code style and Lombok Add Lombok as a provided dependency and apply formatting/refactoring across the microservices-messaging module. Changes include import reordering, Javadoc and logging formatting, consistent lambda/try/catch indentation, small Kafka consumer/producer refinements (Duration/Properties usage and callback formatting), Message class tweaks (@Getter, JSON ctor and toString formatting) and EOF/newline fixes. Unit tests were also reformatted for consistency. These are non-functional style and readability improvements; no behavior changes intended. * Make Kafka producer/consumer testable Refactor KafkaMessageProducer and KafkaMessageConsumer to depend on the Producer/Consumer interfaces and add constructors that accept mockable instances. Extract default producer/consumer creation into factory methods so tests can inject MockProducer/MockConsumer. Update tests across the microservices-messaging module to use MockProducer/MockConsumer, add more meaningful assertions, error/interrupt handling tests, and simplify AppTest. These changes improve unit testability and remove the need for a running Kafka instance while preserving runtime behavior. * Format test Javadoc and reorder imports Normalize Javadoc formatting and reorder static JUnit imports for consistency in messaging tests. Converted multi-line test Javadocs to single-line comments and adjusted import ordering in the following files: - microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java - microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java - microservices-messaging/src/test/java/com/iluwatar/messaging/OrderServiceTest.java No functional changes. * Add MIT headers and exclude .ps1 from license checks Add MIT license headers to microservices-messaging/docker-compose.yml, run-app.ps1 and run-app.sh to ensure license text is present in these scripts. Update root pom.xml to exclude PowerShell (*.ps1) files from the license plugin checks so those files are not processed by the license rule. * Add Kafka docker service to CI workflows Bring up Kafka for tests in CI and PR workflows. Adds steps to run docker compose for microservices-messaging, poll Kafka readiness (up to 20 retries), and always tear down with docker compose down. Enables Maven tests that depend on Kafka. Affects .github/workflows/maven-ci.yml and .github/workflows/maven-pr-builder.yml. * Use kafka-topics instead of kafka-topics.sh Replace calls to kafka-topics.sh with kafka-topics in CI workflows and the Docker Compose healthcheck. Updated .github/workflows/maven-ci.yml, .github/workflows/maven-pr-builder.yml, and microservices-messaging/docker-compose.yml to use the kafka-topics binary for readiness checks and healthchecks. This prevents failures on images that expose the kafka-topics command without the .sh wrapper. * Use docker compose --wait for Kafka startup Replace the custom bash readiness loop with `docker compose ... up -d --wait` in CI and PR workflows. This simplifies startup of the microservices-messaging Kafka service and removes the manual retry/polling logic. Files changed: .github/workflows/maven-ci.yml, .github/workflows/maven-pr-builder.yml. Note: requires a Docker Compose version that supports the `--wait` flag. * Reformat tests for readability Reformatted KafkaMessageConsumerTest and PaymentServiceTest for readability and consistent formatting: reflowed constructor invocation, expanded anonymous HashMap and schedulePollTask lambda blocks, and aligned assertDoesNotThrow parameters. These are pure style changes with no behavioral modifications. * Remove Kafka Docker Compose steps from CI Removed the Start/Stop Kafka Docker Service steps that ran docker compose for microservices-messaging from .github/workflows/maven-ci.yml and .github/workflows/maven-pr-builder.yml. Workflows no longer start or tear down the Kafka docker-compose service during CI/PR runs; other steps (xvfb install, Maven build, Codecov upload, Sonar cache) remain unchanged. * Bump google-java-format to 1.27.0; tidy tests Update pom.xml to use google-java-format 1.27.0 (was 1.17.0). Make minor formatting cleanups in KafkaMessageConsumerTest and PaymentServiceTest by collapsing multi-line calls into single lines. No functional changes. * Downgrade Google Java Format to 1.17.0 Update Spotless googleJavaFormat version in root pom.xml from 1.27.0 to 1.17.0 * Improve Kafka producer and consumer tests Enhance microservices-messaging tests: add consumer tests to verify run() exits immediately when stopped and gracefully handles a WakeupException (ensuring the MockConsumer is closed). Update producer error test to avoid try-with-resources, assert publish behavior and history, trigger failingProducer.errorNext(...) before closing to cover the error callback branch, then close and assert the mock producer is closed. * Refactor App.run and add messaging tests Extract App.run(...) and introduce a package-private sleepMs field to control sleep durations so the demo can be exercised programmatically and sped up for tests. Main now delegates to run; consumers are started there. Tests updated: AppTest sets sleepMs to 0 and adds testRunWithMockObjects using MockProducer/MockConsumer to exercise run without a Kafka broker; KafkaMessageProducerTest adds a null-message publish test; minor formatting fix in KafkaMessageConsumerTest. These changes improve testability and coverage for the microservices-messaging module.
This commit is contained in:
Mukul Howale
@@ -0,0 +1,247 @@
|
||||
---
|
||||
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
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
## 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<Consumer> 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
|
||||
|
||||
|
||||
|
||||
## How to Run the Application
|
||||
|
||||
### Option 1: Automated Script (Recommended)
|
||||
|
||||
Run the helper script from the module directory, which automatically starts Kafka via Docker Compose (if Docker is installed and Kafka is not already running) and launches the application:
|
||||
|
||||
* **Windows (PowerShell)**:
|
||||
```powershell
|
||||
powershell -ExecutionPolicy Bypass -File .\run-app.ps1
|
||||
```
|
||||
* **Linux / macOS**:
|
||||
```bash
|
||||
./run-app.sh
|
||||
```
|
||||
|
||||
### Option 2: Docker Compose
|
||||
|
||||
Start the Kafka container manually via Docker Compose and run the application:
|
||||
|
||||
```bash
|
||||
# Start Kafka container on port 9092
|
||||
docker compose up -d
|
||||
|
||||
# Run the application
|
||||
../mvnw compile exec:java -Dexec.mainClass="com.iluwatar.messaging.App"
|
||||
|
||||
# Stop Kafka container when finished
|
||||
docker compose down
|
||||
```
|
||||
|
||||
## 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/)
|
||||
@@ -0,0 +1,53 @@
|
||||
#
|
||||
# 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.
|
||||
#
|
||||
|
||||
version: '3.8'
|
||||
|
||||
services:
|
||||
kafka:
|
||||
image: confluentinc/cp-kafka:7.5.0
|
||||
container_name: kafka-messaging-demo
|
||||
ports:
|
||||
- "9092:9092"
|
||||
environment:
|
||||
KAFKA_NODE_ID: 1
|
||||
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
|
||||
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
|
||||
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
|
||||
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
|
||||
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
|
||||
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
|
||||
KAFKA_PROCESS_ROLES: 'broker,controller'
|
||||
KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
|
||||
KAFKA_LISTENERS: 'PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092'
|
||||
KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
|
||||
KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
|
||||
KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
|
||||
CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "kafka-topics --bootstrap-server localhost:9092 --list"]
|
||||
interval: 5s
|
||||
timeout: 10s
|
||||
retries: 5
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 113 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 186 KiB |
@@ -0,0 +1,118 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!--
|
||||
|
||||
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.
|
||||
|
||||
-->
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
|
||||
http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>com.iluwatar</groupId>
|
||||
<artifactId>java-design-patterns</artifactId>
|
||||
<version>1.26.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>microservices-messaging</artifactId>
|
||||
<version>1.26.0-SNAPSHOT</version>
|
||||
|
||||
<properties>
|
||||
<kafka.version>3.6.1</kafka.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<!-- Kafka Client -->
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka-clients</artifactId>
|
||||
<version>${kafka.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Lombok -->
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<version>${lombok.version}</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
|
||||
<!-- Logging -->
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>ch.qos.logback</groupId>
|
||||
<artifactId>logback-classic</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- JSON Processing -->
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
<version>2.16.1</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Testing -->
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-engine</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.mockito</groupId>
|
||||
<artifactId>mockito-core</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.datatype</groupId>
|
||||
<artifactId>jackson-datatype-jsr310</artifactId>
|
||||
<version>2.19.2</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-assembly-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<configuration>
|
||||
<archive>
|
||||
<manifest>
|
||||
<mainClass>com.iluwatar.messaging.App</mainClass>
|
||||
</manifest>
|
||||
</archive>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -0,0 +1,89 @@
|
||||
#
|
||||
# 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.
|
||||
#
|
||||
|
||||
# PowerShell script to start Kafka container (if Docker is available) and run the Microservices Messaging App
|
||||
|
||||
$ScriptDir = Split-Path -Parent $MyInvocation.MyCommand.Definition
|
||||
Set-Location $ScriptDir
|
||||
|
||||
Write-Host "======================================================" -ForegroundColor Cyan
|
||||
Write-Host " Starting Microservices Messaging Pattern Application" -ForegroundColor Cyan
|
||||
Write-Host "======================================================" -ForegroundColor Cyan
|
||||
|
||||
# Check if Kafka is already running on port 9092
|
||||
$portActive = $false
|
||||
try {
|
||||
$socket = New-Object System.Net.Sockets.TcpClient("localhost", 9092)
|
||||
if ($socket.Connected) {
|
||||
$portActive = $true
|
||||
$socket.Close()
|
||||
}
|
||||
} catch {
|
||||
$portActive = $false
|
||||
}
|
||||
|
||||
if ($portActive) {
|
||||
Write-Host "[INFO] Kafka is already running on port 9092." -ForegroundColor Green
|
||||
} else {
|
||||
# Check if Docker is available, or add Docker Desktop to PATH
|
||||
if (-not (Get-Command docker -ErrorAction SilentlyContinue)) {
|
||||
$dockerPath = "$env:LOCALAPPDATA\Programs\DockerDesktop\resources\bin"
|
||||
if (Test-Path $dockerPath) {
|
||||
$env:PATH += ";$dockerPath"
|
||||
}
|
||||
}
|
||||
$dockerCmd = Get-Command docker -ErrorAction SilentlyContinue
|
||||
if ($dockerCmd) {
|
||||
Write-Host "[INFO] Starting Kafka container via Docker Compose..." -ForegroundColor Yellow
|
||||
docker compose up -d
|
||||
|
||||
Write-Host "[INFO] Waiting for Kafka to become ready on port 9092..." -ForegroundColor Yellow
|
||||
$retryCount = 0
|
||||
while (-not $portActive -and $retryCount -lt 20) {
|
||||
Start-Sleep -Seconds 2
|
||||
try {
|
||||
$socket = New-Object System.Net.Sockets.TcpClient("localhost", 9092)
|
||||
if ($socket.Connected) {
|
||||
$portActive = $true
|
||||
$socket.Close()
|
||||
}
|
||||
} catch {
|
||||
$retryCount++
|
||||
}
|
||||
}
|
||||
|
||||
if ($portActive) {
|
||||
Write-Host "[INFO] Kafka container started successfully!" -ForegroundColor Green
|
||||
} else {
|
||||
Write-Host "[WARNING] Kafka did not become ready on port 9092 within timeout." -ForegroundColor Red
|
||||
}
|
||||
} else {
|
||||
Write-Host "[WARNING] Docker is not installed or not in PATH." -ForegroundColor Yellow
|
||||
Write-Host "[INFO] Please ensure Kafka is running locally on localhost:9092 before running the app." -ForegroundColor Yellow
|
||||
}
|
||||
}
|
||||
|
||||
Write-Host "[INFO] Compiling and running App.java..." -ForegroundColor Cyan
|
||||
& "..\mvnw.cmd" compile exec:java "-Dexec.mainClass=com.iluwatar.messaging.App"
|
||||
@@ -0,0 +1,62 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# 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.
|
||||
#
|
||||
|
||||
# Shell script to start Kafka container (if Docker is available) and run the Microservices Messaging App
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
cd "$SCRIPT_DIR"
|
||||
|
||||
echo "======================================================"
|
||||
echo " Starting Microservices Messaging Pattern Application"
|
||||
echo "======================================================"
|
||||
|
||||
if nc -z localhost 9092 2>/dev/null || (echo > /dev/tcp/localhost/9092) 2>/dev/null; then
|
||||
echo "[INFO] Kafka is already running on port 9092."
|
||||
else
|
||||
if command -v docker &> /dev/null; then
|
||||
echo "[INFO] Starting Kafka container via Docker Compose..."
|
||||
docker compose up -d
|
||||
|
||||
echo "[INFO] Waiting for Kafka to become ready on port 9092..."
|
||||
retry=0
|
||||
until nc -z localhost 9092 2>/dev/null || (echo > /dev/tcp/localhost/9092) 2>/dev/null || [ $retry -eq 20 ]; do
|
||||
sleep 2
|
||||
retry=$((retry+1))
|
||||
done
|
||||
|
||||
if [ $retry -lt 20 ]; then
|
||||
echo "[INFO] Kafka container started successfully!"
|
||||
else
|
||||
echo "[WARNING] Kafka did not become ready on port 9092 within timeout."
|
||||
fi
|
||||
else
|
||||
echo "[WARNING] Docker is not installed or not in PATH."
|
||||
echo "[INFO] Please ensure Kafka is running locally on localhost:9092."
|
||||
fi
|
||||
fi
|
||||
|
||||
echo "[INFO] Compiling and running App.java..."
|
||||
../mvnw compile exec:java -Dexec.mainClass="com.iluwatar.messaging.App"
|
||||
@@ -0,0 +1,161 @@
|
||||
/*
|
||||
* 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 java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* The Microservices Messaging pattern enables asynchronous communication between services through
|
||||
* Apache Kafka. This example demonstrates how services can communicate without tight coupling.
|
||||
*
|
||||
* <p>In this example:
|
||||
*
|
||||
* <ul>
|
||||
* <li>OrderService acts as a message producer, publishing order events to Kafka
|
||||
* <li>InventoryService, PaymentService, and NotificationService act as consumers
|
||||
* <li>Apache Kafka acts as the message broker, routing messages between services
|
||||
* </ul>
|
||||
*
|
||||
* <p>Key benefits demonstrated:
|
||||
*
|
||||
* <ul>
|
||||
* <li>Loose coupling - services don't directly depend on each other
|
||||
* <li>Asynchronous processing - producers don't wait for consumers
|
||||
* <li>Scalability - multiple consumers can process messages independently
|
||||
* <li>Resilience - if one consumer fails, others continue processing
|
||||
* <li>Message persistence - Kafka stores messages for reliability
|
||||
* </ul>
|
||||
*
|
||||
* <p><b>Prerequisites:</b> This example requires a running Kafka instance. Start Kafka locally:
|
||||
*
|
||||
* <pre>
|
||||
* # 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
|
||||
* </pre>
|
||||
*/
|
||||
public class App {
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(App.class);
|
||||
private static final String BOOTSTRAP_SERVERS = "localhost:9092";
|
||||
|
||||
/** Sleep duration between operations in milliseconds. Package-private for testing. */
|
||||
static long sleepMs = 2000;
|
||||
|
||||
/**
|
||||
* Program entry point.
|
||||
*
|
||||
* @param args command line arguments
|
||||
*/
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
KafkaMessageProducer producer = new KafkaMessageProducer(BOOTSTRAP_SERVERS);
|
||||
|
||||
InventoryService inventoryService = new InventoryService();
|
||||
PaymentService paymentService = new PaymentService();
|
||||
NotificationService notificationService = new NotificationService();
|
||||
|
||||
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);
|
||||
|
||||
run(producer, inventoryConsumer, paymentConsumer, notificationConsumer);
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs the Microservices Messaging Pattern demonstration.
|
||||
*
|
||||
* @param producer the Kafka message producer
|
||||
* @param inventoryConsumer the inventory service consumer
|
||||
* @param paymentConsumer the payment service consumer
|
||||
* @param notificationConsumer the notification service consumer
|
||||
*/
|
||||
static void run(
|
||||
KafkaMessageProducer producer,
|
||||
KafkaMessageConsumer inventoryConsumer,
|
||||
KafkaMessageConsumer paymentConsumer,
|
||||
KafkaMessageConsumer notificationConsumer)
|
||||
throws InterruptedException {
|
||||
LOGGER.info("Starting Microservices Messaging Pattern with Apache Kafka");
|
||||
|
||||
// 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(sleepMs);
|
||||
|
||||
// Create producer service
|
||||
OrderService orderService = new OrderService(producer);
|
||||
|
||||
// Demonstrate the messaging pattern
|
||||
LOGGER.info("\n=== Creating Order ===");
|
||||
orderService.createOrder("ORDER-001");
|
||||
|
||||
Thread.sleep(sleepMs);
|
||||
|
||||
LOGGER.info("\n=== Updating Order ===");
|
||||
orderService.updateOrder("ORDER-001");
|
||||
|
||||
Thread.sleep(sleepMs);
|
||||
|
||||
LOGGER.info("\n=== Cancelling Order ===");
|
||||
orderService.cancelOrder("ORDER-001");
|
||||
|
||||
Thread.sleep(sleepMs);
|
||||
|
||||
// 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");
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/**
|
||||
* InventoryService is a message consumer that processes inventory-related messages from Kafka. It
|
||||
* listens to order events and updates inventory accordingly.
|
||||
*
|
||||
* <p>This service runs in its own Kafka consumer group (inventory-group) which allows it to:
|
||||
*
|
||||
* <ul>
|
||||
* <li>Process messages independently from other services
|
||||
* <li>Scale horizontally by adding more instances to the consumer group
|
||||
* <li>Resume from last committed offset if the service restarts
|
||||
* </ul>
|
||||
*/
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
/*
|
||||
* 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 java.time.Duration;
|
||||
import java.util.Collections;
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import org.apache.kafka.clients.consumer.Consumer;
|
||||
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;
|
||||
|
||||
/** 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 Consumer<String, String> consumer;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final String topic;
|
||||
private final java.util.function.Consumer<Message> 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,
|
||||
java.util.function.Consumer<Message> messageHandler) {
|
||||
this(createDefaultConsumer(bootstrapServers, groupId), topic, messageHandler);
|
||||
}
|
||||
|
||||
KafkaMessageConsumer(
|
||||
Consumer<String, String> consumer,
|
||||
String topic,
|
||||
java.util.function.Consumer<Message> messageHandler) {
|
||||
this.consumer = consumer;
|
||||
this.objectMapper = new ObjectMapper();
|
||||
this.objectMapper.registerModule(new JavaTimeModule());
|
||||
this.topic = topic;
|
||||
this.messageHandler = messageHandler;
|
||||
}
|
||||
|
||||
private static Consumer<String, String> createDefaultConsumer(
|
||||
String bootstrapServers, String groupId) {
|
||||
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");
|
||||
return new KafkaConsumer<>(props);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
consumer.subscribe(Collections.singletonList(topic));
|
||||
LOGGER.info("Consumer subscribed to topic: {}", topic);
|
||||
|
||||
while (running.get()) {
|
||||
ConsumerRecords<String, String> 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();
|
||||
}
|
||||
}
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
/*
|
||||
* 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 java.util.Properties;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.Producer;
|
||||
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;
|
||||
|
||||
/** 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 Producer<String, String> producer;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
/**
|
||||
* Creates a new Kafka message producer.
|
||||
*
|
||||
* @param bootstrapServers Kafka bootstrap servers
|
||||
*/
|
||||
public KafkaMessageProducer(String bootstrapServers) {
|
||||
this(createDefaultProducer(bootstrapServers));
|
||||
}
|
||||
|
||||
KafkaMessageProducer(Producer<String, String> producer) {
|
||||
this.producer = producer;
|
||||
this.objectMapper = new ObjectMapper();
|
||||
this.objectMapper.registerModule(new JavaTimeModule());
|
||||
}
|
||||
|
||||
private static Producer<String, String> createDefaultProducer(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);
|
||||
return new KafkaProducer<>(props);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<String, String> 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");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
/*
|
||||
* 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 java.time.LocalDateTime;
|
||||
import java.util.UUID;
|
||||
import lombok.Getter;
|
||||
|
||||
/** 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
|
||||
+ '}';
|
||||
}
|
||||
}
|
||||
+100
@@ -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.
|
||||
*
|
||||
* <p>This service runs in its own Kafka consumer group (notification-group) which allows it to:
|
||||
*
|
||||
* <ul>
|
||||
* <li>Process messages independently from other services
|
||||
* <li>Scale horizontally by adding more instances to the consumer group
|
||||
* <li>Resume from last committed offset if the service restarts
|
||||
* </ul>
|
||||
*/
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
/*
|
||||
* 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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/**
|
||||
* PaymentService is a message consumer that processes payment-related messages from Kafka. It
|
||||
* listens to order events and handles payment processing.
|
||||
*
|
||||
* <p>This service runs in its own Kafka consumer group (payment-group) which allows it to:
|
||||
*
|
||||
* <ul>
|
||||
* <li>Process messages independently from other services
|
||||
* <li>Scale horizontally by adding more instances to the consumer group
|
||||
* <li>Resume from last committed offset if the service restarts
|
||||
* </ul>
|
||||
*/
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!--
|
||||
|
||||
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.
|
||||
|
||||
-->
|
||||
<configuration>
|
||||
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<root level="INFO">
|
||||
<appender-ref ref="STDOUT" />
|
||||
</root>
|
||||
|
||||
<logger name="com.iluwatar.messaging" level="INFO"/>
|
||||
</configuration>
|
||||
@@ -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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
|
||||
import org.apache.kafka.clients.consumer.MockConsumer;
|
||||
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
|
||||
import org.apache.kafka.clients.producer.MockProducer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/** Unit tests for {@link App}. Tests main application entry point. */
|
||||
class AppTest {
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
// Speed up sleeps so tests finish instantly
|
||||
App.sleepMs = 0;
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
// Restore default so other contexts are unaffected
|
||||
App.sleepMs = 2000;
|
||||
}
|
||||
|
||||
@Test
|
||||
void testAppConstructor() {
|
||||
assertNotNull(new App(), "App should be instantiable");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testRunWithMockObjects() {
|
||||
// Build mock-backed producer and consumers — no Kafka broker required
|
||||
MockProducer<String, String> mockProducer =
|
||||
new MockProducer<>(true, new StringSerializer(), new StringSerializer());
|
||||
KafkaMessageProducer producer = new KafkaMessageProducer(mockProducer);
|
||||
|
||||
MockConsumer<String, String> mc1 = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
|
||||
MockConsumer<String, String> mc2 = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
|
||||
MockConsumer<String, String> mc3 = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
|
||||
|
||||
KafkaMessageConsumer inventoryConsumer =
|
||||
new KafkaMessageConsumer(mc1, "order-topic", msg -> {});
|
||||
KafkaMessageConsumer paymentConsumer = new KafkaMessageConsumer(mc2, "order-topic", msg -> {});
|
||||
KafkaMessageConsumer notificationConsumer =
|
||||
new KafkaMessageConsumer(mc3, "order-topic", msg -> {});
|
||||
|
||||
// Stop consumers immediately so their poll loops exit right away in the executor threads
|
||||
inventoryConsumer.stop();
|
||||
paymentConsumer.stop();
|
||||
notificationConsumer.stop();
|
||||
|
||||
// sleepMs == 0, so all Thread.sleep(sleepMs) return instantly — full run() coverage
|
||||
assertDoesNotThrow(
|
||||
() -> App.run(producer, inventoryConsumer, paymentConsumer, notificationConsumer),
|
||||
"App.run() should complete without throwing");
|
||||
}
|
||||
}
|
||||
+118
@@ -0,0 +1,118 @@
|
||||
/*
|
||||
* 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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* 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");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testHandleMessagesWhenInterrupted() {
|
||||
Thread.currentThread().interrupt();
|
||||
inventoryService.handleMessage(new Message("Order Created: ORDER-001"));
|
||||
org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
|
||||
|
||||
Thread.currentThread().interrupt();
|
||||
inventoryService.handleMessage(new Message("Order Cancelled: ORDER-001"));
|
||||
org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
|
||||
}
|
||||
}
|
||||
+137
@@ -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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.MockConsumer;
|
||||
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
|
||||
import org.apache.kafka.common.TopicPartition;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/** Unit tests for {@link KafkaMessageConsumer}. */
|
||||
class KafkaMessageConsumerTest {
|
||||
|
||||
private MockConsumer<String, String> mockConsumer;
|
||||
private KafkaMessageConsumer kafkaMessageConsumer;
|
||||
private AtomicBoolean handlerCalled;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
|
||||
handlerCalled = new AtomicBoolean(false);
|
||||
kafkaMessageConsumer =
|
||||
new KafkaMessageConsumer(mockConsumer, "test-topic", msg -> handlerCalled.set(true));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testConsumerCanBeInstantiated() {
|
||||
assertNotNull(kafkaMessageConsumer, "KafkaMessageConsumer should be instantiated");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testRunProcessesValidMessageAndStops() throws Exception {
|
||||
ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule());
|
||||
Message msg = new Message("Order Created: 123");
|
||||
String jsonStr = mapper.writeValueAsString(msg);
|
||||
|
||||
TopicPartition tp = new TopicPartition("test-topic", 0);
|
||||
mockConsumer.updateBeginningOffsets(
|
||||
new HashMap<>() {
|
||||
{
|
||||
put(tp, 0L);
|
||||
}
|
||||
});
|
||||
|
||||
mockConsumer.schedulePollTask(
|
||||
() -> {
|
||||
mockConsumer.rebalance(Collections.singletonList(tp));
|
||||
mockConsumer.addRecord(new ConsumerRecord<>("test-topic", 0, 0L, "key", jsonStr));
|
||||
});
|
||||
|
||||
mockConsumer.schedulePollTask(() -> kafkaMessageConsumer.stop());
|
||||
|
||||
kafkaMessageConsumer.run();
|
||||
|
||||
assertTrue(handlerCalled.get(), "Handler should have been invoked");
|
||||
assertTrue(mockConsumer.closed(), "Consumer should be closed");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testRunHandlesInvalidJsonMessage() {
|
||||
TopicPartition tp = new TopicPartition("test-topic", 0);
|
||||
mockConsumer.updateBeginningOffsets(
|
||||
new HashMap<>() {
|
||||
{
|
||||
put(tp, 0L);
|
||||
}
|
||||
});
|
||||
|
||||
mockConsumer.schedulePollTask(
|
||||
() -> {
|
||||
mockConsumer.rebalance(Collections.singletonList(tp));
|
||||
mockConsumer.addRecord(new ConsumerRecord<>("test-topic", 0, 0L, "key", "{invalid json"));
|
||||
});
|
||||
|
||||
mockConsumer.schedulePollTask(() -> kafkaMessageConsumer.stop());
|
||||
|
||||
assertDoesNotThrow(() -> kafkaMessageConsumer.run());
|
||||
assertTrue(mockConsumer.closed(), "Consumer should be closed");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCloseStopsConsumer() {
|
||||
assertDoesNotThrow(() -> kafkaMessageConsumer.close());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testRunWhenAlreadyStopped() {
|
||||
// Stop the consumer before run() so the while loop exits immediately
|
||||
kafkaMessageConsumer.stop();
|
||||
|
||||
// run() should complete without processing any records
|
||||
assertDoesNotThrow(() -> kafkaMessageConsumer.run());
|
||||
assertTrue(mockConsumer.closed(), "Consumer should be closed even when stopped before run");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testRunHandlesConsumerException() {
|
||||
// Schedule a WakeupException during poll to trigger the outer catch block
|
||||
mockConsumer.schedulePollTask(() -> mockConsumer.wakeup());
|
||||
|
||||
// run() must not propagate the exception
|
||||
assertDoesNotThrow(() -> kafkaMessageConsumer.run());
|
||||
assertTrue(mockConsumer.closed(), "Consumer should be closed after exception");
|
||||
}
|
||||
}
|
||||
+93
@@ -0,0 +1,93 @@
|
||||
/*
|
||||
* 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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import org.apache.kafka.clients.producer.MockProducer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/** Unit tests for {@link KafkaMessageProducer}. */
|
||||
class KafkaMessageProducerTest {
|
||||
|
||||
private MockProducer<String, String> mockProducer;
|
||||
private KafkaMessageProducer kafkaMessageProducer;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
mockProducer = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
|
||||
kafkaMessageProducer = new KafkaMessageProducer(mockProducer);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testProducerCanBeInstantiated() {
|
||||
assertNotNull(kafkaMessageProducer, "KafkaMessageProducer should be instantiated");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPublishMessageSuccess() {
|
||||
Message message = new Message("Test Order");
|
||||
|
||||
assertDoesNotThrow(() -> kafkaMessageProducer.publish("test-topic", message));
|
||||
assertEquals(1, mockProducer.history().size());
|
||||
assertEquals("test-topic", mockProducer.history().get(0).topic());
|
||||
assertEquals(message.getId(), mockProducer.history().get(0).key());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPublishMessageErrorCallback() {
|
||||
MockProducer<String, String> failingProducer =
|
||||
new MockProducer<>(false, new StringSerializer(), new StringSerializer());
|
||||
KafkaMessageProducer producerWithError = new KafkaMessageProducer(failingProducer);
|
||||
Message message = new Message("Test Order");
|
||||
|
||||
assertDoesNotThrow(() -> producerWithError.publish("test-topic", message));
|
||||
assertEquals(1, failingProducer.history().size());
|
||||
|
||||
// Trigger error callback BEFORE closing so the exception != null branch is covered
|
||||
failingProducer.errorNext(new RuntimeException("Kafka publish error"));
|
||||
|
||||
assertDoesNotThrow(() -> producerWithError.close());
|
||||
assertTrue(failingProducer.closed());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testClose() {
|
||||
assertDoesNotThrow(() -> kafkaMessageProducer.close());
|
||||
assertTrue(mockProducer.closed());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPublishNullMessageCatchesException() {
|
||||
// Passing null causes message.getId() to throw NPE, which is caught by the
|
||||
// catch(Exception e) block — covering the "Error serializing message" log branch.
|
||||
assertDoesNotThrow(() -> kafkaMessageProducer.publish("test-topic", null));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
/*
|
||||
* 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 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;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
|
||||
import java.time.LocalDateTime;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* 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");
|
||||
}
|
||||
}
|
||||
+134
@@ -0,0 +1,134 @@
|
||||
/*
|
||||
* 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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* 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");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testHandleMessagesWhenInterrupted() {
|
||||
Thread.currentThread().interrupt();
|
||||
notificationService.handleMessage(new Message("Order Created: ORDER-001"));
|
||||
org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
|
||||
|
||||
Thread.currentThread().interrupt();
|
||||
notificationService.handleMessage(new Message("Order Updated: ORDER-001"));
|
||||
org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
|
||||
|
||||
Thread.currentThread().interrupt();
|
||||
notificationService.handleMessage(new Message("Order Cancelled: ORDER-001"));
|
||||
org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
/*
|
||||
* 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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
import org.apache.kafka.clients.producer.MockProducer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/** Unit tests for {@link OrderService}. Tests follow Arrange-Act-Assert pattern. */
|
||||
class OrderServiceTest {
|
||||
|
||||
private MockProducer<String, String> mockKafkaProducer;
|
||||
private KafkaMessageProducer messageProducer;
|
||||
private OrderService orderService;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
mockKafkaProducer = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
|
||||
messageProducer = new KafkaMessageProducer(mockKafkaProducer);
|
||||
orderService = new OrderService(messageProducer);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCreateOrder() {
|
||||
// Arrange
|
||||
var orderId = "ORDER-001";
|
||||
|
||||
// Act
|
||||
assertDoesNotThrow(() -> orderService.createOrder(orderId));
|
||||
|
||||
// Assert
|
||||
assertEquals(1, mockKafkaProducer.history().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testUpdateOrder() {
|
||||
// Arrange
|
||||
var orderId = "ORDER-002";
|
||||
|
||||
// Act
|
||||
assertDoesNotThrow(() -> orderService.updateOrder(orderId));
|
||||
|
||||
// Assert
|
||||
assertEquals(1, mockKafkaProducer.history().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCancelOrder() {
|
||||
// Arrange
|
||||
var orderId = "ORDER-003";
|
||||
|
||||
// Act
|
||||
assertDoesNotThrow(() -> orderService.cancelOrder(orderId));
|
||||
|
||||
// Assert
|
||||
assertEquals(1, mockKafkaProducer.history().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testMultipleOrderOperations() {
|
||||
// Arrange
|
||||
var orderId = "ORDER-004";
|
||||
|
||||
// Act
|
||||
orderService.createOrder(orderId);
|
||||
orderService.updateOrder(orderId);
|
||||
orderService.cancelOrder(orderId);
|
||||
|
||||
// Assert
|
||||
assertEquals(3, mockKafkaProducer.history().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCreateOrderWithDifferentIds() {
|
||||
// Act
|
||||
orderService.createOrder("ORDER-001");
|
||||
orderService.createOrder("ORDER-002");
|
||||
orderService.createOrder("ORDER-003");
|
||||
|
||||
// Assert
|
||||
assertEquals(3, mockKafkaProducer.history().size());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
/*
|
||||
* 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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* 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");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testHandleMessagesWhenInterrupted() {
|
||||
Thread.currentThread().interrupt();
|
||||
paymentService.handleMessage(new Message("Order Created: ORDER-001"));
|
||||
org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
|
||||
|
||||
Thread.currentThread().interrupt();
|
||||
paymentService.handleMessage(new Message("Order Cancelled: ORDER-001"));
|
||||
org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
|
||||
}
|
||||
}
|
||||
@@ -170,6 +170,7 @@
|
||||
<module>microservices-idempotent-consumer</module>
|
||||
<module>microservices-log-aggregation</module>
|
||||
<module>microservices-self-registration</module>
|
||||
<module>microservices-messaging</module>
|
||||
<module>model-view-controller</module>
|
||||
<module>model-view-intent</module>
|
||||
<module>model-view-presenter</module>
|
||||
@@ -453,6 +454,7 @@
|
||||
<exclude>src/test/resources/**</exclude>
|
||||
<exclude>src/main/resources/**</exclude>
|
||||
<exclude>checkstyle-suppressions.xml</exclude>
|
||||
<exclude>**/*.ps1</exclude>
|
||||
</excludes>
|
||||
</licenseSet>
|
||||
</licenseSets>
|
||||
|
||||
Reference in New Issue
Block a user