JAVA-15787 Created new messaging-modules and saas-modules

- Moved jgroups, rabbitmq, spring-amqp, spring-apache-camel, spring-jms to messaging-modules
- Moved twilio, twitter4j, strip to saas-modules
- Renamed existing saas to jira-rest-integration
This commit is contained in:
Dhawal Kapil
2022-11-26 11:54:53 +05:30
parent 8ee7dcd350
commit ad9ddfc6e3
142 changed files with 91 additions and 57 deletions
+12
View File
@@ -0,0 +1,12 @@
## Reliable Messaging with JGroups Tutorial Project
This module contains articles about JGroups.
### Relevant Article:
- [Reliable Messaging with JGroups](https://www.baeldung.com/jgroups)
### Overview
This Maven project contains the Java code for the article linked above.
### Package Organization
Java classes for the intro tutorial are in the org.baeldung.jgroups package.
+34
View File
@@ -0,0 +1,34 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
<artifactId>jgroups</artifactId>
<name>jgroups</name>
<packaging>jar</packaging>
<description>Reliable Messaging with JGroups Tutorial</description>
<parent>
<groupId>com.baeldung</groupId>
<artifactId>messaging-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
<groupId>org.jgroups</groupId>
<artifactId>jgroups</artifactId>
<version>${jgroups.version}</version>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
<version>${commons-cli.version}</version>
</dependency>
</dependencies>
<properties>
<jgroups.version>4.0.10.Final</jgroups.version>
</properties>
</project>
@@ -0,0 +1,212 @@
package com.baeldung.jgroups;
import org.apache.commons.cli.*;
import org.jgroups.*;
import org.jgroups.util.Util;
import java.io.*;
import java.util.List;
import java.util.Optional;
public class JGroupsMessenger extends ReceiverAdapter {
private JChannel channel;
private String userName;
private String clusterName;
private View lastView;
private boolean running = true;
// Our shared state
private Integer messageCount = 0;
/**
* Connect to a JGroups cluster using command line options
* @param args command line arguments
* @throws Exception
*/
private void start(String[] args) throws Exception {
processCommandline(args);
// Create the channel
// This file could be moved, or made a command line option.
channel = new JChannel("src/main/resources/udp.xml");
// Set a name
channel.name(userName);
// Register for callbacks
channel.setReceiver(this);
// Ignore our messages
channel.setDiscardOwnMessages(true);
// Connect
channel.connect(clusterName);
// Start state transfer
channel.getState(null, 0);
// Do the things
processInput();
// Clean up
channel.close();
}
/**
* Quick and dirty implementaton of commons cli for command line args
* @param args the command line args
* @throws ParseException
*/
private void processCommandline(String[] args) throws ParseException {
// Options, parser, friendly help
Options options = new Options();
CommandLineParser parser = new DefaultParser();
HelpFormatter formatter = new HelpFormatter();
options.addOption("u", "user", true, "User name")
.addOption("c", "cluster", true, "Cluster name");
CommandLine line = parser.parse(options, args);
if (line.hasOption("user")) {
userName = line.getOptionValue("user");
} else {
formatter.printHelp("JGroupsMessenger: need a user name.\n", options);
System.exit(-1);
}
if (line.hasOption("cluster")) {
clusterName = line.getOptionValue("cluster");
} else {
formatter.printHelp("JGroupsMessenger: need a cluster name.\n", options);
System.exit(-1);
}
}
// Start it up
public static void main(String[] args) throws Exception {
new JGroupsMessenger().start(args);
}
@Override
public void viewAccepted(View newView) {
// Save view if this is the first
if (lastView == null) {
System.out.println("Received initial view:");
newView.forEach(System.out::println);
} else {
// Compare to last view
System.out.println("Received new view.");
List<Address> newMembers = View.newMembers(lastView, newView);
System.out.println("New members: ");
newMembers.forEach(System.out::println);
List<Address> exMembers = View.leftMembers(lastView, newView);
System.out.println("Exited members:");
exMembers.forEach(System.out::println);
}
lastView = newView;
}
/**
* Loop on console input until we see 'x' to exit
*/
private void processInput() throws Exception {
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while (running) {
try {
// Get a destination, <enter> means broadcast
Address destination = null;
System.out.print("Enter a destination: ");
System.out.flush();
String destinationName = in.readLine().toLowerCase();
if (destinationName.equals("x")) {
running = false;
continue;
} else if (!destinationName.isEmpty()) {
destination = getAddress(destinationName)
. orElseThrow(() -> new Exception("Destination not found"));
}
// Accept a string to send
System.out.print("Enter a message: ");
System.out.flush();
String line = in.readLine().toLowerCase();
sendMessage(destination, line);
} catch (IOException ioe) {
running = false;
}
}
System.out.println("Exiting.");
}
/**
* Send message from here
* @param destination the destination
* @param messageString the message
*/
private void sendMessage(Address destination, String messageString) {
try {
System.out.println("Sending " + messageString + " to " + destination);
Message message = new Message(destination, messageString);
channel.send(message);
} catch (Exception exception) {
System.err.println("Exception sending message: " + exception.getMessage());
running = false;
}
}
@Override
public void receive(Message message) {
// Print source and dest with message
String line = "Message received from: " + message.getSrc() + " to: " + message.getDest() + " -> " + message.getObject();
// Only track the count of broadcast messages
// Tracking direct message would make for a pointless state
if (message.getDest() == null) {
messageCount++;
System.out.println("Message count: " + messageCount);
}
System.out.println(line);
}
@Override
public void getState(OutputStream output) throws Exception {
// Serialize into the stream
Util.objectToStream(messageCount, new DataOutputStream(output));
}
@Override
public void setState(InputStream input) {
// NOTE: since we know that incrementing the count and transferring the state
// is done inside the JChannel's thread, we don't have to worry about synchronizing
// messageCount. For production code it should be synchronized!
try {
// Deserialize
messageCount = Util.objectFromStream(new DataInputStream(input));
} catch (Exception e) {
System.out.println("Error deserialing state!");
}
System.out.println(messageCount + " is the current messagecount.");
}
private Optional<Address> getAddress(String name) {
View view = channel.view();
return view.getMembers().stream().filter(address -> name.equals(address.toString())).findFirst();
}
}
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
</configuration>
@@ -0,0 +1,48 @@
<config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/jgroups.xsd">
<UDP
mcast_port="${jgroups.udp.mcast_port:45588}"
ip_ttl="4"
tos="8"
ucast_recv_buf_size="5M"
ucast_send_buf_size="5M"
mcast_recv_buf_size="5M"
mcast_send_buf_size="5M"
max_bundle_size="64K"
enable_diagnostics="true"
thread_naming_pattern="cl"
thread_pool.min_threads="0"
thread_pool.max_threads="20"
thread_pool.keep_alive_time="30000"/>
<PING />
<MERGE3 max_interval="30000"
min_interval="10000"/>
<FD_SOCK/>
<FD_ALL/>
<VERIFY_SUSPECT timeout="1500" />
<BARRIER />
<pbcast.NAKACK2 xmit_interval="500"
xmit_table_num_rows="100"
xmit_table_msgs_per_row="2000"
xmit_table_max_compaction_time="30000"
use_mcast_xmit="false"
discard_delivered_msgs="true"/>
<UNICAST3 xmit_interval="500"
xmit_table_num_rows="100"
xmit_table_msgs_per_row="2000"
xmit_table_max_compaction_time="60000"
conn_expiry_timeout="0"/>
<pbcast.STABLE desired_avg_gossip="50000"
max_bytes="4M"/>
<pbcast.GMS print_local_addr="true" join_timeout="2000"/>
<UFC max_credits="2M"
min_threshold="0.4"/>
<MFC max_credits="2M"
min_threshold="0.4"/>
<FRAG2 frag_size="60K" />
<RSVP resend_interval="2000" timeout="10000"/>
<pbcast.STATE_TRANSFER />
</config>
+24
View File
@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
<artifactId>messaging-modules</artifactId>
<name>messaging-modules</name>
<packaging>pom</packaging>
<parent>
<artifactId>parent-boot-2</artifactId>
<groupId>com.baeldung</groupId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../parent-boot-2</relativePath>
</parent>
<modules>
<module>jgroups</module>
<module>rabbitmq</module>
<module>spring-amqp</module>
<module>spring-apache-camel</module>
<module>spring-jms</module>
</modules>
</project>
+10
View File
@@ -0,0 +1,10 @@
## RabbitMQ
This module contains articles about RabbitMQ.
### Relevant articles
- [Introduction to RabbitMQ](https://www.baeldung.com/rabbitmq)
- [Exchanges, Queues, and Bindings in RabbitMQ](https://www.baeldung.com/java-rabbitmq-exchanges-queues-bindings)
- [Pub-Sub vs. Message Queues](https://www.baeldung.com/pub-sub-vs-message-queues)
- [Channels and Connections in RabbitMQ](https://www.baeldung.com/java-rabbitmq-channels-connections)
@@ -0,0 +1,14 @@
version: '3.0'
services:
rabbitmq:
image: rabbitmq:3-management
environment:
- RABBITMQ_DEFAULT_USER=guest
- RABBITMQ_DEFAULT_PASS=guest
- RABBITMQ_VM_MEMORY_HIGH_WATERMARK_RELATIVE=0.8
ports:
- "5672:5672"
- "15672:15672"
volumes:
- ./src/rabbitmq/20-mem.conf:/etc/rabbitmq/conf.d/20-mem.conf
+48
View File
@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
<artifactId>rabbitmq</artifactId>
<version>0.1-SNAPSHOT</version>
<name>rabbitmq</name>
<parent>
<groupId>com.baeldung</groupId>
<artifactId>messaging-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud-dependencies.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
<properties>
<spring-cloud-dependencies.version>2020.0.3</spring-cloud-dependencies.version>
</properties>
</project>
@@ -0,0 +1,97 @@
package com.baeldung.benchmark;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.LongSummaryStatistics;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.baeldung.benchmark.Worker.WorkerResult;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class ConnectionPerChannelPublisher implements Callable<Long> {
private static final Logger log = LoggerFactory.getLogger(ConnectionPerChannelPublisher.class);
private final ConnectionFactory factory;
private final int workerCount;
private final int iterations;
private final int payloadSize;
ConnectionPerChannelPublisher(ConnectionFactory factory, int workerCount, int iterations, int payloadSize) {
this.factory = factory;
this.workerCount = workerCount;
this.iterations = iterations;
this.payloadSize = payloadSize;
}
public static void main(String[] args) {
if (args.length != 4) {
System.err.println("Usage: java " + ConnectionPerChannelPublisher.class.getName() + " <host> <#channels> <#messages> <payloadSize>");
System.exit(1);
}
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(args[0]);
int workerCount = Integer.parseInt(args[1]);
int iterations = Integer.parseInt(args[2]);
int payloadSize = Integer.parseInt(args[3]);
// run the benchmark 10x and get the average throughput
LongSummaryStatistics summary = IntStream.range(0, 9)
.mapToObj(idx -> new ConnectionPerChannelPublisher(factory, workerCount, iterations, payloadSize))
.map(p -> p.call())
.collect(Collectors.summarizingLong((l) -> l));
log.info("[I66] workers={}, throughput={}", workerCount, (int)Math.floor(summary.getAverage()));
}
@Override
public Long call() {
try {
List<Worker> workers = new ArrayList<>();
CountDownLatch counter = new CountDownLatch(workerCount);
for (int i = 0; i < workerCount; i++) {
Connection conn = factory.newConnection();
workers.add(new Worker("queue_" + i, conn, iterations, counter, payloadSize));
}
ExecutorService executor = new ThreadPoolExecutor(workerCount, workerCount, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(workerCount, true));
long start = System.currentTimeMillis();
log.info("[I61] Starting {} workers...", workers.size());
executor.invokeAll(workers);
if (counter.await(5, TimeUnit.MINUTES)) {
long elapsed = System.currentTimeMillis() - start;
log.info("[I59] Tasks completed: #workers={}, #iterations={}, elapsed={}ms, stats={}", workerCount, iterations, elapsed);
return throughput(workerCount, iterations, elapsed);
} else {
throw new RuntimeException("[E61] Timeout waiting workers to complete");
}
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
private static long throughput(int workerCount, int iterations, long elapsed) {
return (iterations * workerCount * 1000) / elapsed;
}
}
@@ -0,0 +1,165 @@
package com.baeldung.benchmark;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.LongSummaryStatistics;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class SharedConnectionPublisher {
private static final Logger log = LoggerFactory.getLogger(SharedConnectionPublisher.class);
public static void main(String[] args) {
try {
if ( args.length != 6) {
System.err.println("Usage: java " + SharedConnectionPublisher.class.getName() + " <host> <#channels> <#messages> <payloadSize> <#channels/connection> <extra work time(ms)>");
System.exit(1);
}
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(args[0]);
List<Worker> workers = new ArrayList<>();
int workerCount = Integer.parseInt(args[1]);
int iterations = Integer.parseInt(args[2]);
int payloadSize = Integer.parseInt(args[3]);
int channelsPerConnection = Integer.parseInt(args[4]);
long extraWork = Long.parseLong(args[5]);
log.info("[I35] Creating {} worker{}...", workerCount, (workerCount > 1)?"s":"");
CountDownLatch counter = new CountDownLatch(workerCount);
int connCount = (workerCount + channelsPerConnection-1)/channelsPerConnection;
List<Connection> connections = new ArrayList<>(connCount);
for( int i =0 ; i< connCount; i++) {
log.info("[I59] Creating connection#{}", i);
connections.add(factory.newConnection());
}
for( int i = 0 ; i < workerCount ; i++ ) {
workers.add(new Worker("queue_" + i, connections.get(i % connCount), iterations, counter,payloadSize,extraWork));
}
ExecutorService executor = new ThreadPoolExecutor(workerCount, workerCount, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(workerCount, true));
long start = System.currentTimeMillis();
log.info("[I61] Starting workers...");
List<Future<WorkerResult>> results = executor.invokeAll(workers);
log.info("[I55] Waiting workers to complete...");
if( counter.await(5, TimeUnit.MINUTES) ) {
long elapsed = System.currentTimeMillis() - start - (workerCount*iterations*extraWork);
log.info("[I59] Tasks completed: #workers={}, #iterations={}, elapsed={}ms",
workerCount,
iterations,
elapsed);
LongSummaryStatistics summary = results.stream()
.map(f -> safeGet(f))
.map(r -> r.elapsed)
.collect(Collectors.summarizingLong((l) -> l));
log.info("[I74] stats={}", summary);
log.info("[I79] result: workers={}, throughput={}",workerCount,throughput(workerCount,iterations,elapsed));
}
else {
log.error("[E61] Timeout waiting workers to complete");
}
}
catch(Exception ex) {
throw new RuntimeException(ex);
}
}
private static long throughput(int workerCount, int iterations, long elapsed) {
return (iterations*workerCount*1000)/elapsed;
}
private static <T> T safeGet(Future<T> f) {
try {
return f.get();
}
catch(Exception ex) {
throw new RuntimeException(ex);
}
}
private static class WorkerResult {
public final long elapsed;
WorkerResult(long elapsed) {
this.elapsed = elapsed;
}
}
private static class Worker implements Callable<WorkerResult> {
private final Connection conn;
private final Channel channel;
private int iterations;
private final CountDownLatch counter;
private final String queue;
private final byte[] payload;
private long extraWork;
Worker(String queue, Connection conn, int iterations, CountDownLatch counter,int payloadSize,long extraWork) throws IOException {
this.conn = conn;
this.iterations = iterations;
this.counter = counter;
this.queue = queue;
this.extraWork = extraWork;
channel = conn.createChannel();
channel.queueDeclare(queue, false, false, true, null);
this.payload = new byte[payloadSize];
new Random().nextBytes(payload);
}
@Override
public WorkerResult call() throws Exception {
try {
long start = System.currentTimeMillis();
for ( int i = 0 ; i < iterations ; i++ ) {
channel.basicPublish("", queue, null,payload);
Thread.sleep(extraWork);
}
long elapsed = System.currentTimeMillis() - start - (extraWork*iterations);
channel.queueDelete(queue);
return new WorkerResult(elapsed);
}
finally {
counter.countDown();
}
}
}
}
@@ -0,0 +1,116 @@
package com.baeldung.benchmark;
import java.util.ArrayList;
import java.util.List;
import java.util.LongSummaryStatistics;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.baeldung.benchmark.Worker.WorkerResult;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class SingleConnectionPublisher implements Callable<Long> {
private static final Logger log = LoggerFactory.getLogger(SingleConnectionPublisher.class);
private final ConnectionFactory factory;
private final int workerCount;
private final int iterations;
private final int payloadSize;
SingleConnectionPublisher(ConnectionFactory factory, int workerCount, int iterations, int payloadSize) {
this.factory = factory;
this.workerCount = workerCount;
this.iterations = iterations;
this.payloadSize = payloadSize;
}
public static void main(String[] args) {
if ( args.length != 4) {
System.err.println("Usage: java " + SingleConnectionPublisher.class.getName() + " <host> <#channels> <#messages> <payloadSize>");
System.exit(1);
}
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(args[0]);
int workerCount = Integer.parseInt(args[1]);
int iterations = Integer.parseInt(args[2]);
int payloadSize = Integer.parseInt(args[3]);
LongSummaryStatistics summary = IntStream.range(0, 9)
.mapToObj(idx -> new SingleConnectionPublisher(factory, workerCount, iterations, payloadSize))
.map(p -> p.call())
.collect(Collectors.summarizingLong((l) -> l));
log.info("[I66] workers={}, throughput={}", workerCount, (int)Math.floor(summary.getAverage()));
}
@Override
public Long call() {
try {
Connection connection = factory.newConnection();
CountDownLatch counter = new CountDownLatch(workerCount);
List<Worker> workers = new ArrayList<>();
for( int i = 0 ; i < workerCount ; i++ ) {
workers.add(new Worker("queue_" + i, connection, iterations, counter,payloadSize));
}
ExecutorService executor = new ThreadPoolExecutor(workerCount, workerCount, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(workerCount, true));
long start = System.currentTimeMillis();
log.info("[I61] Starting {} workers...", workers.size());
List<Future<WorkerResult>> results = executor.invokeAll(workers);
if( counter.await(5, TimeUnit.MINUTES) ) {
long elapsed = System.currentTimeMillis() - start;
LongSummaryStatistics summary = results.stream()
.map(f -> safeGet(f))
.map(r -> r.elapsed)
.collect(Collectors.summarizingLong((l) -> l));
log.info("[I59] Tasks completed: #workers={}, #iterations={}, elapsed={}ms, stats={}",
workerCount,
iterations,
elapsed, summary);
return throughput(workerCount,iterations,elapsed);
}
else {
throw new RuntimeException("[E61] Timeout waiting workers to complete");
}
}
catch(Exception ex) {
throw new RuntimeException(ex);
}
}
private static <T> T safeGet(Future<T> f) {
try {
return f.get();
}
catch(Exception ex) {
throw new RuntimeException(ex);
}
}
private static long throughput(int workerCount, int iterations, long elapsed) {
return (iterations*workerCount*1000)/elapsed;
}
}
@@ -0,0 +1,151 @@
package com.baeldung.benchmark;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.LongSummaryStatistics;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class SingleConnectionPublisherNio {
private static final Logger log = LoggerFactory.getLogger(SingleConnectionPublisherNio.class);
public static void main(String[] args) {
try {
if ( args.length != 4) {
System.err.println("Usage: java " + SingleConnectionPublisherNio.class.getName() + " <host> <#channels> <#messages> <payloadSize>");
System.exit(1);
}
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(args[0]);
factory.useNio();
Connection connection = factory.newConnection();
List<Worker> workers = new ArrayList<>();
int workerCount = Integer.parseInt(args[1]);
int iterations = Integer.parseInt(args[2]);
int payloadSize = Integer.parseInt(args[3]);
log.info("[I35] Creating {} worker{}...", workerCount, (workerCount > 1)?"s":"");
CountDownLatch counter = new CountDownLatch(workerCount);
for( int i = 0 ; i < workerCount ; i++ ) {
workers.add(new Worker("queue_" + i, connection, iterations, counter,payloadSize));
}
ExecutorService executor = new ThreadPoolExecutor(workerCount, workerCount, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(workerCount, true));
long start = System.currentTimeMillis();
log.info("[I61] Starting workers...");
List<Future<WorkerResult>> results = executor.invokeAll(workers);
log.info("[I55] Waiting workers to complete...");
if( counter.await(5, TimeUnit.MINUTES) ) {
long elapsed = System.currentTimeMillis() - start;
log.info("[I59] Tasks completed: #workers={}, #iterations={}, elapsed={}ms",
workerCount,
iterations,
elapsed);
LongSummaryStatistics summary = results.stream()
.map(f -> safeGet(f))
.map(r -> r.elapsed)
.collect(Collectors.summarizingLong((l) -> l));
log.info("[I74] stats={}", summary);
}
else {
log.error("[E61] Timeout waiting workers to complete");
}
}
catch(Exception ex) {
throw new RuntimeException(ex);
}
}
private static <T> T safeGet(Future<T> f) {
try {
return f.get();
}
catch(Exception ex) {
throw new RuntimeException(ex);
}
}
private static class WorkerResult {
public final long elapsed;
WorkerResult(long elapsed) {
this.elapsed = elapsed;
}
}
private static class Worker implements Callable<WorkerResult> {
private final Connection conn;
private final Channel channel;
private int iterations;
private final CountDownLatch counter;
private final String queue;
private final byte[] payload;
Worker(String queue, Connection conn, int iterations, CountDownLatch counter,int payloadSize) throws IOException {
this.conn = conn;
this.iterations = iterations;
this.counter = counter;
this.queue = queue;
channel = conn.createChannel();
channel.queueDeclare(queue, false, false, true, null);
this.payload = new byte[payloadSize];
new Random().nextBytes(payload);
}
@Override
public WorkerResult call() throws Exception {
try {
long start = System.currentTimeMillis();
for ( int i = 0 ; i < iterations ; i++ ) {
channel.basicPublish("", queue, null,payload);
}
long elapsed = System.currentTimeMillis() - start;
channel.queueDelete(queue);
return new WorkerResult(elapsed);
}
finally {
counter.countDown();
}
}
}
}
@@ -0,0 +1,59 @@
package com.baeldung.benchmark;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Worker implements Callable<Worker.WorkerResult> {
private final Channel channel;
private int iterations;
private final CountDownLatch counter;
private final String queue;
private final byte[] payload;
Worker(String queue, Connection conn, int iterations, CountDownLatch counter, int payloadSize) throws IOException {
this.iterations = iterations;
this.counter = counter;
this.queue = queue;
channel = conn.createChannel();
channel.queueDelete(queue);
channel.queueDeclare(queue, false, false, true, null);
this.payload = new byte[payloadSize];
new Random().nextBytes(payload);
}
@Override
public WorkerResult call() throws Exception {
try {
long start = System.currentTimeMillis();
for (int i = 0; i < iterations; i++) {
channel.basicPublish("", queue, null, payload);
}
long elapsed = System.currentTimeMillis() - start;
channel.queueDelete(queue);
return new WorkerResult(elapsed);
} finally {
counter.countDown();
}
}
public static class WorkerResult {
public final long elapsed;
WorkerResult(long elapsed) {
this.elapsed = elapsed;
}
}
}
@@ -0,0 +1,31 @@
package com.baeldung.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receiver {
private static final String QUEUE_NAME = "products_queue";
public static void main (String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + message + "'");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
@@ -0,0 +1,27 @@
package com.baeldung.producer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Publisher {
private final static String QUEUE_NAME = "products_queue";
public static void main(String[]args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
String message = "product details";
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}
}
@@ -0,0 +1,43 @@
package com.baeldung.pubsubmq.client;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class ClientApplication {
private static final String MESSAGE_QUEUE = "pizza-message-queue";
@Bean
public Queue queue() {
return new Queue(MESSAGE_QUEUE);
}
@Bean
public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(MESSAGE_QUEUE);
container.setMessageListener(listenerAdapter);
return container;
}
@Bean
public Consumer consumer() {
return new Consumer();
}
@Bean
public MessageListenerAdapter listenerAdapter(Consumer consumer) {
return new MessageListenerAdapter(consumer, "receiveOrder");
}
public static void main(String[] args) {
SpringApplication.run(ClientApplication.class, args);
}
}
@@ -0,0 +1,7 @@
package com.baeldung.pubsubmq.client;
public class Consumer {
public void receiveOrder(String message) {
System.out.printf("Order received: %s%n", message);
}
}
@@ -0,0 +1,28 @@
package com.baeldung.pubsubmq.server;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import javax.annotation.PostConstruct;
public class Publisher {
private RabbitTemplate rabbitTemplate;
private String queue;
private String topic;
public Publisher(RabbitTemplate rabbitTemplate, String queue, String topic) {
this.rabbitTemplate = rabbitTemplate;
this.queue = queue;
this.topic = topic;
}
@PostConstruct
public void postMessages() {
rabbitTemplate.convertAndSend(queue, "1 Pepperoni");
rabbitTemplate.convertAndSend(queue, "3 Margarita");
rabbitTemplate.convertAndSend(queue, "1 Ham and Pineapple (yuck)");
rabbitTemplate.convertAndSend(topic, "notification", "New Deal on T-Shirts: 95% off!");
rabbitTemplate.convertAndSend(topic, "notification", "2 for 1 on all Jeans!");
}
}
@@ -0,0 +1,57 @@
package com.baeldung.pubsubmq.server;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class ServerApplication {
private static final String MESSAGE_QUEUE = "pizza-message-queue";
private static final String PUB_SUB_TOPIC = "notification-topic";
private static final String PUB_SUB_EMAIL_QUEUE = "email-queue";
private static final String PUB_SUB_TEXT_QUEUE = "text-queue";
@Bean
public Queue queue() {
return new Queue(MESSAGE_QUEUE);
}
@Bean
public Queue emailQueue() {
return new Queue(PUB_SUB_EMAIL_QUEUE);
}
@Bean
public Queue textQueue() {
return new Queue(PUB_SUB_TEXT_QUEUE);
}
@Bean
public TopicExchange exchange() {
return new TopicExchange(PUB_SUB_TOPIC);
}
@Bean
public Binding emailBinding(Queue emailQueue, TopicExchange exchange) {
return BindingBuilder.bind(emailQueue).to(exchange).with("notification");
}
@Bean
public Binding textBinding(Queue textQueue, TopicExchange exchange) {
return BindingBuilder.bind(textQueue).to(exchange).with("notification");
}
@Bean
public Publisher publisher(RabbitTemplate rabbitTemplate) {
return new Publisher(rabbitTemplate, MESSAGE_QUEUE, PUB_SUB_TOPIC);
}
public static void main(String[] args) {
SpringApplication.run(ServerApplication.class, args);
}
}
@@ -0,0 +1,38 @@
package com.baeldung.setup;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
public class Setup {
private static final String ORDERS_QUEUE_NAME = "orders-queue";
private static final String ORDERS_EXCHANGE_NAME = "orders-direct-exchange";
private static final String ORDERS_ALTERNATE_EXCHANGE_NAME = "orders-alternate-exchange";
private static final String ORDERS_ROUTING_KEY = "orders-routing-key";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
Map<String, Object> exchangeArguments = new HashMap<>();
exchangeArguments.put("alternate-exchange", ORDERS_ALTERNATE_EXCHANGE_NAME);
channel.exchangeDeclare(ORDERS_EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true, false, exchangeArguments);
Map<String, Object> queueArguments = new HashMap<>();
queueArguments.put("x-message-ttl", 60000);
queueArguments.put("x-max-priority", 10);
channel.queueDeclare(ORDERS_QUEUE_NAME, true, false, false, queueArguments);
channel.queueBind(ORDERS_QUEUE_NAME, ORDERS_EXCHANGE_NAME, ORDERS_ROUTING_KEY);
}
}
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
</configuration>
@@ -0,0 +1,2 @@
# Memory configuration for Rabbit
vm_memory_high_watermark.relative = 0.8
@@ -0,0 +1,18 @@
package com.baeldung.benchmark;
import java.util.Arrays;
import org.junit.jupiter.api.Test;
class ConnectionPerChannelPublisherLiveTest {
@Test
void whenConnectionPerChannel_thenRunBenchmark() throws Exception {
// host, workerCount, iterations, payloadSize
Arrays.asList(1,5,10,20,50,100,150).stream()
.forEach(workers -> {
ConnectionPerChannelPublisher.main(new String[]{"192.168.99.100", Integer.toString(workers), "1000", "4096"});
});
}
}
@@ -0,0 +1,18 @@
package com.baeldung.benchmark;
import java.util.Arrays;
import org.junit.jupiter.api.Test;
class SingleConnectionPublisherLiveTest {
@Test
void whenSingleChannel_thenRunBenchmark() throws Exception {
// host, workerCount, iterations, payloadSize
Arrays.asList(1,5,10,20,50,100,150).stream()
.forEach(workers -> {
SingleConnectionPublisher.main(new String[]{"192.168.99.100", Integer.toString(workers), "1000", "4096"});
});
}
}
+10
View File
@@ -0,0 +1,10 @@
## Spring AMQP
This module contains articles about Spring with the AMQP messaging system
## Relevant articles:
- [Messaging with Spring AMQP](https://www.baeldung.com/spring-amqp)
- [RabbitMQ Message Dispatching with Spring AMQP](https://www.baeldung.com/rabbitmq-spring-amqp)
- [Error Handling with Spring AMQP](https://www.baeldung.com/spring-amqp-error-handling)
- [Exponential Backoff With Spring AMQP](https://www.baeldung.com/spring-amqp-exponential-backoff)
+28
View File
@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
<artifactId>spring-amqp</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>spring-amqp</name>
<description>Introduction to Spring-AMQP</description>
<parent>
<groupId>com.baeldung</groupId>
<artifactId>messaging-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
<properties>
<start-class>com.baeldung.springamqp.simple.HelloWorldMessageApp</start-class>
</properties>
</project>
@@ -0,0 +1,52 @@
package com.baeldung.springamqp.broadcast;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class BroadcastConfig {
private static final boolean NON_DURABLE = false;
public final static String FANOUT_QUEUE_1_NAME = "com.baeldung.spring-amqp-simple.fanout.queue1";
public final static String FANOUT_QUEUE_2_NAME = "com.baeldung.spring-amqp-simple.fanout.queue2";
public final static String FANOUT_EXCHANGE_NAME = "com.baeldung.spring-amqp-simple.fanout.exchange";
public final static String TOPIC_QUEUE_1_NAME = "com.baeldung.spring-amqp-simple.topic.queue1";
public final static String TOPIC_QUEUE_2_NAME = "com.baeldung.spring-amqp-simple.topic.queue2";
public final static String TOPIC_EXCHANGE_NAME = "com.baeldung.spring-amqp-simple.topic.exchange";
public static final String BINDING_PATTERN_IMPORTANT = "*.important.*";
public static final String BINDING_PATTERN_ERROR = "#.error";
@Bean
public Declarables topicBindings() {
Queue topicQueue1 = new Queue(TOPIC_QUEUE_1_NAME, NON_DURABLE);
Queue topicQueue2 = new Queue(TOPIC_QUEUE_2_NAME, NON_DURABLE);
TopicExchange topicExchange = new TopicExchange(TOPIC_EXCHANGE_NAME, NON_DURABLE, false);
return new Declarables(topicQueue1, topicQueue2, topicExchange, BindingBuilder
.bind(topicQueue1)
.to(topicExchange)
.with(BINDING_PATTERN_IMPORTANT), BindingBuilder
.bind(topicQueue2)
.to(topicExchange)
.with(BINDING_PATTERN_ERROR));
}
@Bean
public Declarables fanoutBindings() {
Queue fanoutQueue1 = new Queue(FANOUT_QUEUE_1_NAME, NON_DURABLE);
Queue fanoutQueue2 = new Queue(FANOUT_QUEUE_2_NAME, NON_DURABLE);
FanoutExchange fanoutExchange = new FanoutExchange(FANOUT_EXCHANGE_NAME, NON_DURABLE, false);
return new Declarables(fanoutQueue1, fanoutQueue2, fanoutExchange, BindingBuilder
.bind(fanoutQueue1)
.to(fanoutExchange), BindingBuilder
.bind(fanoutQueue2)
.to(fanoutExchange));
}
}
@@ -0,0 +1,59 @@
package com.baeldung.springamqp.broadcast;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import static com.baeldung.springamqp.broadcast.BroadcastConfig.*;
/**
* Simple test application to send messages to rabbitMQ.
*
*<p>To run this particular application with mvn you use the following command:</p>
* {@code
* mvn spring-boot:run -Dstart-class=com.baeldung.springamqp.broadcast.BroadcastMessageApp
* }
*/
@SpringBootApplication
public class BroadcastMessageApp {
private static String ROUTING_KEY_USER_IMPORTANT_WARN = "user.important.warn";
private static String ROUTING_KEY_USER_IMPORTANT_ERROR = "user.important.error";
public static void main(String[] args) {
SpringApplication.run(BroadcastMessageApp.class, args);
}
@Bean
public ApplicationRunner runner(RabbitTemplate rabbitTemplate) {
String message = " payload is broadcast";
return args -> {
rabbitTemplate.convertAndSend(BroadcastConfig.FANOUT_EXCHANGE_NAME, "", "fanout" + message);
rabbitTemplate.convertAndSend(BroadcastConfig.TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_WARN, "topic important warn" + message);
rabbitTemplate.convertAndSend(BroadcastConfig.TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_ERROR, "topic important error" + message);
};
}
@RabbitListener(queues = { FANOUT_QUEUE_1_NAME })
public void receiveMessageFromFanout1(String message) {
System.out.println("Received fanout 1 message: " + message);
}
@RabbitListener(queues = { FANOUT_QUEUE_2_NAME })
public void receiveMessageFromFanout2(String message) {
System.out.println("Received fanout 2 message: " + message);
}
@RabbitListener(queues = { TOPIC_QUEUE_1_NAME })
public void receiveMessageFromTopic1(String message) {
System.out.println("Received topic 1 (" + BINDING_PATTERN_IMPORTANT + ") message: " + message);
}
@RabbitListener(queues = { TOPIC_QUEUE_2_NAME })
public void receiveMessageFromTopic2(String message) {
System.out.println("Received topic 2 (" + BINDING_PATTERN_ERROR + ") message: " + message);
}
}
@@ -0,0 +1,26 @@
package com.baeldung.springamqp.errorhandling;
import com.baeldung.springamqp.errorhandling.producer.MessageProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication
@EnableScheduling
public class ErrorHandlingApp {
@Autowired
MessageProducer messageProducer;
public static void main(String[] args) {
SpringApplication.run(ErrorHandlingApp.class, args);
}
@EventListener(ApplicationReadyEvent.class)
public void doSomethingAfterStartup() {
messageProducer.sendMessage();
}
}
@@ -0,0 +1,55 @@
package com.baeldung.springamqp.errorhandling.configuration;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES_DLQ;
@Configuration
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "dlx-custom")
public class DLXCustomAmqpConfiguration {
public static final String DLX_EXCHANGE_MESSAGES = QUEUE_MESSAGES + ".dlx";
@Bean
Queue messagesQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES)
.withArgument("x-dead-letter-exchange", DLX_EXCHANGE_MESSAGES)
.build();
}
@Bean
FanoutExchange deadLetterExchange() {
return new FanoutExchange(DLX_EXCHANGE_MESSAGES);
}
@Bean
Queue deadLetterQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES_DLQ).build();
}
@Bean
Binding deadLetterBinding() {
return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange());
}
@Bean
DirectExchange messagesExchange() {
return new DirectExchange(EXCHANGE_MESSAGES);
}
@Bean
Binding bindingMessages() {
return BindingBuilder.bind(messagesQueue()).to(messagesExchange()).with(QUEUE_MESSAGES);
}
}
@@ -0,0 +1,72 @@
package com.baeldung.springamqp.errorhandling.configuration;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES_DLQ;
@Configuration
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "parking-lot-dlx")
public class DLXParkingLotAmqpConfiguration {
public static final String DLX_EXCHANGE_MESSAGES = QUEUE_MESSAGES + ".dlx";
public static final String QUEUE_PARKING_LOT = QUEUE_MESSAGES + ".parking-lot";
public static final String EXCHANGE_PARKING_LOT = QUEUE_MESSAGES + "exchange.parking-lot";
@Bean
FanoutExchange parkingLotExchange() {
return new FanoutExchange(EXCHANGE_PARKING_LOT);
}
@Bean
Queue parkingLotQueue() {
return QueueBuilder.durable(QUEUE_PARKING_LOT).build();
}
@Bean
Binding parkingLotBinding() {
return BindingBuilder.bind(parkingLotQueue()).to(parkingLotExchange());
}
@Bean
Queue messagesQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES)
.withArgument("x-dead-letter-exchange", DLX_EXCHANGE_MESSAGES)
.build();
}
@Bean
FanoutExchange deadLetterExchange() {
return new FanoutExchange(DLX_EXCHANGE_MESSAGES);
}
@Bean
Queue deadLetterQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES_DLQ).build();
}
@Bean
Binding deadLetterBinding() {
return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange());
}
@Bean
DirectExchange messagesExchange() {
return new DirectExchange(EXCHANGE_MESSAGES);
}
@Bean
Binding bindingMessages() {
return BindingBuilder.bind(messagesQueue()).to(messagesExchange()).with(QUEUE_MESSAGES);
}
}
@@ -0,0 +1,63 @@
package com.baeldung.springamqp.errorhandling.configuration;
import com.baeldung.springamqp.errorhandling.errorhandler.CustomFatalExceptionStrategy;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.listener.FatalExceptionStrategy;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.ErrorHandler;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES;
@Configuration
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "fatal-error-strategy")
public class FatalExceptionStrategyAmqpConfiguration {
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
ConnectionFactory connectionFactory,
SimpleRabbitListenerContainerFactoryConfigurer configurer) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
configurer.configure(factory, connectionFactory);
factory.setErrorHandler(errorHandler());
return factory;
}
@Bean
public ErrorHandler errorHandler() {
return new ConditionalRejectingErrorHandler(customExceptionStrategy());
}
@Bean
FatalExceptionStrategy customExceptionStrategy() {
return new CustomFatalExceptionStrategy();
}
@Bean
Queue messagesQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES)
.build();
}
@Bean
DirectExchange messagesExchange() {
return new DirectExchange(EXCHANGE_MESSAGES);
}
@Bean
Binding bindingMessages() {
return BindingBuilder.bind(messagesQueue()).to(messagesExchange()).with(QUEUE_MESSAGES);
}
}
@@ -0,0 +1,55 @@
package com.baeldung.springamqp.errorhandling.configuration;
import com.baeldung.springamqp.errorhandling.errorhandler.CustomErrorHandler;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.ErrorHandler;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES;
@Configuration
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "listener-error")
public class ListenerErrorHandlerAmqpConfiguration {
@Bean
public ErrorHandler errorHandler() {
return new CustomErrorHandler();
}
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory,
SimpleRabbitListenerContainerFactoryConfigurer configurer) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
configurer.configure(factory, connectionFactory);
factory.setErrorHandler(errorHandler());
return factory;
}
@Bean
Queue messagesQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES)
.build();
}
@Bean
DirectExchange messagesExchange() {
return new DirectExchange(EXCHANGE_MESSAGES);
}
@Bean
Binding bindingMessages() {
return BindingBuilder.bind(messagesQueue()).to(messagesExchange()).with(QUEUE_MESSAGES);
}
}
@@ -0,0 +1,55 @@
package com.baeldung.springamqp.errorhandling.configuration;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES_DLQ;
@Configuration
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "routing-dlq")
public class RoutingKeyDLQAmqpConfiguration {
public static final String DLX_EXCHANGE_MESSAGES = QUEUE_MESSAGES + ".dlx";
@Bean
Queue messagesQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES)
.withArgument("x-dead-letter-exchange", DLX_EXCHANGE_MESSAGES)
.build();
}
@Bean
FanoutExchange deadLetterExchange() {
return new FanoutExchange(DLX_EXCHANGE_MESSAGES);
}
@Bean
Queue deadLetterQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES_DLQ).build();
}
@Bean
Binding deadLetterBinding() {
return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange());
}
@Bean
DirectExchange messagesExchange() {
return new DirectExchange(EXCHANGE_MESSAGES);
}
@Bean
Binding bindingMessages() {
return BindingBuilder.bind(messagesQueue()).to(messagesExchange()).with(QUEUE_MESSAGES);
}
}
@@ -0,0 +1,43 @@
package com.baeldung.springamqp.errorhandling.configuration;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "simple-dlq")
public class SimpleDLQAmqpConfiguration {
public static final String QUEUE_MESSAGES = "baeldung-messages-queue";
public static final String QUEUE_MESSAGES_DLQ = QUEUE_MESSAGES + ".dlq";
public static final String EXCHANGE_MESSAGES = "baeldung-messages-exchange";
@Bean
Queue messagesQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES)
.withArgument("x-dead-letter-exchange", "")
.withArgument("x-dead-letter-routing-key", QUEUE_MESSAGES_DLQ)
.build();
}
@Bean
Queue deadLetterQueue() {
return QueueBuilder.durable(QUEUE_MESSAGES_DLQ).build();
}
@Bean
DirectExchange messagesExchange() {
return new DirectExchange(EXCHANGE_MESSAGES);
}
@Bean
Binding bindingMessages() {
return BindingBuilder.bind(messagesQueue()).to(messagesExchange()).with(QUEUE_MESSAGES);
}
}
@@ -0,0 +1,35 @@
package com.baeldung.springamqp.errorhandling.consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES_DLQ;
import static com.baeldung.springamqp.errorhandling.consumer.MessagesConsumer.HEADER_X_RETRIES_COUNT;
public class DLQCustomAmqpContainer {
private static final Logger log = LoggerFactory.getLogger(DLQCustomAmqpContainer.class);
private final RabbitTemplate rabbitTemplate;
public static final int MAX_RETRIES_COUNT = 2;
public DLQCustomAmqpContainer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
@RabbitListener(queues = QUEUE_MESSAGES_DLQ)
public void processFailedMessagesRetryHeaders(Message failedMessage) {
Integer retriesCnt = (Integer) failedMessage.getMessageProperties().getHeaders().get(HEADER_X_RETRIES_COUNT);
if (retriesCnt == null)
retriesCnt = 1;
if (retriesCnt > MAX_RETRIES_COUNT) {
log.info("Discarding message");
return;
}
log.info("Retrying message for the {} time", retriesCnt);
failedMessage.getMessageProperties().getHeaders().put(HEADER_X_RETRIES_COUNT, ++retriesCnt);
rabbitTemplate.send(EXCHANGE_MESSAGES, failedMessage.getMessageProperties().getReceivedRoutingKey(), failedMessage);
}
}
@@ -0,0 +1,63 @@
package com.baeldung.springamqp.errorhandling.consumer;
import com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration;
import com.baeldung.springamqp.errorhandling.errorhandler.BusinessException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MessagesConsumer {
public static final String HEADER_X_RETRIES_COUNT = "x-retries-count";
private static final Logger log = LoggerFactory.getLogger(MessagesConsumer.class);
private final RabbitTemplate rabbitTemplate;
public MessagesConsumer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
@RabbitListener(queues = SimpleDLQAmqpConfiguration.QUEUE_MESSAGES)
public void receiveMessage(Message message) throws BusinessException {
log.info("Received message: {}", message.toString());
throw new BusinessException();
}
@Bean
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "simple-dlq")
public SimpleDLQAmqpContainer simpleAmqpContainer() {
return new SimpleDLQAmqpContainer(rabbitTemplate);
}
@Bean
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "routing-dlq")
public RoutingDLQAmqpContainer routingDLQAmqpContainer() {
return new RoutingDLQAmqpContainer(rabbitTemplate);
}
@Bean
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "dlx-custom")
public DLQCustomAmqpContainer dlqAmqpContainer() {
return new DLQCustomAmqpContainer(rabbitTemplate);
}
@Bean
@ConditionalOnProperty(
value = "amqp.configuration.current",
havingValue = "parking-lot-dlx")
public ParkingLotDLQAmqpContainer parkingLotDLQAmqpContainer() {
return new ParkingLotDLQAmqpContainer(rabbitTemplate);
}
}
@@ -0,0 +1,43 @@
package com.baeldung.springamqp.errorhandling.consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import static com.baeldung.springamqp.errorhandling.configuration.DLXParkingLotAmqpConfiguration.EXCHANGE_PARKING_LOT;
import static com.baeldung.springamqp.errorhandling.configuration.DLXParkingLotAmqpConfiguration.QUEUE_PARKING_LOT;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES_DLQ;
import static com.baeldung.springamqp.errorhandling.consumer.MessagesConsumer.HEADER_X_RETRIES_COUNT;
public class ParkingLotDLQAmqpContainer {
private static final Logger log = LoggerFactory.getLogger(ParkingLotDLQAmqpContainer.class);
private final RabbitTemplate rabbitTemplate;
public static final int MAX_RETRIES_COUNT = 2;
public ParkingLotDLQAmqpContainer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
@RabbitListener(queues = QUEUE_MESSAGES_DLQ)
public void processFailedMessagesRetryWithParkingLot(Message failedMessage) {
Integer retriesCnt = (Integer) failedMessage.getMessageProperties().getHeaders().get(HEADER_X_RETRIES_COUNT);
if (retriesCnt == null)
retriesCnt = 1;
if (retriesCnt > MAX_RETRIES_COUNT) {
log.info("Sending message to the parking lot queue");
rabbitTemplate.send(EXCHANGE_PARKING_LOT, failedMessage.getMessageProperties().getReceivedRoutingKey(), failedMessage);
return;
}
log.info("Retrying message for the {} time", retriesCnt);
failedMessage.getMessageProperties().getHeaders().put(HEADER_X_RETRIES_COUNT, ++retriesCnt);
rabbitTemplate.send(EXCHANGE_MESSAGES, failedMessage.getMessageProperties().getReceivedRoutingKey(), failedMessage);
}
@RabbitListener(queues = QUEUE_PARKING_LOT)
public void processParkingLotQueue(Message failedMessage) {
log.info("Received message in parking lot queue {}", failedMessage.toString());
}
}
@@ -0,0 +1,33 @@
package com.baeldung.springamqp.errorhandling.consumer;
import com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration;
import com.baeldung.springamqp.errorhandling.errorhandler.BusinessException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES_DLQ;
public class RoutingDLQAmqpContainer {
private static final Logger log = LoggerFactory.getLogger(RoutingDLQAmqpContainer.class);
private final RabbitTemplate rabbitTemplate;
public RoutingDLQAmqpContainer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
@RabbitListener(queues = SimpleDLQAmqpConfiguration.QUEUE_MESSAGES)
public void receiveMessage(Message message) throws BusinessException {
log.info("Received message: {}", message.toString());
throw new BusinessException();
}
@RabbitListener(queues = QUEUE_MESSAGES_DLQ)
public void processFailedMessagesRequeue(Message failedMessage) {
log.info("Received failed message, requeueing: {}", failedMessage.toString());
rabbitTemplate.send(EXCHANGE_MESSAGES, failedMessage.getMessageProperties().getReceivedRoutingKey(), failedMessage);
}
}
@@ -0,0 +1,31 @@
package com.baeldung.springamqp.errorhandling.consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES;
import static com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration.QUEUE_MESSAGES_DLQ;
public class SimpleDLQAmqpContainer {
private static final Logger log = LoggerFactory.getLogger(SimpleDLQAmqpContainer.class);
private final RabbitTemplate rabbitTemplate;
public SimpleDLQAmqpContainer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
@RabbitListener(queues = QUEUE_MESSAGES_DLQ)
public void processFailedMessages(Message message) {
log.info("Received failed message: {}", message.toString());
}
@RabbitListener(queues = QUEUE_MESSAGES_DLQ)
public void processFailedMessagesRequeue(Message failedMessage) {
log.info("Received failed message, requeueing: {}", failedMessage.toString());
log.info("Received failed message, requeueing: {}", failedMessage.getMessageProperties().getReceivedRoutingKey());
rabbitTemplate.send(EXCHANGE_MESSAGES, failedMessage.getMessageProperties().getReceivedRoutingKey(), failedMessage);
}
}
@@ -0,0 +1,4 @@
package com.baeldung.springamqp.errorhandling.errorhandler;
public class BusinessException extends Exception {
}
@@ -0,0 +1,14 @@
package com.baeldung.springamqp.errorhandling.errorhandler;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.util.ErrorHandler;
public class CustomErrorHandler implements ErrorHandler {
@Override
public void handleError(Throwable t) {
if (!(t.getCause() instanceof BusinessException)) {
throw new AmqpRejectAndDontRequeueException("Error Handler converted exception to fatal", t);
}
}
}
@@ -0,0 +1,10 @@
package com.baeldung.springamqp.errorhandling.errorhandler;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
public class CustomFatalExceptionStrategy extends ConditionalRejectingErrorHandler.DefaultExceptionStrategy {
@Override
public boolean isFatal(Throwable t) {
return !(t.getCause() instanceof BusinessException);
}
}
@@ -0,0 +1,24 @@
package com.baeldung.springamqp.errorhandling.producer;
import com.baeldung.springamqp.errorhandling.configuration.SimpleDLQAmqpConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
@Service
public class MessageProducer {
private static final Logger log = LoggerFactory.getLogger(MessageProducer.class);
private int messageNumber = 0;
private final RabbitTemplate rabbitTemplate;
public MessageProducer(final RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
public void sendMessage() {
log.info("Sending message...");
rabbitTemplate.convertAndSend(SimpleDLQAmqpConfiguration.EXCHANGE_MESSAGES, SimpleDLQAmqpConfiguration.QUEUE_MESSAGES, "Some message id:" + messageNumber++);
}
}
@@ -0,0 +1,11 @@
package com.baeldung.springamqp.exponentialbackoff;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ExponentialBackoffApp {
public static void main(String[] args) {
SpringApplication.run(ExponentialBackoffApp.class, args);
}
}
@@ -0,0 +1,21 @@
package com.baeldung.springamqp.exponentialbackoff;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
public class ObservableRejectAndDontRequeueRecoverer extends RejectAndDontRequeueRecoverer {
private Runnable observer;
@Override
public void recover(Message message, Throwable cause) {
if(observer != null) {
observer.run();
}
super.recover(message, cause);
}
void setObserver(Runnable observer){
this.observer = observer;
}
}
@@ -0,0 +1,162 @@
package com.baeldung.springamqp.exponentialbackoff;
import org.aopalliance.aop.Advice;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import com.rabbitmq.client.Channel;
@EnableRabbit
@Configuration
public class RabbitConfiguration {
private static Logger logger = LoggerFactory.getLogger(RabbitConfiguration.class);
@Bean
public ConnectionFactory connectionFactory() {
return new CachingConnectionFactory("localhost");
}
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(connectionFactory());
}
@Bean
public RabbitTemplate rabbitTemplate() {
return new RabbitTemplate(connectionFactory());
}
@Bean
public Queue blockingQueue() {
return QueueBuilder.nonDurable("blocking-queue")
.build();
}
@Bean
public Queue nonBlockingQueue() {
return QueueBuilder.nonDurable("non-blocking-queue")
.build();
}
@Bean
public Queue retryWaitEndedQueue() {
return QueueBuilder.nonDurable("retry-wait-ended-queue")
.build();
}
@Bean
public Queue retryQueue1() {
return QueueBuilder.nonDurable("retry-queue-1")
.deadLetterExchange("")
.deadLetterRoutingKey("retry-wait-ended-queue")
.build();
}
@Bean
public Queue retryQueue2() {
return QueueBuilder.nonDurable("retry-queue-2")
.deadLetterExchange("")
.deadLetterRoutingKey("retry-wait-ended-queue")
.build();
}
@Bean
public Queue retryQueue3() {
return QueueBuilder.nonDurable("retry-queue-3")
.deadLetterExchange("")
.deadLetterRoutingKey("retry-wait-ended-queue")
.build();
}
@Bean
public RetryQueues retryQueues() {
return new RetryQueues(1000, 3.0, 10000, retryQueue1(), retryQueue2(), retryQueue3());
}
@Bean
public ObservableRejectAndDontRequeueRecoverer observableRecoverer() {
return new ObservableRejectAndDontRequeueRecoverer();
}
@Bean
public RetryOperationsInterceptor retryInterceptor() {
return RetryInterceptorBuilder.stateless()
.backOffOptions(1000, 3.0, 10000)
.maxAttempts(5)
.recoverer(observableRecoverer())
.build();
}
@Bean
public RetryQueuesInterceptor retryQueuesInterceptor(RabbitTemplate rabbitTemplate, RetryQueues retryQueues) {
return new RetryQueuesInterceptor(rabbitTemplate, retryQueues);
}
@Bean
public SimpleRabbitListenerContainerFactory defaultContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
return factory;
}
@Bean
public SimpleRabbitListenerContainerFactory retryContainerFactory(ConnectionFactory connectionFactory, RetryOperationsInterceptor retryInterceptor) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
Advice[] adviceChain = { retryInterceptor };
factory.setAdviceChain(adviceChain);
return factory;
}
@Bean
public SimpleRabbitListenerContainerFactory retryQueuesContainerFactory(ConnectionFactory connectionFactory, RetryQueuesInterceptor retryInterceptor) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
Advice[] adviceChain = { retryInterceptor };
factory.setAdviceChain(adviceChain);
return factory;
}
@RabbitListener(queues = "blocking-queue", containerFactory = "retryContainerFactory")
public void consumeBlocking(String payload) throws Exception {
logger.info("Processing message from blocking-queue: {}", payload);
throw new Exception("exception occured!");
}
@RabbitListener(queues = "non-blocking-queue", containerFactory = "retryQueuesContainerFactory", ackMode = "MANUAL")
public void consumeNonBlocking(String payload) throws Exception {
logger.info("Processing message from non-blocking-queue: {}", payload);
throw new Exception("Error occured!");
}
@RabbitListener(queues = "retry-wait-ended-queue", containerFactory = "defaultContainerFactory")
public void consumeRetryWaitEndedMessage(String payload, Message message, Channel channel) throws Exception {
MessageProperties props = message.getMessageProperties();
rabbitTemplate().convertAndSend(props.getHeader("x-original-exchange"), props.getHeader("x-original-routing-key"), message);
}
}
@@ -0,0 +1,34 @@
package com.baeldung.springamqp.exponentialbackoff;
import org.springframework.amqp.core.Queue;
public class RetryQueues {
private Queue[] queues;
private long initialInterval;
private double factor;
private long maxWait;
public RetryQueues(long initialInterval, double factor, long maxWait, Queue... queues) {
this.queues = queues;
this.initialInterval = initialInterval;
this.factor = factor;
this.maxWait = maxWait;
}
public boolean retriesExhausted(int retry) {
return retry >= queues.length;
}
public String getQueueName(int retry) {
return queues[retry].getName();
}
public long getTimeToWait(int retry) {
double time = initialInterval * Math.pow(factor, (double) retry);
if (time > maxWait) {
return maxWait;
}
return (long) time;
}
}
@@ -0,0 +1,109 @@
package com.baeldung.springamqp.exponentialbackoff;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import com.rabbitmq.client.Channel;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
public class RetryQueuesInterceptor implements MethodInterceptor {
private RabbitTemplate rabbitTemplate;
private RetryQueues retryQueues;
private Runnable observer;
public RetryQueuesInterceptor(RabbitTemplate rabbitTemplate, RetryQueues retryQueues) {
this.rabbitTemplate = rabbitTemplate;
this.retryQueues = retryQueues;
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
return tryConsume(invocation, this::ack, (mac, e) -> {
try {
int retryCount = tryGetRetryCountOrFail(mac, e);
sendToNextRetryQueue(mac, retryCount);
} catch (Throwable t) {
if (observer != null) {
observer.run();
}
throw new RuntimeException(t);
}
});
}
void setObserver(Runnable observer) {
this.observer = observer;
}
private Object tryConsume(MethodInvocation invocation, Consumer<MessageAndChannel> successHandler, BiConsumer<MessageAndChannel, Throwable> errorHandler) throws Throwable {
MessageAndChannel mac = new MessageAndChannel((Message) invocation.getArguments()[1], (Channel) invocation.getArguments()[0]);
Object ret = null;
try {
ret = invocation.proceed();
successHandler.accept(mac);
} catch (Throwable e) {
errorHandler.accept(mac, e);
}
return ret;
}
private void ack(MessageAndChannel mac) {
try {
mac.channel.basicAck(mac.message.getMessageProperties()
.getDeliveryTag(), false);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
private int tryGetRetryCountOrFail(MessageAndChannel mac, Throwable originalError) throws Throwable {
MessageProperties props = mac.message.getMessageProperties();
String xRetriedCountHeader = (String) props.getHeader("x-retried-count");
final int xRetriedCount = xRetriedCountHeader == null ? 0 : Integer.valueOf(xRetriedCountHeader);
if (retryQueues.retriesExhausted(xRetriedCount)) {
mac.channel.basicReject(props.getDeliveryTag(), false);
throw originalError;
}
return xRetriedCount;
}
private void sendToNextRetryQueue(MessageAndChannel mac, int retryCount) throws Exception {
String retryQueueName = retryQueues.getQueueName(retryCount);
rabbitTemplate.convertAndSend(retryQueueName, mac.message, m -> {
MessageProperties props = m.getMessageProperties();
props.setExpiration(String.valueOf(retryQueues.getTimeToWait(retryCount)));
props.setHeader("x-retried-count", String.valueOf(retryCount + 1));
props.setHeader("x-original-exchange", props.getReceivedExchange());
props.setHeader("x-original-routing-key", props.getReceivedRoutingKey());
return m;
});
mac.channel.basicReject(mac.message.getMessageProperties()
.getDeliveryTag(), false);
}
private class MessageAndChannel {
private Message message;
private Channel channel;
private MessageAndChannel(Message message, Channel channel) {
this.message = message;
this.channel = channel;
}
}
}
@@ -0,0 +1,38 @@
package com.baeldung.springamqp.simple;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class HelloWorldMessageApp {
private static final boolean NON_DURABLE = false;
private static final String MY_QUEUE_NAME = "myQueue";
public static void main(String[] args) {
SpringApplication.run(HelloWorldMessageApp.class, args);
}
@Bean
public ApplicationRunner runner(RabbitTemplate template) {
return args -> {
template.convertAndSend("myQueue", "Hello, world!");
};
}
@Bean
public Queue myQueue() {
return new Queue(MY_QUEUE_NAME, NON_DURABLE);
}
@RabbitListener(queues = MY_QUEUE_NAME)
public void listen(String in) {
System.out.println("Message read from myQueue : " + in);
}
}
@@ -0,0 +1,3 @@
spring.rabbitmq.listener.simple.default-requeue-rejected=false
spring.main.allow-bean-definition-overriding=true
amqp.configuration.current=parking-lot-dlx
@@ -0,0 +1,58 @@
package com.baeldung.springamqp.exponentialbackoff;
import java.util.concurrent.CountDownLatch;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* This live test requires:
*
* - A running RabbitMQ instance on localhost (e.g. docker run -p 5672:5672 -p 15672:15672 --name rabbit rabbitmq:3-management)
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = { RabbitConfiguration.class })
public class ExponentialBackoffLiveTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private ObservableRejectAndDontRequeueRecoverer observableRecoverer;
@Autowired
private RetryQueuesInterceptor retryQueues;
@Test
public void whenSendToBlockingQueue_thenAllMessagesProcessed() throws Exception {
int nb = 2;
CountDownLatch latch = new CountDownLatch(nb);
observableRecoverer.setObserver(() -> latch.countDown());
for (int i = 1; i <= nb; i++) {
rabbitTemplate.convertAndSend("blocking-queue", "blocking message " + i);
}
latch.await();
}
@Test
public void whenSendToNonBlockingQueue_thenAllMessageProcessed() throws Exception {
int nb = 2;
CountDownLatch latch = new CountDownLatch(nb);
retryQueues.setObserver(() -> latch.countDown());
for (int i = 1; i <= nb; i++) {
rabbitTemplate.convertAndSend("non-blocking-queue", "non-blocking message " + i);
}
latch.await();
}
}
@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<include
resource="org/springframework/boot/logging/logback/base.xml" />
<logger name="org.springframework" level="INFO" />
</configuration>
@@ -0,0 +1 @@
/src/test/destination-folder/*
@@ -0,0 +1,27 @@
## Spring Apache Camel
This module contains articles about Spring with Apache Camel
### Relevant Articles
- [Apache Camel](http://camel.apache.org/)
- [Enterprise Integration Patterns](http://www.enterpriseintegrationpatterns.com/patterns/messaging/toc.html)
- [Introduction To Apache Camel](http://www.baeldung.com/apache-camel-intro)
- [Integration Patterns With Apache Camel](http://www.baeldung.com/camel-integration-patterns)
- [Using Apache Camel with Spring](http://www.baeldung.com/spring-apache-camel-tutorial)
- [Unmarshalling a JSON Array Using camel-jackson](https://www.baeldung.com/java-camel-jackson-json-array)
### Framework Versions:
- Spring 4.2.4
- Apache Camel 2.16.1
### Build and Run Application
To build this application execute:
`mvn clean install`
To run this application you can either run our main class App from your IDE or you can execute following maven command:
`mvn exec:java -Dexec.mainClass="com.baeldung.camel.main.App"`
@@ -0,0 +1,68 @@
<?xml version="1.0" encoding="UTF-8"?>
<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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.baeldung.apache.camel</groupId>
<artifactId>spring-apache-camel</artifactId>
<name>spring-apache-camel</name>
<packaging>jar</packaging>
<url>http://maven.apache.org</url>
<parent>
<groupId>com.baeldung</groupId>
<artifactId>messaging-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-core</artifactId>
<version>${env.camel.version}</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-spring</artifactId>
<version>${env.camel.version}</version>
<exclusions>
<exclusion>
<artifactId>commons-logging</artifactId>
<groupId>commons-logging</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-stream</artifactId>
<version>${env.camel.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${env.spring.version}</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-spring-javaconfig</artifactId>
<version>${env.camel.version}</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-jackson</artifactId>
<version>${env.camel.version}</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-test</artifactId>
<version>${env.camel.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<properties>
<env.camel.version>2.18.1</env.camel.version>
<env.spring.version>4.3.4.RELEASE</env.spring.version>
</properties>
</project>
@@ -0,0 +1,16 @@
package com.baeldung.camel.file;
import org.apache.camel.builder.RouteBuilder;
public class ContentBasedFileRouter extends RouteBuilder {
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER_TXT = "src/test/destination-folder-txt";
private static final String DESTINATION_FOLDER_OTHER = "src/test/destination-folder-other";
@Override
public void configure() throws Exception {
from("file://" + SOURCE_FOLDER + "?delete=true").choice().when(simple("${file:ext} == 'txt'")).to("file://" + DESTINATION_FOLDER_TXT).otherwise().to("file://" + DESTINATION_FOLDER_OTHER);
}
}
@@ -0,0 +1,18 @@
package com.baeldung.camel.file;
import org.apache.camel.LoggingLevel;
import org.apache.camel.builder.RouteBuilder;
public class DeadLetterChannelFileRouter extends RouteBuilder {
private static final String SOURCE_FOLDER = "src/test/source-folder";
@Override
public void configure() throws Exception {
errorHandler(deadLetterChannel("log:dead?level=ERROR").maximumRedeliveries(3)
.redeliveryDelay(1000).retryAttemptedLogLevel(LoggingLevel.ERROR));
from("file://" + SOURCE_FOLDER + "?delete=true").process((exchange) -> {
throw new IllegalArgumentException("Exception thrown!");
});
}
}
@@ -0,0 +1,20 @@
package com.baeldung.camel.file;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
public class FileProcessor implements Processor {
public void process(Exchange exchange) throws Exception {
String originalFileName = (String) exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
Date date = new Date();
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH-mm-ss");
String changedFileName = dateFormat.format(date) + originalFileName;
exchange.getIn().setHeader(Exchange.FILE_NAME, changedFileName);
}
}
@@ -0,0 +1,15 @@
package com.baeldung.camel.file;
import org.apache.camel.builder.RouteBuilder;
public class FileRouter extends RouteBuilder {
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER = "src/test/destination-folder";
@Override
public void configure() throws Exception {
from("file://" + SOURCE_FOLDER + "?delete=true").process(new FileProcessor()).to("file://" + DESTINATION_FOLDER);
}
}
@@ -0,0 +1,14 @@
package com.baeldung.camel.file;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
public class MessageTranslatorFileRouter extends RouteBuilder {
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER = "src/test/destination-folder";
@Override
public void configure() throws Exception {
from("file://" + SOURCE_FOLDER + "?delete=true").transform(body().append(header(Exchange.FILE_NAME))).to("file://" + DESTINATION_FOLDER);
}
}
@@ -0,0 +1,18 @@
package com.baeldung.camel.file;
import org.apache.camel.builder.RouteBuilder;
public class MulticastFileRouter extends RouteBuilder {
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER_WORLD = "src/test/destination-folder-world";
private static final String DESTINATION_FOLDER_HELLO = "src/test/destination-folder-hello";
@Override
public void configure() throws Exception {
from("file://" + SOURCE_FOLDER + "?delete=true").multicast().to("direct:append", "direct:prepend").end();
from("direct:append").transform(body().append("World")).to("file://" + DESTINATION_FOLDER_WORLD);
from("direct:prepend").transform(body().prepend("Hello")).to("file://" + DESTINATION_FOLDER_HELLO);
}
}
@@ -0,0 +1,15 @@
package com.baeldung.camel.file;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
public class SplitterFileRouter extends RouteBuilder {
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER = "src/test/destination-folder";
@Override
public void configure() throws Exception {
from("file://" + SOURCE_FOLDER + "?delete=true").split(body().convertToString().tokenize("\n")).setHeader(Exchange.FILE_NAME, body()).to("file://" + DESTINATION_FOLDER);
}
}
@@ -0,0 +1,26 @@
package com.baeldung.camel.file.cfg;
import java.util.Arrays;
import java.util.List;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.spring.javaconfig.CamelConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.baeldung.camel.file.ContentBasedFileRouter;
@Configuration
public class ContentBasedFileRouterConfig extends CamelConfiguration {
@Bean
ContentBasedFileRouter getContentBasedFileRouter() {
return new ContentBasedFileRouter();
}
@Override
public List<RouteBuilder> routes() {
return Arrays.asList(getContentBasedFileRouter());
}
}
@@ -0,0 +1,24 @@
package com.baeldung.camel.jackson;
public class Fruit {
private String name;
private int id;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
}
@@ -0,0 +1,17 @@
package com.baeldung.camel.jackson;
import java.util.List;
public class FruitList {
private List<Fruit> fruits;
public List<Fruit> getFruits() {
return fruits;
}
public void setFruits(List<Fruit> fruits) {
this.fruits = fruits;
}
}
@@ -0,0 +1,12 @@
package com.baeldung.camel.main;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class App {
public static void main(final String[] args) throws Exception {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("camel-context.xml");
// Keep main thread alive for some time to let application finish processing the input files.
Thread.sleep(5000);
applicationContext.close();
}
}
@@ -0,0 +1,14 @@
package com.baeldung.camel.processor;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
public class FileProcessor implements Processor {
public void process(Exchange exchange) throws Exception {
String originalFileContent = exchange.getIn().getBody(String.class);
String upperCaseFileContent = originalFileContent.toUpperCase();
exchange.getIn().setBody(upperCaseFileContent);
}
}
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="contentBasedFileRouter" class="com.baeldung.camel.file.ContentBasedFileRouter" />
<camelContext xmlns="http://camel.apache.org/schema/spring">
<routeBuilder ref="contentBasedFileRouter" />
</camelContext>
</beans>
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="deadLetterChannelFileRouter" class="com.baeldung.camel.file.DeadLetterChannelFileRouter" />
<camelContext xmlns="http://camel.apache.org/schema/spring">
<routeBuilder ref="deadLetterChannelFileRouter" />
</camelContext>
</beans>
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="messageTranslatorFileRouter" class="com.baeldung.camel.file.MessageTranslatorFileRouter" />
<camelContext xmlns="http://camel.apache.org/schema/spring">
<routeBuilder ref="messageTranslatorFileRouter" />
</camelContext>
</beans>
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="multicastFileRouter" class="com.baeldung.camel.file.MulticastFileRouter" />
<camelContext xmlns="http://camel.apache.org/schema/spring">
<routeBuilder ref="multicastFileRouter" />
</camelContext>
</beans>
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="splitterFileRouter" class="com.baeldung.camel.file.SplitterFileRouter" />
<camelContext xmlns="http://camel.apache.org/schema/spring">
<routeBuilder ref="splitterFileRouter" />
</camelContext>
</beans>
@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="fileRouter" class="com.baeldung.camel.file.FileRouter" />
<bean id="fileProcessor" class="com.baeldung.camel.file.FileProcessor" />
<camelContext xmlns="http://camel.apache.org/schema/spring">
<routeBuilder ref="fileRouter" />
</camelContext>
</beans>
@@ -0,0 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<camelContext xmlns="http://camel.apache.org/schema/spring">
<route customId="true" id="route1">
<!-- Read files from input directory -->
<from uri="file://src/test/data/input" />
<!-- Transform content to UpperCase -->
<process ref="myFileProcessor" />
<!-- Write converted file content -->
<to uri="file://src/test/data/outputUpperCase" />
<!-- Transform content to LowerCase -->
<transform>
<simple>${body.toLowerCase()}</simple>
</transform>
<!-- Write converted file content -->
<to uri="file://src/test/data/outputLowerCase" />
<!-- Display process completion message on console -->
<transform>
<simple>.......... File content conversion completed ..........</simple>
</transform>
<to uri="stream:out" />
</route>
</camelContext>
<bean id="myFileProcessor" class="com.baeldung.camel.processor.FileProcessor" />
</beans>
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
</configuration>
@@ -0,0 +1 @@
This is data that will be processed by a Camel route!
@@ -0,0 +1,71 @@
package com.apache.camel.file.processor;
import java.io.File;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import com.baeldung.camel.file.cfg.ContentBasedFileRouterConfig;
@RunWith(JUnit4.class)
public class ContentBasedFileRouterIntegrationTest {
private static final long DURATION_MILIS = 10000;
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER_TXT = "src/test/destination-folder-txt";
private static final String DESTINATION_FOLDER_OTHER = "src/test/destination-folder-other";
@Before
public void setUp() throws Exception {
File sourceFolder = new File(SOURCE_FOLDER);
File destinationFolderTxt = new File(DESTINATION_FOLDER_TXT);
File destinationFolderOther = new File(DESTINATION_FOLDER_OTHER);
cleanFolder(sourceFolder);
cleanFolder(destinationFolderTxt);
cleanFolder(destinationFolderOther);
sourceFolder.mkdirs();
File file1 = new File(SOURCE_FOLDER + "/File1.txt");
File file2 = new File(SOURCE_FOLDER + "/File2.csv");
file1.createNewFile();
file2.createNewFile();
}
private void cleanFolder(File folder) {
File[] files = folder.listFiles();
if (files != null) {
for (File file : files) {
if (file.isFile()) {
file.delete();
}
}
}
}
@Test
@Ignore
public void routeWithXMLConfigTest() throws InterruptedException {
AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext(
"camel-context-ContentBasedFileRouterTest.xml");
Thread.sleep(DURATION_MILIS);
applicationContext.close();
}
@Test
@Ignore
public void routeWithJavaConfigTest() throws InterruptedException {
AbstractApplicationContext applicationContext = new AnnotationConfigApplicationContext(
ContentBasedFileRouterConfig.class);
Thread.sleep(DURATION_MILIS);
applicationContext.close();
}
}
@@ -0,0 +1,43 @@
package com.apache.camel.file.processor;
import java.io.File;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class DeadLetterChannelFileRouterIntegrationTest {
private static final long DURATION_MILIS = 10000;
private static final String SOURCE_FOLDER = "src/test/source-folder";
@Before
public void setUp() throws Exception {
File sourceFolder = new File(SOURCE_FOLDER);
cleanFolder(sourceFolder);
sourceFolder.mkdirs();
File file = new File(SOURCE_FOLDER + "/File.txt");
file.createNewFile();
}
private void cleanFolder(File folder) {
File[] files = folder.listFiles();
if (files != null) {
for (File file : files) {
if (file.isFile()) {
file.delete();
}
}
}
}
@Test
public void routeTest() throws InterruptedException {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("camel-context-DeadLetterChannelFileRouter.xml");
Thread.sleep(DURATION_MILIS);
applicationContext.close();
}
}
@@ -0,0 +1,68 @@
package com.apache.camel.file.processor;
import java.io.File;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import com.baeldung.camel.file.FileProcessor;
public class FileProcessorIntegrationTest {
private static final long DURATION_MILIS = 10000;
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER = "src/test/destination-folder";
@Before
public void setUp() throws Exception {
File sourceFolder = new File(SOURCE_FOLDER);
File destinationFolder = new File(DESTINATION_FOLDER);
cleanFolder(sourceFolder);
cleanFolder(destinationFolder);
sourceFolder.mkdirs();
File file1 = new File(SOURCE_FOLDER + "/File1.txt");
File file2 = new File(SOURCE_FOLDER + "/File2.txt");
file1.createNewFile();
file2.createNewFile();
}
private void cleanFolder(File folder) {
File[] files = folder.listFiles();
if (files != null) {
for (File file : files) {
if (file.isFile()) {
file.delete();
}
}
}
}
@Test
public void moveFolderContentJavaDSLTest() throws Exception {
final CamelContext camelContext = new DefaultCamelContext();
camelContext.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
from("file://" + SOURCE_FOLDER + "?delete=true").process(new FileProcessor()).to("file://" + DESTINATION_FOLDER);
}
});
camelContext.start();
Thread.sleep(DURATION_MILIS);
camelContext.stop();
}
@Test
public void moveFolderContentSpringDSLTest() throws InterruptedException {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("camel-context-test.xml");
Thread.sleep(DURATION_MILIS);
applicationContext.close();
}
}
@@ -0,0 +1,50 @@
package com.apache.camel.file.processor;
import java.io.File;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class MessageTranslatorFileRouterIntegrationTest {
private static final long DURATION_MILIS = 10000;
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER = "src/test/destination-folder";
@Before
public void setUp() throws Exception {
File sourceFolder = new File(SOURCE_FOLDER);
File destinationFolder = new File(DESTINATION_FOLDER);
cleanFolder(sourceFolder);
cleanFolder(destinationFolder);
sourceFolder.mkdirs();
File file1 = new File(SOURCE_FOLDER + "/File1.txt");
File file2 = new File(SOURCE_FOLDER + "/File2.txt");
file1.createNewFile();
file2.createNewFile();
}
private void cleanFolder(File folder) {
File[] files = folder.listFiles();
if (files != null) {
for (File file : files) {
if (file.isFile()) {
file.delete();
}
}
}
}
@Test
@Ignore
public void routeTest() throws InterruptedException {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("camel-context-MessageTranslatorFileRouterTest.xml");
Thread.sleep(DURATION_MILIS);
applicationContext.close();
}
}
@@ -0,0 +1,53 @@
package com.apache.camel.file.processor;
import java.io.File;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class MulticastFileRouterIntegrationTest {
private static final long DURATION_MILIS = 10000;
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER_WORLD = "src/test/destination-folder-world";
private static final String DESTINATION_FOLDER_HELLO = "src/test/destination-folder-hello";
@Before
public void setUp() throws Exception {
File sourceFolder = new File(SOURCE_FOLDER);
File destinationFolderWorld = new File(DESTINATION_FOLDER_WORLD);
File destinationFolderHello = new File(DESTINATION_FOLDER_HELLO);
cleanFolder(sourceFolder);
cleanFolder(destinationFolderWorld);
cleanFolder(destinationFolderHello);
sourceFolder.mkdirs();
File file1 = new File(SOURCE_FOLDER + "/File1.txt");
File file2 = new File(SOURCE_FOLDER + "/File2.txt");
file1.createNewFile();
file2.createNewFile();
}
private void cleanFolder(File folder) {
File[] files = folder.listFiles();
if (files != null) {
for (File file : files) {
if (file.isFile()) {
file.delete();
}
}
}
}
@Test
@Ignore
public void routeTest() throws InterruptedException {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("camel-context-MulticastFileRouterTest.xml");
Thread.sleep(DURATION_MILIS);
applicationContext.close();
}
}
@@ -0,0 +1,52 @@
package com.apache.camel.file.processor;
import java.io.File;
import java.io.FileWriter;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class SplitterFileRouterIntegrationTest {
private static final long DURATION_MILIS = 10000;
private static final String SOURCE_FOLDER = "src/test/source-folder";
private static final String DESTINATION_FOLDER = "src/test/destination-folder";
@Before
public void setUp() throws Exception {
File sourceFolder = new File(SOURCE_FOLDER);
File destinationFolder = new File(DESTINATION_FOLDER);
cleanFolder(sourceFolder);
cleanFolder(destinationFolder);
sourceFolder.mkdirs();
File file = new File(SOURCE_FOLDER + "/File.txt");
FileWriter fileWriter = new FileWriter(file, false);
fileWriter.write("Hello\nWorld");
file.createNewFile();
fileWriter.close();
}
private void cleanFolder(File folder) {
File[] files = folder.listFiles();
if (files != null) {
for (File file : files) {
if (file.isFile()) {
file.delete();
}
}
}
}
@Test
@Ignore
public void routeTests() throws InterruptedException {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("camel-context-SplitterFileRouter.xml");
Thread.sleep(DURATION_MILIS);
applicationContext.close();
}
}
@@ -0,0 +1,117 @@
package com.apache.camel.main;
import com.baeldung.camel.main.App;
import junit.framework.TestCase;
import org.apache.camel.util.FileUtil;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Paths;
public class AppIntegrationTest extends TestCase {
private static final String FILE_NAME = "file.txt";
private static final String SAMPLE_INPUT_DIR = "src/test/data/sampleInputFile/";
private static final String TEST_INPUT_DIR = "src/test/data/input/";
private static final String UPPERCASE_OUTPUT_DIR = "src/test/data/outputUpperCase/";
private static final String LOWERCASE_OUTPUT_DIR = "src/test/data/outputLowerCase/";
@Before
public void setUp() throws Exception {
// Prepare input file for test
copySampleFileToInputDirectory();
}
@After
public void tearDown() throws Exception {
System.out.println("Deleting the test input and output files...");
deleteFile(TEST_INPUT_DIR);
deleteFile(LOWERCASE_OUTPUT_DIR);
deleteFile(UPPERCASE_OUTPUT_DIR);
}
@Test
public final void testMain() throws Exception {
App.main(null);
String inputFileContent = readFileContent(SAMPLE_INPUT_DIR + FILE_NAME);
String outputUpperCase = readFileContent(UPPERCASE_OUTPUT_DIR + FILE_NAME);
String outputLowerCase = readFileContent(LOWERCASE_OUTPUT_DIR + FILE_NAME);
System.out.println("Input File content = [" + inputFileContent + "]");
System.out.println("UpperCaseOutput file content = [" + outputUpperCase + "]");
System.out.println("LowerCaseOtput file content = [" + outputLowerCase + "]");
assertEquals(inputFileContent.toUpperCase(), outputUpperCase);
assertEquals(inputFileContent.toLowerCase(), outputLowerCase);
}
private String readFileContent(String path) throws IOException {
byte[] encoded = Files.readAllBytes(Paths.get(path));
return new String(encoded);
}
private void deleteFile(String path) {
try {
FileUtil.removeDir(new File(path));
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* Copy sample input file to input directory.
*/
private void copySampleFileToInputDirectory() {
File sourceFile = new File(SAMPLE_INPUT_DIR + FILE_NAME);
File destFile = new File(TEST_INPUT_DIR + FILE_NAME);
if (!sourceFile.exists()) {
System.out.println("Sample input file not found at location = [" + SAMPLE_INPUT_DIR + FILE_NAME + "]. Please provide this file.");
}
if (!destFile.exists()) {
try {
System.out.println("Creating input file = [" + TEST_INPUT_DIR + FILE_NAME + "]");
File destDir = new File(TEST_INPUT_DIR);
if (!destDir.exists()) {
destDir.mkdir();
}
destFile.createNewFile();
} catch (IOException e) {
e.printStackTrace();
}
}
FileChannel source = null;
FileChannel destination = null;
try {
source = new FileInputStream(sourceFile).getChannel();
destination = new FileOutputStream(destFile).getChannel();
if (destination != null && source != null) {
destination.transferFrom(source, 0, source.size());
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
source.close();
} catch (Exception e) {
e.printStackTrace();
}
try {
destination.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
@@ -0,0 +1,13 @@
package com.baeldung;
import org.junit.Test;
import com.baeldung.camel.main.App;
public class SpringContextTest {
@Test
public final void testMain() throws Exception {
App.main(null);
}
}
@@ -0,0 +1,60 @@
package com.baeldung.camel.jackson;
import java.io.IOException;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jackson.ListJacksonDataFormat;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;
public class FruitArrayJacksonUnmarshalUnitTest extends CamelTestSupport {
@Test
public void givenJsonFruitArray_whenUnmarshalled_thenSuccess() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:marshalledObject");
mock.expectedMessageCount(1);
mock.message(0).body().isInstanceOf(List.class);
String json = readJsonFromFile("/json/fruit-array.json");
template.sendBody("direct:jsonInput", json);
assertMockEndpointsSatisfied();
@SuppressWarnings("unchecked")
List<Fruit> fruitList = mock.getReceivedExchanges().get(0).getIn().getBody(List.class);
assertNotNull("Fruit lists should not be null", fruitList);
assertEquals("There should be two fruits", 2, fruitList.size());
Fruit fruit = fruitList.get(0);
assertEquals("Fruit name", "Banana", fruit.getName());
assertEquals("Fruit id", 100, fruit.getId());
fruit = fruitList.get(1);
assertEquals("Fruit name", "Apple", fruit.getName());
assertEquals("Fruit id", 101, fruit.getId());
}
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
from("direct:jsonInput").unmarshal(new ListJacksonDataFormat(Fruit.class))
.to("mock:marshalledObject");
}
};
}
private String readJsonFromFile(String path) throws URISyntaxException, IOException {
URL resource = FruitArrayJacksonUnmarshalUnitTest.class.getResource(path);
return new String(Files.readAllBytes(Paths.get(resource.toURI())));
}
}
@@ -0,0 +1,59 @@
package com.baeldung.camel.jackson;
import java.io.IOException;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jackson.JacksonDataFormat;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;
public class FruitListJacksonUnmarshalUnitTest extends CamelTestSupport {
@Test
public void givenJsonFruitList_whenUnmarshalled_thenSuccess() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:marshalledObject");
mock.expectedMessageCount(1);
mock.message(0).body().isInstanceOf(FruitList.class);
String json = readJsonFromFile("/json/fruit-list.json");
template.sendBody("direct:jsonInput", json);
assertMockEndpointsSatisfied();
FruitList fruitList = mock.getReceivedExchanges().get(0).getIn().getBody(FruitList.class);
assertNotNull("Fruit lists should not be null", fruitList);
List<Fruit> fruits = fruitList.getFruits();
assertEquals("There should be two fruits", 2, fruits.size());
Fruit fruit = fruits.get(0);
assertEquals("Fruit name", "Banana", fruit.getName());
assertEquals("Fruit id", 100, fruit.getId());
fruit = fruits.get(1);
assertEquals("Fruit name", "Apple", fruit.getName());
assertEquals("Fruit id", 101, fruit.getId());
}
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
from("direct:jsonInput").unmarshal(new JacksonDataFormat(FruitList.class))
.to("mock:marshalledObject");
}
};
}
private String readJsonFromFile(String path) throws URISyntaxException, IOException {
URL resource = FruitListJacksonUnmarshalUnitTest.class.getResource(path);
return new String(Files.readAllBytes(Paths.get(resource.toURI())));
}
}
@@ -0,0 +1,10 @@
[
{
"id": 100,
"name": "Banana"
},
{
"id": 101,
"name": "Apple"
}
]
@@ -0,0 +1,12 @@
{
"fruits": [
{
"id": 100,
"name": "Banana"
},
{
"id": 101,
"name": "Apple"
}
]
}
+7
View File
@@ -0,0 +1,7 @@
## Spring JMS
This module contains articles about Spring with JMS
### Relevant Articles:
- [Getting Started with Spring JMS](https://www.baeldung.com/spring-jms)
- [Testing Spring JMS](https://www.baeldung.com/spring-jms-testing)
+88
View File
@@ -0,0 +1,88 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
<artifactId>spring-jms</artifactId>
<name>spring-jms</name>
<packaging>war</packaging>
<description>Introduction to Spring JMS</description>
<parent>
<groupId>com.baeldung</groupId>
<artifactId>messaging-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<dependencies>
<!-- Spring JMS -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${springframework.version}</version>
<exclusions>
<exclusion>
<artifactId>commons-logging</artifactId>
<groupId>commons-logging</groupId>
</exclusion>
</exclusions>
</dependency>
<!-- ActiveMQ -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>${activemq.version}</version>
</dependency>
<!-- Test -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<version>${spring-boot-test.version}</version>
<scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.mockito/mockito-core -->
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>4.6.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.activemq.tooling</groupId>
<artifactId>activemq-junit</artifactId>
<version>5.16.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>1.17.3</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-war-plugin</artifactId>
<version>${maven-war-plugin.version}</version>
<configuration>
<warSourceDirectory>src/main/webapp</warSourceDirectory>
<warName>spring-jms</warName>
<failOnMissingWebXml>false</failOnMissingWebXml>
</configuration>
</plugin>
</plugins>
</pluginManagement>
<finalName>spring-jms</finalName>
</build>
<properties>
<springframework.version>4.3.4.RELEASE</springframework.version>
<activemq.version>5.14.1</activemq.version>
<spring-boot-test.version>1.5.10.RELEASE</spring-boot-test.version>
</properties>
</project>
@@ -0,0 +1,23 @@
package com.baeldung.spring.jms;
public class Employee {
private String name;
private Integer age;
public Employee(String name, Integer age) {
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public Integer getAge() {
return age;
}
public String toString() {
return "Employee: name(" + name + "), age(" + age + ")";
}
}
@@ -0,0 +1,17 @@
package com.baeldung.spring.jms;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.ErrorHandler;
public class SampleJmsErrorHandler implements ErrorHandler {
private static final Logger LOG = LoggerFactory.getLogger(SampleJmsErrorHandler.class);
@Override
public void handleError(Throwable t) {
LOG.warn("In default jms error handler...");
LOG.error("Error Message : {}", t.getMessage());
}
}
@@ -0,0 +1,33 @@
package com.baeldung.spring.jms;
import org.springframework.jms.core.JmsTemplate;
import javax.jms.Queue;
import java.util.HashMap;
import java.util.Map;
public class SampleJmsMessageSender {
private JmsTemplate jmsTemplate;
private Queue queue;
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void setQueue(Queue queue) {
this.queue = queue;
}
public void simpleSend() {
jmsTemplate.send(queue, s -> s.createTextMessage("hello queue world"));
}
public void sendMessage(final Employee employee) {
this.jmsTemplate.convertAndSend(employee);
}
public void sendTextMessage(String msg) {
this.jmsTemplate.send(queue, s -> s.createTextMessage(msg));
}
}
@@ -0,0 +1,44 @@
package com.baeldung.spring.jms;
import org.springframework.jms.core.JmsTemplate;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.TextMessage;
import java.util.Map;
public class SampleListener implements MessageListener {
private JmsTemplate jmsTemplate;
private Queue queue;
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void setQueue(Queue queue) {
this.queue = queue;
}
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
String msg = ((TextMessage) message).getText();
System.out.println("Received message: " + msg);
if (msg == null) {
throw new IllegalArgumentException("Null value received...");
}
} catch (JMSException ex) {
throw new RuntimeException(ex);
}
}
}
public Employee receiveMessage() throws JMSException {
Map map = (Map) this.jmsTemplate.receiveAndConvert();
return new Employee((String) map.get("name"), (Integer) map.get("age"));
}
}
@@ -0,0 +1,26 @@
package com.baeldung.spring.jms;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.Session;
public class SampleMessageConverter implements MessageConverter {
public Message toMessage(Object object, Session session) throws JMSException, MessageConversionException {
Employee employee = (Employee) object;
MapMessage message = session.createMapMessage();
message.setString("name", employee.getName());
message.setInt("age", employee.getAge());
return message;
}
public Object fromMessage(Message message) throws JMSException, MessageConversionException {
MapMessage mapMessage = (MapMessage) message;
return new Employee(mapMessage.getString("name"), mapMessage.getInt("age"));
}
}
@@ -0,0 +1,13 @@
package com.baeldung.spring.jms.testing;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;
@ComponentScan
public class JmsApplication {
public static void main(String[] args) {
ApplicationContext context = new AnnotationConfigApplicationContext(JmsApplication.class);
}
}
@@ -0,0 +1,34 @@
package com.baeldung.spring.jms.testing;
import javax.jms.ConnectionFactory;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.core.JmsTemplate;
@Configuration
@EnableJms
public class JmsConfig {
@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerFactory() {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(connectionFactory());
return factory;
}
@Bean
public ConnectionFactory connectionFactory() {
return new ActiveMQConnectionFactory("tcp://localhost:61616");
}
@Bean
public JmsTemplate jmsTemplate() {
return new JmsTemplate(connectionFactory());
}
}
@@ -0,0 +1,20 @@
package com.baeldung.spring.jms.testing;
import javax.jms.JMSException;
import javax.jms.TextMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
@Component
public class MessageListener {
private static final Logger logger = LoggerFactory.getLogger(MessageListener.class);
@JmsListener(destination = "queue-1")
public void sampleJmsListenerMethod(TextMessage message) throws JMSException {
logger.info("JMS listener received text message: {}", message.getText());
}
}

Some files were not shown because too many files have changed in this diff Show More