From 8cc588d7669c8d731a907ab6bc2f5252366ca672 Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 22 Jul 2016 13:19:16 +0200 Subject: [PATCH] Simplyfied creation of HazelcastMsg from InterceptPublishMessage --- .../HazelcastInterceptHandler.java | 3 +-- .../io/moquette/interception/HazelcastMsg.java | 18 +++++++----------- .../io/moquette/server/HazelcastListener.java | 13 +++++++------ 3 files changed, 15 insertions(+), 19 deletions(-) diff --git a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java index 7792a424..ce21d926 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java @@ -23,8 +23,7 @@ public class HazelcastInterceptHandler extends AbstractInterceptHandler { public void onPublish(InterceptPublishMessage msg) { LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array())); ITopic topic = hz.getTopic("moquette"); - HazelcastMsg hazelcastMsg = new HazelcastMsg(msg.getClientID(), msg.getTopicName(), msg.getQos().byteValue(), - msg.getPayload().array(), msg.getUsername()); + HazelcastMsg hazelcastMsg = new HazelcastMsg(msg); topic.publish(hazelcastMsg); } diff --git a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java index 4d400764..92ff4235 100644 --- a/broker/src/main/java/io/moquette/interception/HazelcastMsg.java +++ b/broker/src/main/java/io/moquette/interception/HazelcastMsg.java @@ -1,5 +1,7 @@ package io.moquette.interception; +import io.moquette.interception.messages.InterceptPublishMessage; + import java.io.Serializable; /** @@ -8,17 +10,15 @@ import java.io.Serializable; public class HazelcastMsg implements Serializable { private final String clientId; - private final String userName; private final byte qos; private final byte[] payload; private final String topic; - public HazelcastMsg(String clientId, String topic, byte qos, byte[] payload, String userName) { - this.clientId = clientId; - this.qos = qos; - this.payload = payload; - this.userName = userName; - this.topic = topic; + public HazelcastMsg(InterceptPublishMessage msg) { + this.clientId = msg.getClientID(); + this.topic = msg.getTopicName(); + this.qos = msg.getQos().byteValue(); + this.payload = msg.getPayload().array(); } public String getClientId() { @@ -33,10 +33,6 @@ public class HazelcastMsg implements Serializable { return payload; } - public String getUserName() { - return userName; - } - public String getTopic() { return topic; } diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java index 13738441..7af2dc51 100644 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ b/broker/src/main/java/io/moquette/server/HazelcastListener.java @@ -26,14 +26,15 @@ public class HazelcastListener implements MessageListener { public void onMessage(Message msg) { try { if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) { - LOG.info("{} received from hazelcast for topic {} message: {}", msg.getMessageObject().getClientId(), - msg.getMessageObject().getTopic(), msg.getMessageObject().getPayload()); + HazelcastMsg hzMsg = msg.getMessageObject(); + LOG.info("{} received from hazelcast for topic {} message: {}", hzMsg.getClientId(), + hzMsg.getTopic(), hzMsg.getPayload()); PublishMessage publishMessage = new PublishMessage(); - publishMessage.setTopicName(msg.getMessageObject().getTopic()); - publishMessage.setQos(AbstractMessage.QOSType.valueOf(msg.getMessageObject().getQos())); - publishMessage.setPayload(ByteBuffer.wrap(msg.getMessageObject().getPayload())); + publishMessage.setTopicName(hzMsg.getTopic()); + publishMessage.setQos(AbstractMessage.QOSType.valueOf(hzMsg.getQos())); + publishMessage.setPayload(ByteBuffer.wrap(hzMsg.getPayload())); publishMessage.setLocal(false); - publishMessage.setClientId(msg.getMessageObject().getClientId()); + publishMessage.setClientId(hzMsg.getClientId()); server.internalPublish(publishMessage); } } catch (Exception ex) {