anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Simplyfied creation of HazelcastMsg from InterceptPublishMessage

Browse Source
This commit is contained in:
andsel
2016-07-22 13:19:16 +02:00
parent 29fd6d03b9
commit 8cc588d766
3 changed files with 15 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java
+1 -2
View File
@@ -23,8 +23,7 @@ public class HazelcastInterceptHandler extends AbstractInterceptHandler {
public void onPublish(InterceptPublishMessage msg) { public void onPublish(InterceptPublishMessage msg) {
LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array())); LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array()));
ITopic<HazelcastMsg> topic = hz.getTopic("moquette"); ITopic<HazelcastMsg> topic = hz.getTopic("moquette");
HazelcastMsg hazelcastMsg = new HazelcastMsg(msg.getClientID(), msg.getTopicName(), msg.getQos().byteValue(), HazelcastMsg hazelcastMsg = new HazelcastMsg(msg);
msg.getPayload().array(), msg.getUsername());
topic.publish(hazelcastMsg); topic.publish(hazelcastMsg);
} }
broker/src/main/java/io/moquette/interception/HazelcastMsg.java
+7 -11
View File
@@ -1,5 +1,7 @@
package io.moquette.interception; package io.moquette.interception;
import io.moquette.interception.messages.InterceptPublishMessage;
import java.io.Serializable; import java.io.Serializable;
/** /**
@@ -8,17 +10,15 @@ import java.io.Serializable;
public class HazelcastMsg implements Serializable { public class HazelcastMsg implements Serializable {
private final String clientId; private final String clientId;
private final String userName;
private final byte qos; private final byte qos;
private final byte[] payload; private final byte[] payload;
private final String topic; private final String topic;
public HazelcastMsg(String clientId, String topic, byte qos, byte[] payload, String userName) { public HazelcastMsg(InterceptPublishMessage msg) {
this.clientId = clientId; this.clientId = msg.getClientID();
this.qos = qos; this.topic = msg.getTopicName();
this.payload = payload; this.qos = msg.getQos().byteValue();
this.userName = userName; this.payload = msg.getPayload().array();
this.topic = topic;
} }
public String getClientId() { public String getClientId() {
@@ -33,10 +33,6 @@ public class HazelcastMsg implements Serializable {
return payload; return payload;
} }
public String getUserName() {
return userName;
}
public String getTopic() { public String getTopic() {
return topic; return topic;
} }
broker/src/main/java/io/moquette/server/HazelcastListener.java
+7 -6
View File
@@ -26,14 +26,15 @@ public class HazelcastListener implements MessageListener<HazelcastMsg> {
public void onMessage(Message<HazelcastMsg> msg) { public void onMessage(Message<HazelcastMsg> msg) {
try { try {
if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) { if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) {
LOG.info("{} received from hazelcast for topic {} message: {}", msg.getMessageObject().getClientId(), HazelcastMsg hzMsg = msg.getMessageObject();
msg.getMessageObject().getTopic(), msg.getMessageObject().getPayload()); LOG.info("{} received from hazelcast for topic {} message: {}", hzMsg.getClientId(),
hzMsg.getTopic(), hzMsg.getPayload());
PublishMessage publishMessage = new PublishMessage(); PublishMessage publishMessage = new PublishMessage();
publishMessage.setTopicName(msg.getMessageObject().getTopic()); publishMessage.setTopicName(hzMsg.getTopic());
publishMessage.setQos(AbstractMessage.QOSType.valueOf(msg.getMessageObject().getQos())); publishMessage.setQos(AbstractMessage.QOSType.valueOf(hzMsg.getQos()));
publishMessage.setPayload(ByteBuffer.wrap(msg.getMessageObject().getPayload())); publishMessage.setPayload(ByteBuffer.wrap(hzMsg.getPayload()));
publishMessage.setLocal(false); publishMessage.setLocal(false);
publishMessage.setClientId(msg.getMessageObject().getClientId()); publishMessage.setClientId(hzMsg.getClientId());
server.internalPublish(publishMessage); server.internalPublish(publishMessage);
} }
} catch (Exception ex) { } catch (Exception ex) {