From e73a2a8b9eaecd96a4feadd45c988686e73692c9 Mon Sep 17 00:00:00 2001 From: andrea Date: Sat, 15 Feb 2014 12:06:03 +0100 Subject: [PATCH] Fix issue #30 rewind of message ByteBuffer --- .../messaging/spi/impl/ProtocolProcessor.java | 8 +- .../spi/impl/ProtocolProcessorTest.java | 99 ++++++++++++++++++- tools_scripts/publisher_client.groovy | 4 +- tools_scripts/subscriber_client.groovy | 11 ++- 4 files changed, 112 insertions(+), 10 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index d3f6d83e..ae57fda8 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -245,15 +245,15 @@ class ProtocolProcessor implements EventHandler { * */ private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retain, Integer messageID) { LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}, content", topic); - LOG.debug("content <{}>", new String(message.array())); if (LOG.isDebugEnabled()) { + LOG.debug("content <{}>", DebugUtils.payload2Str(message)); LOG.debug("subscription tree {}", subscriptions.dumpTree()); } for (final Subscription sub : subscriptions.matches(topic)) { if (qos.ordinal() > sub.getRequestedQos().ordinal()) { qos = sub.getRequestedQos(); } - + message.rewind(); LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopic(), qos, sub.isActive()); @@ -298,7 +298,9 @@ class ProtocolProcessor implements EventHandler { pubMessage.setPayload(message); LOG.info("send publish message to <{}> on topic <{}>", clientId, topic); - LOG.debug("content <{}>", new String(message.array())); + if (LOG.isDebugEnabled()) { + LOG.debug("content <{}>", DebugUtils.payload2Str(message)); + } if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { pubMessage.setMessageID(messageID); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index f3b01543..8147dea9 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -22,7 +22,6 @@ import org.dna.mqtt.moquette.server.ServerChannel; import static org.junit.Assert.*; import static org.mockito.Mockito.*; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; /** @@ -31,6 +30,7 @@ import org.junit.Test; */ public class ProtocolProcessorTest { final static String FAKE_CLIENT_ID = "FAKE_123"; + final static String FAKE_CLIENT_ID2 = "FAKE_456"; final static String FAKE_PUBLISHER_ID = "Publisher"; final static String FAKE_TOPIC = "/news"; @@ -76,6 +76,45 @@ public class ProtocolProcessorTest { } + class MockReceiverChannel implements ServerChannel { +// byte m_returnCode; + AbstractMessage m_receivedMessage; + private Map m_attributes = new HashMap(); + + public Object getAttribute(Object key) { + return m_attributes.get(key); + } + + public void setAttribute(Object key, Object value) { + m_attributes.put(key, value); + } + + public void setIdleTime(int idleTime) { + } + + public void close(boolean immediately) { + } + + public AbstractMessage getMessage() { + return this.m_receivedMessage; + } + +// public byte getReturnCode() { +// return this.m_returnCode; +// } + + public void write(Object value) { + try { + this.m_receivedMessage = (AbstractMessage) value; +// if (this.m_receivedMessage instanceof PublishMessage) { +// T buf = (T) this.m_receivedMessage; +// } + } catch (Exception ex) { + throw new AssertionError("Wrong return code"); + } + } + } + @Before public void setUp() throws InterruptedException { connMsg = new ConnectMessage(); @@ -151,8 +190,6 @@ public class ProtocolProcessorTest { throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); } } - - }; //simulate a connect that register a clientID to an IoSession @@ -175,6 +212,62 @@ public class ProtocolProcessorTest { //TODO check received message attributes } + @Test + public void testPublishToMultipleSubscribers() throws InterruptedException { + final Subscription subscription = new Subscription(FAKE_CLIENT_ID, + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + final Subscription subscriptionClient2 = new Subscription(FAKE_CLIENT_ID2, + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + + //subscriptions.matches(topic) redefine the method to return true + SubscriptionsStore subs = new SubscriptionsStore() { + @Override + public List matches(String topic) { + if (topic.equals(FAKE_TOPIC)) { + return Arrays.asList(subscription, subscriptionClient2); + } else { + throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic); + } + } + }; + + //simulate a connect that register a clientID to an IoSession + subs.init(m_storageService); + m_processor.init(subs, m_storageService); + + MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setProcotolVersion((byte)3); + connectMessage.setClientID(FAKE_CLIENT_ID); + connectMessage.setCleanSession(subscription.isCleanSession()); + m_processor.processConnect(firstReceiverSession, connectMessage); + + //connect the second fake subscriber + MockReceiverChannel secondReceiverSession = new MockReceiverChannel(); + ConnectMessage connectMessage2 = new ConnectMessage(); + connectMessage2.setProcotolVersion((byte)3); + connectMessage2.setClientID(FAKE_CLIENT_ID2); + connectMessage2.setCleanSession(subscription.isCleanSession()); + m_processor.processConnect(secondReceiverSession, connectMessage2); + + //Exercise + ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI", null); + m_processor.processPublish(pubEvt); + + //Verify + Thread.sleep(100); //ugly but we dependend on the asynch that pull data from back disruptor + PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverSession.getMessage(); + assertNotNull(pub2FirstSubscriber); + String firstMessageContent = DebugUtils.payload2Str(pub2FirstSubscriber.getPayload()); + assertEquals("Hello", firstMessageContent); + + PublishMessage pub2SecondSubscriber = (PublishMessage) secondReceiverSession.getMessage(); + assertNotNull(pub2SecondSubscriber); + String secondMessageContent = DebugUtils.payload2Str(pub2SecondSubscriber.getPayload()); + assertEquals("Hello", secondMessageContent); + } + @Test public void testSubscribe() { //Exercise diff --git a/tools_scripts/publisher_client.groovy b/tools_scripts/publisher_client.groovy index cde9d2c7..98915e00 100644 --- a/tools_scripts/publisher_client.groovy +++ b/tools_scripts/publisher_client.groovy @@ -17,7 +17,7 @@ MqttMessage message = new MqttMessage('Hello world!!'.bytes) message.setQos(0) print "publishing.." long startTime = System.currentTimeMillis() -(1..100000).each { +(1..10).each { client.publish('topic', 'Hello world!!'.bytes, 0, false) } client.publish('/exit', 'Exit'.bytes, 0, false) @@ -25,4 +25,4 @@ long stopTime = System.currentTimeMillis() long spentTime = stopTime -startTime println "published in ${spentTime} ms" client.disconnect() -println "disconnected" \ No newline at end of file +println "disconnected" diff --git a/tools_scripts/subscriber_client.groovy b/tools_scripts/subscriber_client.groovy index 1f55d587..a8d11974 100644 --- a/tools_scripts/subscriber_client.groovy +++ b/tools_scripts/subscriber_client.groovy @@ -11,7 +11,6 @@ import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit - class SubscriberCallback implements MqttCallback { int m_numReceived = 0 @@ -42,6 +41,7 @@ class SubscriberCallback implements MqttCallback { client.disconnect() m_latch.countDown() } else { + println "-received ${message} on ${topic} with QoS ${message.qos}" m_numReceived++ } } @@ -51,11 +51,18 @@ class SubscriberCallback implements MqttCallback { } } +if (args.size() < 2) { + println "Usage subscriber " + return +} + String host = args[0] +int qos = args[1] as int String tmpDir = System.getProperty("java.io.tmpdir") MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir) -MqttClient client = new MqttClient("tcp://${host}:1883", "SubscriberClient", dataStore) +int rnd = (Math.random() * 100) as int +MqttClient client = new MqttClient("tcp://${host}:1883", "SubscriberClient${rnd}", dataStore) def callback = new SubscriberCallback() client.callback = callback client.connect()