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

Implemented the raw connect message handling

Browse Source
This commit is contained in:
selva.andre
2012-01-22 18:02:00 +01:00
parent aadff9c9bd
commit 463aa25693
5 changed files with 173 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+20
View File
@@ -0,0 +1,20 @@
package org.dna.mqtt.moquette.messaging.spi;
/**
* Interface to the underling messaging system used to publish, subscribe.
*
* @author andrea
*/
public class IMessaging {
/**
* Publish a message on the messaging system. The qos is the level of quality
* the retain flag is true iff the message should be persisted for future
* notification also after all the current subscriber to the given topic has
* been notified.
*/
public void publish(String topic, Object message, byte qos, boolean retain) {
throw new UnsupportedOperationException("Not yet implemented");
}
}
src/main/java/org/dna/mqtt/moquette/server/IAuthenticator.java
+11
View File
@@ -0,0 +1,11 @@
package org.dna.mqtt.moquette.server;
/**
* username and password checker
*
* @author andrea
*/
public interface IAuthenticator {
boolean checkValid(String username, String password);
}
src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+58 -10
View File
@@ -1,9 +1,12 @@
package org.dna.mqtt.moquette.server;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
@@ -15,19 +18,23 @@ import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
*/
public class MQTTHandler extends IoHandlerAdapter {
private static final Logger LOG = Logger.getLogger(MQTTHandler.class.getName());
Map<String, IoSession> m_clientIDs = new HashMap<String, IoSession>();
private IMessaging m_messaging;
private IAuthenticator m_authenticator;
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
AbstractMessage msg = (AbstractMessage) message;
LOG.fine("Received a message of type " + msg.getMessageType());
switch (msg.getMessageType()) {
case CONNECT:
handleConnect(session, (ConnectMessage) msg);
}
}
protected void handleConnect(IoSession session, ConnectMessage connectMessage) {
if (connectMessage.getProcotolVersion() != 0x03) {
protected void handleConnect(IoSession session, ConnectMessage msg) {
if (msg.getProcotolVersion() != 0x03) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
session.write(badProto);
@@ -35,20 +42,61 @@ public class MQTTHandler extends IoHandlerAdapter {
return;
}
//if an old client with the same ID already exists close its session.s
if (m_clientIDs.containsKey(connectMessage.getClientID())) {
m_clientIDs.get(connectMessage.getClientID()).close(false);
if (msg.getClientID().length() > 23) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
session.write(okResp);
return;
}
m_clientIDs.put(connectMessage.getClientID(), session);
session.setAttribute("keepAlive", connectMessage.getKeepAlive());
session.setAttribute("cleanSession", connectMessage.isCleanSession());
//if an old client with the same ID already exists close its session.
if (m_clientIDs.containsKey(msg.getClientID())) {
m_clientIDs.get(msg.getClientID()).close(false);
}
m_clientIDs.put(msg.getClientID(), session);
//TODO Handle will flag
//TODO handle user authentication
int keepAlive = msg.getKeepAlive();
session.setAttribute("keepAlive", keepAlive);
session.setAttribute("cleanSession", msg.isCleanSession());
session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
//Handle will flag
if (msg.isWillFlag()) {
m_messaging.publish(msg.getWillTopic(), msg.getWillMessage(), msg.getWillQos(), msg.isWillRetain());
}
//handle user authentication
if (msg.isUserFlag()) {
String pwd = null;
if (msg.isPasswordFlag()) {
pwd = msg.getPassword();
}
if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.write(okResp);
return;
}
}
//TODO handle clean session flag
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
session.write(okResp);
}
@Override
public void sessionIdle(IoSession session, IdleStatus status) {
session.close(false);
}
public void setMessaging(IMessaging messaging) {
m_messaging = messaging;
}
public void setAuthenticator(IAuthenticator authenticator) {
m_authenticator = authenticator;
}
}
src/main/java/org/dna/mqtt/moquette/server/Server.java
+47
View File
@@ -0,0 +1,47 @@
package org.dna.mqtt.moquette.server;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.logging.Logger;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.dna.mqtt.moquette.proto.ConnAckEncoder;
import org.dna.mqtt.moquette.proto.ConnectDecoder;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
/**
* Launch a configured version of the server.
* @author andrea
*/
public class Server {
private static final Logger LOG = Logger.getLogger(Server.class.getName());
public static final int PORT = 9191;
public static final int DEFAULT_CONNECT_TIMEOUT = 10;
public static void main(String[] args) throws IOException {
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
decoder.addMessageDecoder(new ConnectDecoder());
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());
IoAcceptor acceptor = new NioSocketAcceptor();
acceptor.getFilterChain().addLast( "logger", new LoggingFilter() );
acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
acceptor.setHandler( new MQTTHandler() );
acceptor.getSessionConfig().setReadBufferSize( 2048 );
acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT );
acceptor.bind( new InetSocketAddress(PORT) );
LOG.info("Server binded");
}
}
src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java
+31 -1
View File
@@ -31,7 +31,7 @@ public class MQTTHandlerTest {
ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage();
assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, buf.getReturnCode());
} catch (Exception ex) {
throw new AssertionError("Wrong return code");
fail("Wrong return code");
}
}
});
@@ -39,4 +39,34 @@ public class MQTTHandlerTest {
//Exercise
h.handleConnect(session, connMsg);
}
@Test
public void testConnect_badClientID() {
MQTTHandler h = new MQTTHandler();
ConnectMessage connMsg = new ConnectMessage();
connMsg.setProcotolVersion((byte) 0x03);
connMsg.setClientID("extremely_long_clientID_grtaer_than_23");
IoSession session = new DummySession();
session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
@Override
public void filterWrite(NextFilter nextFilter, IoSession session,
WriteRequest writeRequest) throws Exception {
try {
ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage();
assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, buf.getReturnCode());
} catch (Exception ex) {
fail("Wrong return code");
}
}
});
//Exercise
h.handleConnect(session, connMsg);
}
}