forked from D-Net/dnet-hadoop
Merge branch 'hadoop_aggregator' of code-repo.d4science.org:D-Net/dnet-hadoop into hadoop_aggregator
This commit is contained in:
commit
4ed1e306b6
|
|
@ -53,11 +53,6 @@
|
||||||
<groupId>com.fasterxml.jackson.core</groupId>
|
<groupId>com.fasterxml.jackson.core</groupId>
|
||||||
<artifactId>jackson-databind</artifactId>
|
<artifactId>jackson-databind</artifactId>
|
||||||
</dependency>
|
</dependency>
|
||||||
<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
|
|
||||||
<dependency>
|
|
||||||
<groupId>com.rabbitmq</groupId>
|
|
||||||
<artifactId>amqp-client</artifactId>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>net.sf.saxon</groupId>
|
<groupId>net.sf.saxon</groupId>
|
||||||
<artifactId>Saxon-HE</artifactId>
|
<artifactId>Saxon-HE</artifactId>
|
||||||
|
|
|
||||||
|
|
@ -1,76 +0,0 @@
|
||||||
|
|
||||||
package eu.dnetlib.message;
|
|
||||||
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.util.Map;
|
|
||||||
|
|
||||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
||||||
|
|
||||||
public class Message {
|
|
||||||
|
|
||||||
private String workflowId;
|
|
||||||
|
|
||||||
private String jobName;
|
|
||||||
|
|
||||||
private MessageType type;
|
|
||||||
|
|
||||||
private Map<String, String> body;
|
|
||||||
|
|
||||||
public static Message fromJson(final String json) throws IOException {
|
|
||||||
final ObjectMapper jsonMapper = new ObjectMapper();
|
|
||||||
return jsonMapper.readValue(json, Message.class);
|
|
||||||
}
|
|
||||||
|
|
||||||
public Message() {
|
|
||||||
}
|
|
||||||
|
|
||||||
public Message(String workflowId, String jobName, MessageType type, Map<String, String> body) {
|
|
||||||
this.workflowId = workflowId;
|
|
||||||
this.jobName = jobName;
|
|
||||||
this.type = type;
|
|
||||||
this.body = body;
|
|
||||||
}
|
|
||||||
|
|
||||||
public String getWorkflowId() {
|
|
||||||
return workflowId;
|
|
||||||
}
|
|
||||||
|
|
||||||
public void setWorkflowId(String workflowId) {
|
|
||||||
this.workflowId = workflowId;
|
|
||||||
}
|
|
||||||
|
|
||||||
public String getJobName() {
|
|
||||||
return jobName;
|
|
||||||
}
|
|
||||||
|
|
||||||
public void setJobName(String jobName) {
|
|
||||||
this.jobName = jobName;
|
|
||||||
}
|
|
||||||
|
|
||||||
public MessageType getType() {
|
|
||||||
return type;
|
|
||||||
}
|
|
||||||
|
|
||||||
public void setType(MessageType type) {
|
|
||||||
this.type = type;
|
|
||||||
}
|
|
||||||
|
|
||||||
public Map<String, String> getBody() {
|
|
||||||
return body;
|
|
||||||
}
|
|
||||||
|
|
||||||
public void setBody(Map<String, String> body) {
|
|
||||||
this.body = body;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public String toString() {
|
|
||||||
final ObjectMapper jsonMapper = new ObjectMapper();
|
|
||||||
try {
|
|
||||||
return jsonMapper.writeValueAsString(this);
|
|
||||||
} catch (JsonProcessingException e) {
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,47 +0,0 @@
|
||||||
|
|
||||||
package eu.dnetlib.message;
|
|
||||||
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.nio.charset.StandardCharsets;
|
|
||||||
import java.util.concurrent.LinkedBlockingQueue;
|
|
||||||
|
|
||||||
import com.rabbitmq.client.AMQP;
|
|
||||||
import com.rabbitmq.client.Channel;
|
|
||||||
import com.rabbitmq.client.DefaultConsumer;
|
|
||||||
import com.rabbitmq.client.Envelope;
|
|
||||||
|
|
||||||
public class MessageConsumer extends DefaultConsumer {
|
|
||||||
|
|
||||||
final LinkedBlockingQueue<Message> queueMessages;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Constructs a new instance and records its association to the passed-in channel.
|
|
||||||
*
|
|
||||||
* @param channel the channel to which this consumer is attached
|
|
||||||
* @param queueMessages
|
|
||||||
*/
|
|
||||||
public MessageConsumer(Channel channel, LinkedBlockingQueue<Message> queueMessages) {
|
|
||||||
super(channel);
|
|
||||||
this.queueMessages = queueMessages;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void handleDelivery(
|
|
||||||
String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
|
|
||||||
throws IOException {
|
|
||||||
final String json = new String(body, StandardCharsets.UTF_8);
|
|
||||||
Message message = Message.fromJson(json);
|
|
||||||
try {
|
|
||||||
this.queueMessages.put(message);
|
|
||||||
System.out.println("Receiving Message " + message);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
if (message.getType() == MessageType.REPORT)
|
|
||||||
throw new RuntimeException("Error on sending message");
|
|
||||||
else {
|
|
||||||
// TODO LOGGING EXCEPTION
|
|
||||||
}
|
|
||||||
} finally {
|
|
||||||
getChannel().basicAck(envelope.getDeliveryTag(), false);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,136 +0,0 @@
|
||||||
|
|
||||||
package eu.dnetlib.message;
|
|
||||||
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.util.HashMap;
|
|
||||||
import java.util.Map;
|
|
||||||
import java.util.concurrent.LinkedBlockingQueue;
|
|
||||||
import java.util.concurrent.TimeoutException;
|
|
||||||
|
|
||||||
import com.rabbitmq.client.Channel;
|
|
||||||
import com.rabbitmq.client.Connection;
|
|
||||||
import com.rabbitmq.client.ConnectionFactory;
|
|
||||||
|
|
||||||
public class MessageManager {
|
|
||||||
|
|
||||||
private final String messageHost;
|
|
||||||
|
|
||||||
private final String username;
|
|
||||||
|
|
||||||
private final String password;
|
|
||||||
|
|
||||||
private Connection connection;
|
|
||||||
|
|
||||||
private final Map<String, Channel> channels = new HashMap<>();
|
|
||||||
|
|
||||||
private boolean durable;
|
|
||||||
|
|
||||||
private boolean autodelete;
|
|
||||||
|
|
||||||
private final LinkedBlockingQueue<Message> queueMessages;
|
|
||||||
|
|
||||||
public MessageManager(
|
|
||||||
String messageHost,
|
|
||||||
String username,
|
|
||||||
String password,
|
|
||||||
final LinkedBlockingQueue<Message> queueMessages) {
|
|
||||||
this.queueMessages = queueMessages;
|
|
||||||
this.messageHost = messageHost;
|
|
||||||
this.username = username;
|
|
||||||
this.password = password;
|
|
||||||
}
|
|
||||||
|
|
||||||
public MessageManager(
|
|
||||||
String messageHost,
|
|
||||||
String username,
|
|
||||||
String password,
|
|
||||||
boolean durable,
|
|
||||||
boolean autodelete,
|
|
||||||
final LinkedBlockingQueue<Message> queueMessages) {
|
|
||||||
this.queueMessages = queueMessages;
|
|
||||||
this.messageHost = messageHost;
|
|
||||||
this.username = username;
|
|
||||||
this.password = password;
|
|
||||||
|
|
||||||
this.durable = durable;
|
|
||||||
this.autodelete = autodelete;
|
|
||||||
}
|
|
||||||
|
|
||||||
private Connection createConnection() throws IOException, TimeoutException {
|
|
||||||
ConnectionFactory factory = new ConnectionFactory();
|
|
||||||
factory.setHost(this.messageHost);
|
|
||||||
factory.setUsername(this.username);
|
|
||||||
factory.setPassword(this.password);
|
|
||||||
return factory.newConnection();
|
|
||||||
}
|
|
||||||
|
|
||||||
private Channel createChannel(
|
|
||||||
final Connection connection,
|
|
||||||
final String queueName,
|
|
||||||
final boolean durable,
|
|
||||||
final boolean autodelete)
|
|
||||||
throws Exception {
|
|
||||||
Map<String, Object> args = new HashMap<>();
|
|
||||||
args.put("x-message-ttl", 10000);
|
|
||||||
Channel channel = connection.createChannel();
|
|
||||||
channel.queueDeclare(queueName, durable, false, this.autodelete, args);
|
|
||||||
return channel;
|
|
||||||
}
|
|
||||||
|
|
||||||
private Channel getOrCreateChannel(final String queueName, boolean durable, boolean autodelete)
|
|
||||||
throws Exception {
|
|
||||||
if (channels.containsKey(queueName)) {
|
|
||||||
return channels.get(queueName);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (this.connection == null) {
|
|
||||||
this.connection = createConnection();
|
|
||||||
}
|
|
||||||
channels.put(queueName, createChannel(this.connection, queueName, durable, autodelete));
|
|
||||||
return channels.get(queueName);
|
|
||||||
}
|
|
||||||
|
|
||||||
public void close() throws IOException {
|
|
||||||
channels
|
|
||||||
.values()
|
|
||||||
.forEach(
|
|
||||||
ch -> {
|
|
||||||
try {
|
|
||||||
ch.close();
|
|
||||||
} catch (Exception e) {
|
|
||||||
// TODO LOG
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
this.connection.close();
|
|
||||||
}
|
|
||||||
|
|
||||||
public boolean sendMessage(final Message message, String queueName) throws Exception {
|
|
||||||
try {
|
|
||||||
Channel channel = getOrCreateChannel(queueName, this.durable, this.autodelete);
|
|
||||||
channel.basicPublish("", queueName, null, message.toString().getBytes());
|
|
||||||
return true;
|
|
||||||
} catch (Throwable e) {
|
|
||||||
throw new RuntimeException(e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public boolean sendMessage(
|
|
||||||
final Message message, String queueName, boolean durable_var, boolean autodelete_var)
|
|
||||||
throws Exception {
|
|
||||||
try {
|
|
||||||
Channel channel = getOrCreateChannel(queueName, durable_var, autodelete_var);
|
|
||||||
channel.basicPublish("", queueName, null, message.toString().getBytes());
|
|
||||||
return true;
|
|
||||||
} catch (Throwable e) {
|
|
||||||
throw new RuntimeException(e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public void startConsumingMessage(
|
|
||||||
final String queueName, final boolean durable, final boolean autodelete) throws Exception {
|
|
||||||
|
|
||||||
Channel channel = createChannel(createConnection(), queueName, durable, autodelete);
|
|
||||||
channel.basicConsume(queueName, false, new MessageConsumer(channel, queueMessages));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,6 +0,0 @@
|
||||||
|
|
||||||
package eu.dnetlib.message;
|
|
||||||
|
|
||||||
public enum MessageType {
|
|
||||||
ONGOING, REPORT
|
|
||||||
}
|
|
||||||
|
|
@ -1,51 +0,0 @@
|
||||||
|
|
||||||
package eu.dnetlib.message;
|
|
||||||
|
|
||||||
import static org.junit.jupiter.api.Assertions.*;
|
|
||||||
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.util.HashMap;
|
|
||||||
import java.util.Map;
|
|
||||||
|
|
||||||
import org.junit.jupiter.api.Test;
|
|
||||||
|
|
||||||
public class MessageTest {
|
|
||||||
|
|
||||||
@Test
|
|
||||||
public void fromJsonTest() throws IOException {
|
|
||||||
Message m = new Message();
|
|
||||||
m.setWorkflowId("wId");
|
|
||||||
m.setType(MessageType.ONGOING);
|
|
||||||
m.setJobName("Collection");
|
|
||||||
Map<String, String> body = new HashMap<>();
|
|
||||||
body.put("parsedItem", "300");
|
|
||||||
body.put("ExecutionTime", "30s");
|
|
||||||
|
|
||||||
m.setBody(body);
|
|
||||||
System.out.println("m = " + m);
|
|
||||||
Message m1 = Message.fromJson(m.toString());
|
|
||||||
assertEquals(m1.getWorkflowId(), m.getWorkflowId());
|
|
||||||
assertEquals(m1.getType(), m.getType());
|
|
||||||
assertEquals(m1.getJobName(), m.getJobName());
|
|
||||||
|
|
||||||
assertNotNull(m1.getBody());
|
|
||||||
m1.getBody().keySet().forEach(it -> assertEquals(m1.getBody().get(it), m.getBody().get(it)));
|
|
||||||
assertEquals(m1.getJobName(), m.getJobName());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
public void toStringTest() {
|
|
||||||
final String expectedJson = "{\"workflowId\":\"wId\",\"jobName\":\"Collection\",\"type\":\"ONGOING\",\"body\":{\"ExecutionTime\":\"30s\",\"parsedItem\":\"300\"}}";
|
|
||||||
Message m = new Message();
|
|
||||||
m.setWorkflowId("wId");
|
|
||||||
m.setType(MessageType.ONGOING);
|
|
||||||
m.setJobName("Collection");
|
|
||||||
Map<String, String> body = new HashMap<>();
|
|
||||||
body.put("parsedItem", "300");
|
|
||||||
body.put("ExecutionTime", "30s");
|
|
||||||
|
|
||||||
m.setBody(body);
|
|
||||||
|
|
||||||
assertEquals(expectedJson, m.toString());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
5
pom.xml
5
pom.xml
|
|
@ -374,11 +374,6 @@
|
||||||
<scope>provided</scope>
|
<scope>provided</scope>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
<dependency>
|
|
||||||
<groupId>com.rabbitmq</groupId>
|
|
||||||
<artifactId>amqp-client</artifactId>
|
|
||||||
<version>5.6.0</version>
|
|
||||||
</dependency>
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>com.jayway.jsonpath</groupId>
|
<groupId>com.jayway.jsonpath</groupId>
|
||||||
<artifactId>json-path</artifactId>
|
<artifactId>json-path</artifactId>
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue