zenon/distributed-systems-technologies
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Implement QueueManager (3.1.1)

Browse Source
This commit is contained in:
Tobias Eidelpes 2021-05-17 15:55:23 +02:00
parent be629aa432
commit 8cb4304abc
1 changed files with 67 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
ass3-messaging/src/main/java/dst/ass3/messaging/impl/QueueManager.java Normal file
View File

@ -0,0 +1,67 @@
package dst.ass3.messaging.impl;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import dst.ass3.messaging.Constants;
import dst.ass3.messaging.IQueueManager;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class QueueManager implements IQueueManager {
private Connection connection;
private Channel channel;
private String viennaQueue;
private String linzQueue;
private String berlinQueue;
@Override
public void setUp() {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(Constants.RMQ_HOST);
factory.setUsername(Constants.RMQ_USER);
factory.setPassword(Constants.RMQ_PASSWORD);
try{
connection = factory.newConnection();
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
try {
channel = connection.createChannel();
channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, "topic");
channel.exchangeDeclare(Constants.QUEUE_AT_VIENNA, "direct", true);
channel.exchangeDeclare(Constants.QUEUE_AT_LINZ, "direct", true);
channel.exchangeDeclare(Constants.QUEUE_DE_BERLIN, "direct", true);
channel.queueBind(viennaQueue, Constants.QUEUE_AT_VIENNA, Constants.ROUTING_KEY_AT_VIENNA.toLowerCase());
channel.queueBind(linzQueue, Constants.QUEUE_AT_LINZ, Constants.ROUTING_KEY_AT_LINZ.toLowerCase());
channel.queueBind(berlinQueue, Constants.QUEUE_DE_BERLIN, Constants.ROUTING_KEY_DE_BERLIN.toLowerCase());
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void tearDown() {
try {
channel.queueDelete(berlinQueue);
channel.queueDelete(linzQueue);
channel.queueDelete(viennaQueue);
channel.exchangeDelete(Constants.QUEUE_DE_BERLIN);
channel.exchangeDelete(Constants.QUEUE_AT_LINZ);
channel.exchangeDelete(Constants.QUEUE_AT_VIENNA);
channel.exchangeDelete(Constants.TOPIC_EXCHANGE);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void close() throws IOException {
if (connection != null)
connection.close();
}
}