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

Refactoring, dropped unuseful ServerChannel abstraction in favor on Netty's Channel

Browse Source
This commit is contained in:
andsel
2016-01-24 19:21:24 +01:00
parent beb556c73c
commit 7b1db3e1db
13 changed files with 164 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/ServerChannel.java
-35
View File
@@ -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<Object> key);
void setAttribute(AttributeKey<Object> key, Object value);
void setIdleTime(int idleTime);
void close(boolean immediately);
void write(Object value);
}
broker/src/main/java/io/moquette/server/netty/NettyChannel.java
+1 -2
View File
@@ -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;
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+8 -8
View File
@@ -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();
broker/src/main/java/io/moquette/server/netty/NettyUtils.java
+4 -4
View File
@@ -27,13 +27,13 @@ import io.netty.util.AttributeKey;
*/
public class NettyUtils {
public static Object getAttribute(ChannelHandlerContext channel, AttributeKey<Object> key) {
Attribute<Object> attr = channel.attr(key);
public static Object getAttribute(ChannelHandlerContext ctx, AttributeKey<Object> key) {
Attribute<Object> attr = ctx.channel().attr(key);
return attr.get();
}
public static void setAttribute(ChannelHandlerContext channel, AttributeKey<Object> key, Object value) {
Attribute<Object> attr = channel.attr(key);
public static void setAttribute(ChannelHandlerContext ctx, AttributeKey<Object> key, Object value) {
Attribute<Object> attr = ctx.channel().attr(key);
attr.set(value);
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+24 -24
View File
@@ -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) {
broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java
-3
View File
@@ -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);
broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java
+1 -1
View File
@@ -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);
broker/src/test/java/io/moquette/spi/impl/DummyChannel.java
-78
View File
@@ -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<Object, Object> m_attributes = new HashMap<>();
public Object getAttribute(AttributeKey<Object> key) {
return m_attributes.get(key);
}
public void setAttribute(AttributeKey<Object> 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;
}
}
broker/src/test/java/io/moquette/spi/impl/DummyChannelHandlerContext.java
-7
View File
@@ -1,7 +0,0 @@
package io.moquette.spi.impl;
/**
* Created by andrea on 6/7/15.
*/
public class DummyChannelHandlerContext {
}
broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java
+1 -2
View File
@@ -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<Object, Object> m_attributes = new HashMap<>();
broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java
+50
View File
@@ -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());
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+42 -62
View File
@@ -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<InterceptHandler> 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<QOSType> qosSubAcked = ((SubAckMessage) m_session.getReceivedMessage()).types();
Object recvSubAckMessage = m_session.readOutbound();
assertTrue(recvSubAckMessage instanceof SubAckMessage);
List<QOSType> 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);
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+33 -30
View File
@@ -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));