diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 58ea792e..1d19009a 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -15,7 +15,7 @@ public interface IMessaging { * notification also after all the current subscriber to the given topic has * been notified. */ - void publish(String topic, Object message, byte qos, boolean retain) ; + void publish(String topic, byte[] message, QOSType qos, boolean retain); /** * Subscribe a client to a specified topic with a defined level diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java new file mode 100644 index 00000000..63676fe4 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java @@ -0,0 +1,13 @@ +package org.dna.mqtt.moquette.messaging.spi; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; + +/** + * + * @author andrea + */ +public interface INotifier { + + public void notify(String clientId, String topic, QOSType qOSType, byte[] payload); + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 9cee1561..b75b7707 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.util.ArrayList; import java.util.List; import org.dna.mqtt.moquette.messaging.spi.IMessaging; +import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** @@ -10,19 +11,36 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; * @author andrea */ public class SimpleMessaging implements IMessaging { - + //TODO is poorly performace on concurrent load private List subscriptions = new ArrayList(); + + private INotifier m_notifier; + + public void setNotifier(INotifier notifier) { + m_notifier = notifier; + } - public void publish(String topic, Object message, byte qos, boolean retain) { - throw new UnsupportedOperationException("Not supported yet."); + public void publish(String topic, byte[] message, QOSType qos, boolean retain) { + //TODO for this moment no retain management and no qos > 0 + //trivial implementation + QOSType defQos = QOSType.MOST_ONE; + synchronized(subscriptions) { + for (Subscription sub : subscriptions) { + if (sub.match(topic)) { + m_notifier.notify(sub.clientId, topic, defQos, message); + } + } + } } public void subscribe(String clientId, String topic, QOSType qos) { Subscription newSubscription = new Subscription(clientId, topic, qos); - if (subscriptions.contains(newSubscription)) { - return; + synchronized(subscriptions) { + if (subscriptions.contains(newSubscription)) { + return; + } + subscriptions.add(newSubscription); } - subscriptions.add(newSubscription); } protected List getSubscriptions() { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java index 4c484348..da4b27d4 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java @@ -62,4 +62,11 @@ public class Subscription { hash = 37 * hash + (this.topic != null ? this.topic.hashCode() : 0); return hash; } + + /** + * Trivial match method + */ + boolean match(String topic) { + return this.topic.equals(topic); + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 296fd75d..69f7803e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -1,5 +1,7 @@ package org.dna.mqtt.moquette.server; +import org.dna.mqtt.moquette.messaging.spi.INotifier; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import java.util.HashMap; import java.util.Map; @@ -18,7 +20,7 @@ import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; * * @author andrea */ -public class MQTTHandler extends IoHandlerAdapter { +public class MQTTHandler extends IoHandlerAdapter implements INotifier { protected static final String ATTR_CLIENTID = "ClientID"; private static final Logger LOG = Logger.getLogger(MQTTHandler.class.getName()); @@ -37,6 +39,8 @@ public class MQTTHandler extends IoHandlerAdapter { handleConnect(session, (ConnectMessage) msg); case SUBSCRIBE: handleSubscribe(session, (SubscribeMessage) msg); + case PUBLISH: + handlePublish(session, (PublishMessage) msg); } } @@ -74,7 +78,9 @@ public class MQTTHandler extends IoHandlerAdapter { //Handle will flag if (msg.isWillFlag()) { - m_messaging.publish(msg.getWillTopic(), msg.getWillMessage(), msg.getWillQos(), msg.isWillRetain()); + QOSType willQos = QOSType.values()[msg.getWillQos()]; + m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(), + willQos, msg.isWillRetain()); } //handle user authentication @@ -114,7 +120,12 @@ public class MQTTHandler extends IoHandlerAdapter { } session.write(ackMessage); } - + + protected void handlePublish(IoSession session, PublishMessage message) { + m_messaging.publish(message.getTopicName(), message.getPayload(), + message.getQos(), message.isRetainFlag()); + } + @Override public void sessionIdle(IoSession session, IdleStatus status) { session.close(false); @@ -127,4 +138,12 @@ public class MQTTHandler extends IoHandlerAdapter { public void setAuthenticator(IAuthenticator authenticator) { m_authenticator = authenticator; } + + public void notify(String clientId, String topic, QOSType qOSType, byte[] payload) { + PublishMessage pubMessage = new PublishMessage(); + pubMessage.setTopicName(topic); + pubMessage.setQos(qOSType); + pubMessage.setPayload(payload); + m_clientIDs.get(clientId).getSession().write(pubMessage); + } } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index c94ca0ec..e6904143 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -1,9 +1,12 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import org.mockito.ArgumentCaptor; import java.util.List; +import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.junit.Test; import static org.junit.Assert.*; +import static org.mockito.Mockito.*; /** * @@ -68,4 +71,23 @@ public class SimpleMessagingTest { } return count; } + + + @Test + public void testPublish() { + String clientID = "FAKE_123"; + String topic = "/news"; + SimpleMessaging messaging = new SimpleMessaging(); + messaging.subscribe(clientID, topic, QOSType.MOST_ONE); + INotifier notifier = mock(INotifier.class); + messaging.setNotifier(notifier); + + //Exercise + messaging.publish(topic, "Hello".getBytes(), QOSType.MOST_ONE, false); + + //Verify + ArgumentCaptor argument = ArgumentCaptor.forClass(byte[].class); + verify(notifier).notify(eq(clientID), eq(topic), any(QOSType.class), argument.capture()); + assertEquals("Hello", new String(argument.getValue())); + } } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index ca872c95..22a59916 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -85,7 +85,8 @@ public class MQTTHandlerTest { //Verify assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); - verify(mockedMessaging).publish(eq("topic"), eq("Topic message"), anyByte(), anyBoolean()); + verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), + any(AbstractMessage.QOSType.class), anyBoolean()); }