From 8ade5f68aa9e37ad09b94ac8b655ccc5c5603b3f Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 13 Sep 2013 10:24:04 +0200 Subject: [PATCH] Extracted front MINA connector --- .../org/dna/mqtt/moquette/server/Server.java | 98 +------------- .../mqtt/moquette/server/ServerAcceptor.java | 15 +++ .../moquette/server/mina/MinaAcceptor.java | 122 ++++++++++++++++++ 3 files changed, 142 insertions(+), 93 deletions(-) create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index ec0b6392..d4872f78 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -1,21 +1,9 @@ package org.dna.mqtt.moquette.server; -import org.dna.mqtt.moquette.proto.messages.PubCompMessage; import java.io.File; import java.io.IOException; -import java.net.InetSocketAddress; -import org.apache.mina.core.service.IoAcceptor; -import org.apache.mina.core.service.IoServiceStatistics; -import org.apache.mina.core.session.IdleStatus; -import org.apache.mina.core.session.IoSession; -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.transport.socket.nio.NioSocketAcceptor; -import org.dna.mqtt.commons.Constants; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; -import org.dna.mqtt.moquette.proto.*; -import org.dna.mqtt.moquette.proto.messages.*; +import org.dna.mqtt.moquette.server.mina.MinaAcceptor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** @@ -26,17 +14,11 @@ public class Server { private static final Logger LOG = LoggerFactory.getLogger(Server.class); -// public static final int PORT = 1883; -// public static final int DEFAULT_CONNECT_TIMEOUT = 10; public static final String STORAGE_FILE_PATH = System.getProperty("user.home") + File.separator + "moquette_store.hawtdb"; - //TODO one notifier per outbound queue, else no serialization of message flow is granted!! - //private static final int NOTIFIER_POOL_SIZE = 1; - private IoAcceptor m_acceptor; + private ServerAcceptor m_acceptor; SimpleMessaging messaging; -// Thread messagingEventLoop; - //private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/; public static void main(String[] args) throws IOException { final Server server = new Server(); @@ -52,87 +34,17 @@ public class Server { } public void startServer() throws IOException { - DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); - decoder.addMessageDecoder(new ConnectDecoder()); - decoder.addMessageDecoder(new PublishDecoder()); - decoder.addMessageDecoder(new PubAckDecoder()); - decoder.addMessageDecoder(new PubRelDecoder()); - decoder.addMessageDecoder(new PubRecDecoder()); - decoder.addMessageDecoder(new PubCompDecoder()); - decoder.addMessageDecoder(new SubscribeDecoder()); - decoder.addMessageDecoder(new UnsubscribeDecoder()); - decoder.addMessageDecoder(new DisconnectDecoder()); - decoder.addMessageDecoder(new PingReqDecoder()); - - DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); -// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); - encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); - encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); - encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder()); - encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder()); - encoder.addMessageEncoder(PubRecMessage.class, new PubRecEncoder()); - encoder.addMessageEncoder(PubCompMessage.class, new PubCompEncoder()); - encoder.addMessageEncoder(PubRelMessage.class, new PubRelEncoder()); - encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); - encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder()); - - m_acceptor = new NioSocketAcceptor(); - - m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") ); - m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); - - MQTTHandler handler = new MQTTHandler(); messaging = SimpleMessaging.getInstance(); messaging.init(); - //TODO fix this hugly wiring - handler.setMessaging(messaging); -// messaging.setNotifier(handler); -// messagingEventLoop = new Thread(messaging); -// messagingEventLoop.setName("Event Loop" + System.currentTimeMillis()); -// messagingEventLoop.start(); - - m_acceptor.setHandler(handler); - ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); - ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true); - m_acceptor.getSessionConfig().setReadBufferSize( 2048 ); - m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT ); - m_acceptor.getStatistics().setThroughputCalculationInterval(10); - m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis()); - m_acceptor.bind( new InetSocketAddress(Constants.PORT) ); - LOG.info("Server binded"); - + m_acceptor = new MinaAcceptor(); + m_acceptor.initialize(messaging); } public void stopServer() { LOG.info("Server stopping..."); - messaging.stop(); - //messaging.stop(); -// messagingEventLoop.interrupt(); -// LOG.info("shutting down evet loop"); -// try { -// messagingEventLoop.join(); -// } catch (InterruptedException ex) { -// LOG.error(null, ex); -// } - - /*m_notifierPool.shutdown();*/ - - //log statistics - IoServiceStatistics statistics = m_acceptor.getStatistics(); - statistics.updateThroughput(System.currentTimeMillis()); - System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput())); - System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput())); - - for(IoSession session: m_acceptor.getManagedSessions().values()) { - if(session.isConnected() && !session.isClosing()){ - session.close(false); - } - } - - m_acceptor.unbind(); - m_acceptor.dispose(); + m_acceptor.close(); LOG.info("Server stopped"); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java new file mode 100644 index 00000000..7ce5e549 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java @@ -0,0 +1,15 @@ +package org.dna.mqtt.moquette.server; + +import java.io.IOException; +import org.dna.mqtt.moquette.messaging.spi.IMessaging; + +/** + * + * @author andrea + */ +public interface ServerAcceptor { + + void initialize(IMessaging messaging) throws IOException; + + void close(); +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java new file mode 100644 index 00000000..324586aa --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java @@ -0,0 +1,122 @@ +package org.dna.mqtt.moquette.server.mina; + +import java.io.IOException; +import java.net.InetSocketAddress; +import org.apache.mina.core.service.IoAcceptor; +import org.apache.mina.core.service.IoServiceStatistics; +import org.apache.mina.core.session.IdleStatus; +import org.apache.mina.core.session.IoSession; +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.transport.socket.nio.NioSocketAcceptor; +import org.dna.mqtt.commons.Constants; +import org.dna.mqtt.moquette.messaging.spi.IMessaging; +import org.dna.mqtt.moquette.proto.ConnAckEncoder; +import org.dna.mqtt.moquette.proto.ConnectDecoder; +import org.dna.mqtt.moquette.proto.DisconnectDecoder; +import org.dna.mqtt.moquette.proto.MQTTLoggingFilter; +import org.dna.mqtt.moquette.proto.PingReqDecoder; +import org.dna.mqtt.moquette.proto.PingRespEncoder; +import org.dna.mqtt.moquette.proto.PubAckDecoder; +import org.dna.mqtt.moquette.proto.PubAckEncoder; +import org.dna.mqtt.moquette.proto.PubCompDecoder; +import org.dna.mqtt.moquette.proto.PubCompEncoder; +import org.dna.mqtt.moquette.proto.PubRecDecoder; +import org.dna.mqtt.moquette.proto.PubRecEncoder; +import org.dna.mqtt.moquette.proto.PubRelDecoder; +import org.dna.mqtt.moquette.proto.PubRelEncoder; +import org.dna.mqtt.moquette.proto.PublishDecoder; +import org.dna.mqtt.moquette.proto.PublishEncoder; +import org.dna.mqtt.moquette.proto.SubAckEncoder; +import org.dna.mqtt.moquette.proto.SubscribeDecoder; +import org.dna.mqtt.moquette.proto.UnsubAckEncoder; +import org.dna.mqtt.moquette.proto.UnsubscribeDecoder; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.dna.mqtt.moquette.proto.messages.PingRespMessage; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; +import org.dna.mqtt.moquette.proto.messages.PubCompMessage; +import org.dna.mqtt.moquette.proto.messages.PubRecMessage; +import org.dna.mqtt.moquette.proto.messages.PubRelMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; +import org.dna.mqtt.moquette.server.MQTTHandler; +import org.dna.mqtt.moquette.server.ServerAcceptor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * + * @author andrea + */ +public class MinaAcceptor implements ServerAcceptor { + + private static final Logger LOG = LoggerFactory.getLogger(MinaAcceptor.class); + + private IoAcceptor m_acceptor; + + public void initialize(IMessaging messaging) throws IOException { + DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); + decoder.addMessageDecoder(new ConnectDecoder()); + decoder.addMessageDecoder(new PublishDecoder()); + decoder.addMessageDecoder(new PubAckDecoder()); + decoder.addMessageDecoder(new PubRelDecoder()); + decoder.addMessageDecoder(new PubRecDecoder()); + decoder.addMessageDecoder(new PubCompDecoder()); + decoder.addMessageDecoder(new SubscribeDecoder()); + decoder.addMessageDecoder(new UnsubscribeDecoder()); + decoder.addMessageDecoder(new DisconnectDecoder()); + decoder.addMessageDecoder(new PingReqDecoder()); + + DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); +// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); + encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); + encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); + encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder()); + encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder()); + encoder.addMessageEncoder(PubRecMessage.class, new PubRecEncoder()); + encoder.addMessageEncoder(PubCompMessage.class, new PubCompEncoder()); + encoder.addMessageEncoder(PubRelMessage.class, new PubRelEncoder()); + encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); + encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder()); + + m_acceptor = new NioSocketAcceptor(); + + m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") ); + m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); + + MQTTHandler handler = new MQTTHandler(); + + handler.setMessaging(messaging); + + m_acceptor.setHandler(handler); + ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); + ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true); + m_acceptor.getSessionConfig().setReadBufferSize( 2048 ); + m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT ); + m_acceptor.getStatistics().setThroughputCalculationInterval(10); + m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis()); + m_acceptor.bind( new InetSocketAddress(Constants.PORT) ); + LOG.info("Server binded"); + } + + public void close() { + //log statistics + IoServiceStatistics statistics = m_acceptor.getStatistics(); + statistics.updateThroughput(System.currentTimeMillis()); + System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput())); + System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput())); + + for(IoSession session: m_acceptor.getManagedSessions().values()) { + if(session.isConnected() && !session.isClosing()){ + session.close(false); + } + } + + m_acceptor.unbind(); + m_acceptor.dispose(); + } + +}