Skip to content

C
CONF
Commit 77fd0858 authored by Spiros Koulouzis's avatar Spiros Koulouzis
Browse files
Options

use new channel for each request

parent 535d96b9
Hide whitespace changes
Inline Side-by-side
Showing with 99 additions and 98 deletions
docker-compose/docker-compose.yml
View file @ 77fd0858
......@@ -82,32 +82,32 @@ services:
#- "3001:3000"
manager:
depends_on:
- rabbit
- mongo
- sure-tosca
image: qcdis/manager
environment:
RABBITMQ_HOST: rabbit
MONGO_HOST: mongo
SURE_TOSCA_BASE_PATH: http://sure-tosca:8081/tosca-sure/1.0.0
CREDENTIAL_SECRET: top_secret
ports:
- "8080:8080"
#manager:
#depends_on:
#- rabbit
#- mongo
#- sure-tosca
#image: qcdis/manager
#environment:
#RABBITMQ_HOST: rabbit
#MONGO_HOST: mongo
#SURE_TOSCA_BASE_PATH: http://sure-tosca:8081/tosca-sure/1.0.0
#CREDENTIAL_SECRET: top_secret
#ports:
#- "8080:8080"
sure-tosca:
image: qcdis/sure-tosca
ports:
- "8081:8081"
#planner:
#depends_on:
#- rabbit
#- sure-tosca
#image: qcdis/planner
#environment:
#RABBITMQ_HOST: rabbit
planner:
depends_on:
- rabbit
- sure-tosca
image: qcdis/planner
environment:
RABBITMQ_HOST: rabbit
provisioner:
depends_on:
......
manager/src/main/java/nl/uva/sne/drip/rpc/DRIPCaller.java
View file @ 77fd0858
......@@ -43,9 +43,9 @@ import org.springframework.stereotype.Service;
@Service
public class DRIPCaller implements AutoCloseable {
private Connection connection;
private Channel channel;
private String replyQueueName;
// private Connection connection;
// private Channel channel;
// private String replyQueueName;
private String requestQeueName;
private final ObjectMapper mapper;
private final ConnectionFactory factory;
......@@ -63,92 +63,93 @@ public class DRIPCaller implements AutoCloseable {
}
public void init() throws IOException, TimeoutException {
if (connection == null || !connection.isOpen()) {
connection = factory.newConnection();
channel = connection.createChannel();
// create a single callback queue per client not per requests.
Map<String, Object> args = new HashMap<>();
// args.put("x-message-ttl", 60000);
// replyQueueName = channel.queueDeclare("myqueue", false, false, false, args).getQueue();
replyQueueName = channel.queueDeclare().getQueue();
}
}
/**
* @return the connection
*/
public Connection getConnection() {
return connection;
}
/**
* @return the channel
*/
public Channel getChannel() {
return channel;
}
/**
* @return the replyQueueName
*/
public String getReplyQueueName() {
return replyQueueName;
// if (connection == null || !connection.isOpen()) {
// connection = factory.newConnection();
// }
}
// /**
// * @return the connection
// */
// public Connection getConnection() {
// return connection;
// }
// /**
// * @return the channel
// */
// public Channel getChannel() {
// return channel;
// }
// /**
// * @return the replyQueueName
// */
// public String getReplyQueueName() {
// return replyQueueName;
// }
@Override
public void close() throws IOException, TimeoutException {
if (channel != null && channel.isOpen()) {
channel.close();
}
if (connection != null && connection.isOpen()) {
connection.close();
}
// if (connection != null && connection.isOpen()) {
// connection.close();
// }
}
public Message call(Message r) throws IOException, TimeoutException, InterruptedException {
Channel channel = null;
Connection connection = null;
try {
String jsonInString = mapper.writeValueAsString(r);
int timeOut = 25;
if (getRequestQeueName().equals("planner")) {
timeOut = 5;
}
if (getRequestQeueName().equals("provisioner")) {
timeOut = 10;
}
connection = factory.newConnection();
String jsonInString = mapper.writeValueAsString(r);
int timeOut = 25;
if (getRequestQeueName().equals("planner")) {
timeOut = 5;
}
if (getRequestQeueName().equals("provisioner")) {
timeOut = 10;
}
//Build a correlation ID to distinguish responds
final String corrId = UUID.randomUUID().toString();
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.correlationId(corrId)
.expiration(String.valueOf(timeOut*60000))
.replyTo(getReplyQueueName())
.build();
Logger.getLogger(DRIPCaller.class.getName()).log(Level.INFO, "Sending: {0} to queue: {1}", new Object[]{jsonInString, getRequestQeueName()});
getChannel().basicPublish("", getRequestQeueName(), props, jsonInString.getBytes("UTF-8"));
final BlockingQueue<String> response = new ArrayBlockingQueue(1);
getChannel().basicConsume(getReplyQueueName(), true, new DefaultConsumer(getChannel()) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (properties.getCorrelationId().equals(corrId)) {
response.offer(new String(body, "UTF-8"));
channel = connection.createChannel();
Map<String, Object> args = new HashMap<>();
String replyQueueName = channel.queueDeclare().getQueue();
//Build a correlation ID to distinguish responds
final String corrId = UUID.randomUUID().toString();
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.correlationId(corrId)
.expiration(String.valueOf(timeOut * 60000))
.replyTo(replyQueueName)
.build();
Logger.getLogger(DRIPCaller.class.getName()).log(Level.INFO, "Sending: {0} to queue: {1}", new Object[]{jsonInString, getRequestQeueName()});
channel.basicPublish("", getRequestQeueName(), props, jsonInString.getBytes("UTF-8"));
final BlockingQueue<String> response = new ArrayBlockingQueue(1);
channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (properties.getCorrelationId().equals(corrId)) {
response.offer(new String(body, "UTF-8"));
}
}
}
});
});
// String resp = response.take();
String resp = response.poll(timeOut, TimeUnit.MINUTES);
Logger.getLogger(DRIPCaller.class.getName()).log(Level.INFO, "Got: {0}", resp);
if (resp == null) {
getChannel().queueDeleteNoWait(getReplyQueueName(), false, true);
close();
throw new TimeoutException("Timeout on qeue: " + getRequestQeueName());
String resp = response.poll(timeOut, TimeUnit.MINUTES);
Logger.getLogger(DRIPCaller.class.getName()).log(Level.INFO, "Got: {0}", resp);
if (resp == null) {
throw new TimeoutException("Timeout on qeue: " + getRequestQeueName());
}
return mapper.readValue(resp, Message.class);
} finally {
if (channel != null && channel.isOpen()) {
channel.close();
}
if (connection != null && connection.isOpen()) {
connection.close();
}
}
return mapper.readValue(resp, Message.class);
}
/**
......
pom.xml
View file @ 77fd0858
......@@ -9,7 +9,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.1.RELEASE</version>
<version>2.2.2.RELEASE</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment