diff --git a/broker/src/main/java/io/moquette/server/ServerChannel.java b/broker/src/main/java/io/moquette/server/ServerChannel.java deleted file mode 100644 index 6d16ee21..00000000 --- a/broker/src/main/java/io/moquette/server/ServerChannel.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * Copyright (c) 2012-2015 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ -package io.moquette.server; - -import io.netty.util.AttributeKey; - -/** - * - * @author andrea - */ -public interface ServerChannel { - - Object getAttribute(AttributeKey key); - - void setAttribute(AttributeKey key, Object value); - - void setIdleTime(int idleTime); - - void close(boolean immediately); - - void write(Object value); -} diff --git a/broker/src/main/java/io/moquette/server/netty/NettyChannel.java b/broker/src/main/java/io/moquette/server/netty/NettyChannel.java index d0041b4a..222eaf33 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyChannel.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyChannel.java @@ -20,13 +20,12 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.handler.timeout.IdleStateHandler; import io.netty.util.Attribute; import io.netty.util.AttributeKey; -import io.moquette.server.ServerChannel; /** * * @author andrea */ -public class NettyChannel implements ServerChannel { +public class NettyChannel { private ChannelHandlerContext m_channel; diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index b1fa11a0..cc9db83e 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -52,28 +52,28 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { m_processor.processConnect(ctx.channel(), (ConnectMessage) msg); break; case SUBSCRIBE: - m_processor.processSubscribe(new NettyChannel(ctx), (SubscribeMessage) msg); + m_processor.processSubscribe(ctx.channel(), (SubscribeMessage) msg); break; case UNSUBSCRIBE: - m_processor.processUnsubscribe(new NettyChannel(ctx), (UnsubscribeMessage) msg); + m_processor.processUnsubscribe(ctx.channel(), (UnsubscribeMessage) msg); break; case PUBLISH: - m_processor.processPublish(new NettyChannel(ctx), (PublishMessage) msg); + m_processor.processPublish(ctx.channel(), (PublishMessage) msg); break; case PUBREC: - m_processor.processPubRec(new NettyChannel(ctx), (PubRecMessage) msg); + m_processor.processPubRec(ctx.channel(), (PubRecMessage) msg); break; case PUBCOMP: - m_processor.processPubComp(new NettyChannel(ctx), (PubCompMessage) msg); + m_processor.processPubComp(ctx.channel(), (PubCompMessage) msg); break; case PUBREL: - m_processor.processPubRel(new NettyChannel(ctx), (PubRelMessage) msg); + m_processor.processPubRel(ctx.channel(), (PubRelMessage) msg); break; case DISCONNECT: - m_processor.processDisconnect(new NettyChannel(ctx)); + m_processor.processDisconnect(ctx.channel()); break; case PUBACK: - m_processor.processPubAck(new NettyChannel(ctx), (PubAckMessage) msg); + m_processor.processPubAck(ctx.channel(), (PubAckMessage) msg); break; case PINGREQ: PingRespMessage pingResp = new PingRespMessage(); diff --git a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java index 8ce3e341..2611eb3c 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java @@ -27,13 +27,13 @@ import io.netty.util.AttributeKey; */ public class NettyUtils { - public static Object getAttribute(ChannelHandlerContext channel, AttributeKey key) { - Attribute attr = channel.attr(key); + public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey key) { + Attribute attr = ctx.channel().attr(key); return attr.get(); } - public static void setAttribute(ChannelHandlerContext channel, AttributeKey key, Object value) { - Attribute attr = channel.attr(key); + public static void setAttribute(ChannelHandlerContext ctx, AttributeKey key, Object value) { + Attribute attr = ctx.channel().attr(key); attr.set(value); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 4a03d6d7..d14906aa 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -21,7 +21,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import io.moquette.server.ConnectionDescriptor; -import io.moquette.server.ServerChannel; import io.moquette.server.netty.NettyChannel; import io.moquette.spi.ClientSession; import io.moquette.spi.IMatchingCondition; @@ -142,7 +141,7 @@ public class ProtocolProcessor { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); LOG.warn("processConnect sent bad proto ConnAck"); - session.write(badProto); + session.writeAndFlush(badProto); session.close(); return; } @@ -150,7 +149,7 @@ public class ProtocolProcessor { if (msg.getClientID() == null || msg.getClientID().length() == 0) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); - session.write(okResp); + session.writeAndFlush(okResp); m_interceptor.notifyClientConnected(msg); return; } @@ -166,6 +165,7 @@ public class ProtocolProcessor { } if (!m_authenticator.checkValid(msg.getUsername(), pwd)) { failedCredentials(session); + session.close(); return; } session.attr(NettyChannel.ATTR_KEY_USERNAME).set(msg.getUsername()); @@ -217,7 +217,7 @@ public class ProtocolProcessor { if (!msg.isCleanSession() && isSessionAlreadyStored) { okResp.setSessionPresent(true); } - session.write(okResp); + session.writeAndFlush(okResp); m_interceptor.notifyClientConnected(msg); if (!isSessionAlreadyStored) { @@ -247,7 +247,7 @@ public class ProtocolProcessor { private void failedCredentials(Channel session) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); - session.write(okResp); + session.writeAndFlush(okResp); } /** @@ -270,8 +270,8 @@ public class ProtocolProcessor { } } - public void processPubAck(ServerChannel session, PubAckMessage msg) { - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + public void processPubAck(Channel session, PubAckMessage msg) { + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); int messageID = msg.getMessageID(); //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); @@ -535,14 +535,14 @@ public class ProtocolProcessor { LOG.debug("topic {} doesn't have read credentials", topic); return; } - session.write(pubMessage); + session.writeAndFlush(pubMessage); //Or writeAdnFlush but flush is costly!! } private void sendPubRec(String clientID, int messageID) { LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); - m_clientIDs.get(clientID).session.write(pubRecMessage); + m_clientIDs.get(clientID).session.writeAndFlush(pubRecMessage); } private void sendPubAck(String clientId, int messageID) { @@ -558,7 +558,7 @@ public class ProtocolProcessor { if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } - m_clientIDs.get(clientId).session.write(pubAckMessage); + m_clientIDs.get(clientId).session.writeAndFlush(pubAckMessage); } catch(Throwable t) { LOG.error(null, t); } @@ -568,8 +568,8 @@ public class ProtocolProcessor { * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish * to all interested subscribers. * */ - public void processPubRel(ServerChannel session, PubRelMessage msg) { - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + public void processPubRel(Channel session, PubRelMessage msg) { + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); @@ -594,11 +594,11 @@ public class ProtocolProcessor { PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); - m_clientIDs.get(clientID).session.write(pubCompMessage); + m_clientIDs.get(clientID).session.writeAndFlush(pubCompMessage); } - public void processPubRec(ServerChannel session, PubRecMessage msg) { - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + public void processPubRec(Channel session, PubRecMessage msg) { + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); int messageID = msg.getMessageID(); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); @@ -611,11 +611,11 @@ public class ProtocolProcessor { pubRelMessage.setMessageID(messageID); pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE); - session.write(pubRelMessage); + session.writeAndFlush(pubRelMessage); } - public void processPubComp(ServerChannel session, PubCompMessage msg) { - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); + public void processPubComp(Channel session, PubCompMessage msg) { + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); int messageID = msg.getMessageID(); LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory @@ -624,15 +624,15 @@ public class ProtocolProcessor { targetSession.secondPhaseAcknowledged(messageID); } - public void processDisconnect(ServerChannel session) throws InterruptedException { - String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION); + public void processDisconnect(Channel session) throws InterruptedException { + String clientID = (String) session.attr(NettyChannel.ATTR_KEY_CLIENTID).get(); + boolean cleanSession = (Boolean) session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).get(); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); clientSession.disconnect(); m_clientIDs.remove(clientID); - session.close(true); + session.close(); //cleanup the will store m_willStore.remove(clientID); @@ -691,7 +691,7 @@ public class ProtocolProcessor { ackMessage.setMessageID(messageID); LOG.info("replying with UnsubAck to MSG ID {}", messageID); - session.write(ackMessage); + session.writeAndFlush(ackMessage); } public void processSubscribe(Channel session, SubscribeMessage msg) { @@ -721,7 +721,7 @@ public class ProtocolProcessor { if (LOG.isTraceEnabled()) { LOG.trace("subscription tree {}", subscriptions.dumpTree()); } - session.write(ackMessage); + session.writeAndFlush(ackMessage); //fire the publish for(Subscription subscription : newSubscriptions) { diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index dc16f54b..22e6a613 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -33,9 +33,6 @@ import java.util.concurrent.TimeUnit; import static org.junit.Assert.*; -/** - * Created by andrea on 08/12/15. - */ public class ServerIntegrationEmbeddedPublishTest { private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationEmbeddedPublishTest.class); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index c56eb722..422399ca 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -451,7 +451,7 @@ public class ServerIntegrationPahoTest { clientXA.subscribe("topic", 0); MqttClient clientXB = createClient("publisher", "X"); - clientXB.publish("topic", "Hello".getBytes(), 2, true); + clientXB.publish("topic", "Hello".getBytes(), 2, false); TestCallback cbSubscriber2 = new TestCallback(); MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2); diff --git a/broker/src/test/java/io/moquette/spi/impl/DummyChannel.java b/broker/src/test/java/io/moquette/spi/impl/DummyChannel.java deleted file mode 100644 index 5090b05a..00000000 --- a/broker/src/test/java/io/moquette/spi/impl/DummyChannel.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Copyright (c) 2012-2015 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ -package io.moquette.spi.impl; - -import io.netty.channel.embedded.EmbeddedChannel; -import io.netty.util.AttributeKey; -import io.moquette.proto.messages.AbstractMessage; -import io.moquette.proto.messages.ConnAckMessage; - -import java.util.HashMap; -import java.util.Map; - -/** - * - * @author andrea - */ -public class DummyChannel extends EmbeddedChannel { - - AbstractMessage m_receivedMessage; - - byte m_returnCode; - - private boolean m_channelClosed = false; - - private Map m_attributes = new HashMap<>(); - - public Object getAttribute(AttributeKey key) { - return m_attributes.get(key); - } - - public void setAttribute(AttributeKey key, Object value) { - m_attributes.put(key, value); - } - - public void setIdleTime(int idleTime) { - } - - public void close(boolean immediately) { - m_channelClosed = true; - } - - boolean isClosed() { - return this.m_channelClosed; - } - -// public void write(Object value) { -// try { -// m_receivedMessage = (AbstractMessage) value; -// if (m_receivedMessage instanceof ConnAckMessage) { -// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; -// m_returnCode = buf.getReturnCode(); -// } -// } catch (Exception ex) { -// throw new AssertionError("Wrong return code"); -// } -// } - - public byte getReturnCode() { - return m_returnCode; - } - - public AbstractMessage getReceivedMessage() { - return m_receivedMessage; - } -} diff --git a/broker/src/test/java/io/moquette/spi/impl/DummyChannelHandlerContext.java b/broker/src/test/java/io/moquette/spi/impl/DummyChannelHandlerContext.java deleted file mode 100644 index d35cb6c9..00000000 --- a/broker/src/test/java/io/moquette/spi/impl/DummyChannelHandlerContext.java +++ /dev/null @@ -1,7 +0,0 @@ -package io.moquette.spi.impl; - -/** - * Created by andrea on 6/7/15. - */ -public class DummyChannelHandlerContext { -} diff --git a/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java b/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java index c6a13fa0..ea6b6e8d 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java +++ b/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java @@ -16,7 +16,6 @@ import io.netty.util.AttributeKey; import io.moquette.proto.messages.AbstractMessage; -import io.moquette.server.ServerChannel; import java.util.HashMap; import java.util.Map; @@ -24,7 +23,7 @@ import java.util.Map; /** * This a synchronous channel that avoid output ring buffer from Processor */ -class MockReceiverChannel implements ServerChannel { +class MockReceiverChannel /*implements ServerChannel*/ { // byte m_returnCode; AbstractMessage m_receivedMessage; private Map m_attributes = new HashMap<>(); diff --git a/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java b/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java new file mode 100644 index 00000000..9ec5dd06 --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java @@ -0,0 +1,50 @@ +/* + * Copyright (c) 2012-2015 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.spi.impl; + +import io.moquette.proto.messages.ConnAckMessage; +import io.moquette.proto.messages.SubAckMessage; +import io.netty.channel.Channel; +import io.netty.channel.embedded.EmbeddedChannel; + +import static io.moquette.proto.messages.ConnAckMessage.CONNECTION_ACCEPTED; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * Some useful assertions used by Netty's EmbeddedChannel in tests. + */ +class NettyChannelAssertions { + + static void assertEqualsConnAck(byte expectedCode, Object connAck) { + assertTrue(connAck instanceof ConnAckMessage); + ConnAckMessage connAckMsg = (ConnAckMessage) connAck; + assertEquals(expectedCode, connAckMsg.getReturnCode()); + } + + static void assertConnAckAccepted(EmbeddedChannel channel) { + channel.flush(); + assertEqualsConnAck(CONNECTION_ACCEPTED, channel.readOutbound()); + } + + static void assertEqualsSubAck(/*byte expectedCode,*/ Object subAck) { + assertTrue(subAck instanceof SubAckMessage); + //SubAckMessage connAckMsg = (SubAckMessage) connAck; + //assertEquals(expectedCode, connAckMsg.getReturnCode()); + } + +} diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index af5056c9..84bc3d3c 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -34,9 +34,7 @@ import org.junit.Test; import java.nio.ByteBuffer; import java.util.*; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - +import static io.moquette.spi.impl.NettyChannelAssertions.assertConnAckAccepted; import static org.junit.Assert.*; import static org.mockito.Mockito.*; @@ -59,7 +57,7 @@ public class ProtocolProcessorTest { final static List EMPTY_OBSERVERS = Collections.emptyList(); final static BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); - DummyChannel m_session; + EmbeddedChannel m_session; ConnectMessage connMsg; ProtocolProcessor m_processor; @@ -73,7 +71,7 @@ public class ProtocolProcessorTest { connMsg = new ConnectMessage(); connMsg.setProtocolVersion((byte) 0x03); - m_session = new DummyChannel(); + m_session = new EmbeddedChannel(); //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); @@ -96,7 +94,7 @@ public class ProtocolProcessorTest { @Test - public void testPublish() throws InterruptedException { + public void testPublishToItself() throws InterruptedException { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); @@ -132,11 +130,11 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "FakeCLI"); + m_session.attr(NettyChannel.ATTR_KEY_USERNAME).set("FakeCLI"); m_processor.processPublish(m_session, msg); //Verify - assertNotNull(m_session.getReceivedMessage()); + assertNotNull(m_session.readOutbound()); //TODO check received message attributes } @@ -171,6 +169,7 @@ public class ProtocolProcessorTest { connectMessage.setClientID(FAKE_CLIENT_ID); connectMessage.setCleanSession(true); m_processor.processConnect(firstReceiverSession, connectMessage); + assertConnAckAccepted(firstReceiverSession); //connect the second fake subscriber EmbeddedChannel secondReceiverSession = new EmbeddedChannel(); @@ -180,7 +179,8 @@ public class ProtocolProcessorTest { connectMessage2.setCleanSession(true); connectMessage2.setCleanSession(true); m_processor.processConnect(secondReceiverSession, connectMessage2); - + assertConnAckAccepted(secondReceiverSession); + //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); buffer.rewind(); @@ -189,34 +189,34 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(false); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "FakeCLI"); + m_session.attr(NettyChannel.ATTR_KEY_USERNAME).set("FakeCLI"); m_processor.processPublish(m_session, msg); + //Verify - Thread.sleep(100); //ugly but we depend on the asynch that pull data from back disruptor - PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverSession.getMessage(); + PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverSession.readOutbound(); assertNotNull(pub2FirstSubscriber); String firstMessageContent = DebugUtils.payload2Str(pub2FirstSubscriber.getPayload()); assertEquals("Hello", firstMessageContent); - - PublishMessage pub2SecondSubscriber = (PublishMessage) secondReceiverSession.getMessage(); + + PublishMessage pub2SecondSubscriber = (PublishMessage) secondReceiverSession.readOutbound(); assertNotNull(pub2SecondSubscriber); String secondMessageContent = DebugUtils.payload2Str(pub2SecondSubscriber.getPayload()); assertEquals("Hello", secondMessageContent); } - + @Test public void testSubscribe() { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); - m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); + m_processor.processSubscribe(m_session, msg); //Verify - assertTrue(m_session.getReceivedMessage() instanceof SubAckMessage); + assertTrue(m_session.readOutbound() instanceof SubAckMessage); Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); assertTrue(subscriptions.contains(expectedSubscription)); } @@ -225,8 +225,8 @@ public class ProtocolProcessorTest { public void testDoubleSubscribe() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -246,8 +246,8 @@ public class ProtocolProcessorTest { public void testSubscribeWithBadFormattedTopic() { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); @@ -256,8 +256,9 @@ public class ProtocolProcessorTest { //Verify assertEquals(0, subscriptions.size()); - assertTrue(m_session.getReceivedMessage() instanceof SubAckMessage); - List qosSubAcked = ((SubAckMessage) m_session.getReceivedMessage()).types(); + Object recvSubAckMessage = m_session.readOutbound(); + assertTrue(recvSubAckMessage instanceof SubAckMessage); + List qosSubAcked = ((SubAckMessage)recvSubAckMessage).types(); assertEquals(1, qosSubAcked.size()); assertEquals(QOSType.FAILURE, qosSubAcked.get(0)); } @@ -270,40 +271,19 @@ public class ProtocolProcessorTest { UnsubscribeMessage msg = new UnsubscribeMessage(); msg.setMessageID(1); msg.addTopicFilter(BAD_FORMATTED_TOPIC); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_CLIENT_ID); + m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); //Exercise m_processor.processUnsubscribe(m_session, msg); //Verify - assertTrue(m_session.isClosed()); + assertFalse("If client unsubscribe with bad topic than channel must be closed", m_session.isOpen()); } @Test public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { - final CountDownLatch publishRecvSignal = new CountDownLatch(1); - m_session = new DummyChannel() { - @Override - public void write(Object value) { - try { - System.out.println("filterReceived class " + value.getClass().getName()); - if (value instanceof PublishMessage) { - m_receivedMessage = (AbstractMessage) value; - publishRecvSignal.countDown(); - } - - if (m_receivedMessage instanceof ConnAckMessage) { - ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; - m_returnCode = buf.getReturnCode(); - } - } catch (Exception ex) { - throw new AssertionError("Wrong return code"); - } - } - }; - //simulate a connect that register a clientID to an IoSession final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); @@ -330,16 +310,17 @@ public class ProtocolProcessorTest { connectMessage.setProtocolVersion((byte) 3); connectMessage.setCleanSession(true); m_processor.processConnect(m_session, connectMessage); + assertConnAckAccepted(m_session); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage pubmsg = new PublishMessage(); pubmsg.setTopicName(FAKE_TOPIC); pubmsg.setQos(QOSType.MOST_ONE); pubmsg.setPayload(buffer); pubmsg.setRetainFlag(true); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_PUBLISHER_ID); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(FAKE_PUBLISHER_ID); m_processor.processPublish(m_session, pubmsg); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); - + m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); + //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(QOSType.MOST_ONE.byteValue(), "#")); @@ -347,11 +328,10 @@ public class ProtocolProcessorTest { //Verify //wait the latch - assertTrue(publishRecvSignal.await(1, TimeUnit.SECONDS)); //no timeout - assertNotNull(m_session.getReceivedMessage()); - assertTrue(m_session.getReceivedMessage() instanceof PublishMessage); - PublishMessage pubMessage = (PublishMessage) m_session.getReceivedMessage(); - assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); + Object pubMessage = m_session.readOutbound(); + assertNotNull(pubMessage); + assertTrue(pubMessage instanceof PublishMessage); + assertEquals(FAKE_TOPIC, ((PublishMessage) pubMessage).getTopicName()); } @Test @@ -397,11 +377,11 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(true); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "Publisher"); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set("Publisher"); m_processor.processPublish(m_session, msg); //Verify no message is received - assertNull(m_session.getReceivedMessage()); + assertNull(m_session.readOutbound()); } @@ -424,11 +404,11 @@ public class ProtocolProcessorTest { msg.setQos(QOSType.MOST_ONE); msg.setPayload(buffer); msg.setRetainFlag(true); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "Publisher"); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set("Publisher"); m_processor.processPublish(m_session, msg); //Verify no message is received - assertNull(m_session.getReceivedMessage()); + assertNull(m_session.readOutbound()); } @@ -442,7 +422,7 @@ public class ProtocolProcessorTest { connMsg.setClientID("Publisher"); m_processor.processConnect(m_session, connMsg); //prepare and existing retained store - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "Publisher"); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set("Publisher"); ByteBuffer payload = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage msg = new PublishMessage(); msg.setTopicName(FAKE_TOPIC); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 85a2f8e3..04486a69 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -16,7 +16,6 @@ package io.moquette.spi.impl; import io.moquette.proto.messages.AbstractMessage; -import io.moquette.proto.messages.ConnAckMessage; import io.moquette.proto.messages.ConnectMessage; import io.moquette.proto.messages.SubscribeMessage; import io.moquette.spi.IMessagesStore; @@ -25,6 +24,7 @@ import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.server.netty.NettyChannel; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; +import io.netty.channel.embedded.EmbeddedChannel; import org.junit.Before; import org.junit.Test; @@ -32,7 +32,11 @@ import java.util.HashMap; import java.util.Map; import static io.moquette.parser.netty.Utils.VERSION_3_1_1; -import static org.junit.Assert.assertEquals; +import static io.moquette.proto.messages.ConnAckMessage.BAD_USERNAME_OR_PASSWORD; +import static io.moquette.proto.messages.ConnAckMessage.CONNECTION_ACCEPTED; +import static io.moquette.proto.messages.ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION; +import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck; +import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -42,7 +46,7 @@ import static org.junit.Assert.assertTrue; */ public class ProtocolProcessor_CONNECT_Test { - DummyChannel m_session; + EmbeddedChannel m_session; ConnectMessage connMsg; ProtocolProcessor m_processor; @@ -56,7 +60,7 @@ public class ProtocolProcessor_CONNECT_Test { connMsg = new ConnectMessage(); connMsg.setProtocolVersion((byte) 0x03); - m_session = new DummyChannel(); + m_session = new EmbeddedChannel(); //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); @@ -85,7 +89,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_session.getReturnCode()); + assertEqualsConnAck(UNNACEPTABLE_PROTOCOL_VERSION, m_session.readOutbound()); } @Test @@ -96,7 +100,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); } @Test @@ -112,7 +116,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); //TODO verify the call /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/ @@ -130,7 +134,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); } @Test @@ -144,7 +148,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.BAD_USERNAME_OR_PASSWORD, m_session.getReturnCode()); + assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); } @Test @@ -159,7 +163,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.BAD_USERNAME_OR_PASSWORD, m_session.getReturnCode()); + assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); } @Test @@ -171,7 +175,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.BAD_USERNAME_OR_PASSWORD, m_session.getReturnCode()); + assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); } @Test @@ -185,7 +189,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.BAD_USERNAME_OR_PASSWORD, m_session.getReturnCode()); + assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); } @Test @@ -197,7 +201,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processConnect(m_session, connMsg); //Verify - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); } @Test @@ -209,7 +213,7 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setUsername(ProtocolProcessorTest.TEST_USER); connMsg.setPassword(ProtocolProcessorTest.TEST_PWD); m_processor.processConnect(m_session, connMsg); - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); //create another connect same clientID but with bad credentials ConnectMessage evilClientConnMsg = new ConnectMessage(); @@ -219,17 +223,17 @@ public class ProtocolProcessor_CONNECT_Test { evilClientConnMsg.setPasswordFlag(true); evilClientConnMsg.setUsername(ProtocolProcessorTest.EVIL_TEST_USER); evilClientConnMsg.setPassword(ProtocolProcessorTest.EVIL_TEST_PWD); - DummyChannel evilSession = new DummyChannel(); + EmbeddedChannel evilSession = new EmbeddedChannel(); //Exercise m_processor.processConnect(evilSession, evilClientConnMsg); //Verify //the evil client gets a not auth notification - assertEquals(ConnAckMessage.BAD_USERNAME_OR_PASSWORD, evilSession.getReturnCode()); + assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, evilSession.readOutbound()); //the good client remains connected - assertFalse(m_session.isClosed()); - assertTrue(evilSession.isClosed()); + assertTrue(m_session.isOpen()); + assertFalse(evilSession.isOpen()); } @@ -239,8 +243,8 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setProtocolVersion(VERSION_3_1_1); connMsg.setClientID("CliID"); connMsg.setCleanSession(false); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "CliID"); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set("CliID"); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set(false); //Connect a first time m_processor.processConnect(m_session, connMsg); @@ -248,15 +252,11 @@ public class ProtocolProcessor_CONNECT_Test { m_processor.processDisconnect(m_session); //Exercise, reconnect - MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); + EmbeddedChannel firstReceiverSession = new EmbeddedChannel(); m_processor.processConnect(firstReceiverSession, connMsg); //Verify - AbstractMessage recvMsg = firstReceiverSession.getMessage(); - assertTrue(recvMsg instanceof ConnAckMessage); - ConnAckMessage connAckMsg = (ConnAckMessage) recvMsg; - assertTrue(connAckMsg.isSessionPresent()); - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, connAckMsg.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound()); } @@ -268,24 +268,27 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setClientID("CliID"); connMsg.setCleanSession(false); m_processor.processConnect(m_session, connMsg); - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); //subscribe SubscribeMessage subscribeMsg = new SubscribeMessage(); subscribeMsg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), ProtocolProcessorTest.FAKE_TOPIC)); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "CliID"); - m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_session.attr(NettyChannel.ATTR_KEY_CLIENTID).set("CliID"); + m_session.attr(NettyChannel.ATTR_KEY_CLEANSESSION).set(false); m_processor.processSubscribe(m_session, subscribeMsg); Subscription expectedSubscription = new Subscription("CliID", ProtocolProcessorTest.FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); assertTrue(subscriptions.contains(expectedSubscription)); + assertEqualsSubAck(m_session.readOutbound()); //disconnect m_processor.processDisconnect(m_session); + assertFalse(m_session.isOpen()); //reconnect clean session a false + m_session = new EmbeddedChannel(); m_processor.processConnect(m_session, connMsg); - assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); //verify that the first subscription is still preserved assertTrue(subscriptions.contains(expectedSubscription));