From 626f12e619a6ed7b66021b36609189f93fa5daf9 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Wed, 5 Sep 2012 13:56:50 +0200 Subject: [PATCH] Added brute force MINA testing client and server --- benchmarking/pom.xml | 16 +-- .../bechnmark/mina/DummyClientHandler.java | 23 ++++ .../mqtt/bechnmark/mina/MQTTBulkClient.java | 126 ++++++++++++++++++ .../mqtt/bechnmark/mina/MQTTDropHandler.java | 44 ++++++ .../mqtt/bechnmark/mina/MQTTDropServer.java | 97 ++++++++++++++ 5 files changed, 292 insertions(+), 14 deletions(-) create mode 100644 benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/DummyClientHandler.java create mode 100644 benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java create mode 100644 benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropHandler.java create mode 100644 benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropServer.java diff --git a/benchmarking/pom.xml b/benchmarking/pom.xml index 98f4655c..aa939d8e 100644 --- a/benchmarking/pom.xml +++ b/benchmarking/pom.xml @@ -76,32 +76,20 @@ 1.5 - + diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/DummyClientHandler.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/DummyClientHandler.java new file mode 100644 index 00000000..038b8a5f --- /dev/null +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/DummyClientHandler.java @@ -0,0 +1,23 @@ +package org.dna.mqtt.bechnmark.mina; + +import org.apache.mina.core.service.IoHandlerAdapter; +import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class DummyClientHandler extends IoHandlerAdapter { + + private static final Logger LOG = LoggerFactory.getLogger(DummyClientHandler.class); + + @Override + public void messageReceived(IoSession session, Object message) throws Exception { + AbstractMessage msg = (AbstractMessage) message; + LOG.debug("Received a message of type " + msg.getMessageType()); + } +} + diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java new file mode 100644 index 00000000..bf2728de --- /dev/null +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java @@ -0,0 +1,126 @@ +package org.dna.mqtt.bechnmark.mina; + +import java.io.IOException; +import java.net.InetSocketAddress; +import org.apache.mina.core.RuntimeIoException; +import org.apache.mina.core.future.ConnectFuture; +import org.apache.mina.core.future.WriteFuture; +import org.apache.mina.core.service.IoConnector; +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.NioSocketConnector; +import org.dna.mqtt.moquette.MQTTException; +import org.dna.mqtt.moquette.PublishException; +import org.dna.mqtt.moquette.proto.*; +import org.dna.mqtt.moquette.proto.messages.*; +import org.dna.mqtt.moquette.server.Server; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class MQTTBulkClient { + public int NUM_MESSAGES = 10000; + + private static final Logger LOG = LoggerFactory.getLogger(MQTTBulkClient.class); + + private String m_hostname; + private int m_port = 1883; + //internal management used for conversation with the server + private IoConnector m_connector; + private IoSession m_session; + boolean m_withWaitWriteFuture = false; + + public static void main(String[] args) throws IOException { + MQTTBulkClient client = new MQTTBulkClient(); + if (args.length > 0) { + client.m_hostname = args[0]; + client.m_withWaitWriteFuture = Boolean.parseBoolean(args[1]); + client.NUM_MESSAGES = Integer.parseInt(args[2]); + } else { + client.m_hostname = "localhost"; + } + client.init(); + client.connect(); + + long start = System.currentTimeMillis(); + for (int i = 0; i < client.NUM_MESSAGES; i++) { + client.publish("/topic", "Hello world".getBytes(), AbstractMessage.QOSType.MOST_ONE, false); + } + long stop = System.currentTimeMillis(); + LOG.info("Client sent " + client.NUM_MESSAGES + " in " + (stop - start) + " ms"); + client.shutdown(); + } + + protected void init() { + DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); + decoder.addMessageDecoder(new ConnAckDecoder()); + decoder.addMessageDecoder(new SubAckDecoder()); + decoder.addMessageDecoder(new UnsubAckDecoder()); + decoder.addMessageDecoder(new PublishDecoder()); + decoder.addMessageDecoder(new PubAckDecoder()); + decoder.addMessageDecoder(new PingRespDecoder()); + + DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); + encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); + encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); + encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder()); + encoder.addMessageEncoder(UnsubscribeMessage.class, new UnsubscribeEncoder()); + encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder()); + encoder.addMessageEncoder(PingReqMessage.class, new PingReqEncoder()); + + m_connector = new NioSocketConnector(); + +// m_connector.getFilterChain().addLast("logger", new LoggingFilter()); + m_connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder)); + + m_connector.setHandler(new DummyClientHandler()); + m_connector.getSessionConfig().setReadBufferSize(2048); + m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Server.DEFAULT_CONNECT_TIMEOUT); + } + + public void connect() throws MQTTException { + try { + ConnectFuture future = m_connector.connect(new InetSocketAddress(m_hostname, m_port)); + LOG.debug("Client waiting to connect to server"); + future.awaitUninterruptibly(); + m_session = future.getSession(); + } catch (RuntimeIoException e) { + LOG.debug("Failed to connect", e); + } + } + + public void publish(String topic, byte[] payload, AbstractMessage.QOSType qos, boolean retain) throws PublishException { + PublishMessage msg = new PublishMessage(); + msg.setRetainFlag(retain); + msg.setTopicName(topic); + msg.setPayload(payload); + + //QoS 0 case + msg.setQos(AbstractMessage.QOSType.MOST_ONE); + WriteFuture wf = m_session.write(msg); + + if (m_withWaitWriteFuture) { + try { + wf.await(); + } catch (InterruptedException ex) { + LOG.debug(null, ex); + throw new PublishException(ex); + } + + Throwable ex = wf.getException(); + if (ex != null) { + throw new PublishException(ex); + } + } + } + + public void shutdown() { + m_connector.dispose(); + } +} diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropHandler.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropHandler.java new file mode 100644 index 00000000..8ba3195e --- /dev/null +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropHandler.java @@ -0,0 +1,44 @@ +package org.dna.mqtt.bechnmark.mina; + +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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; + +/** + * + * @author andrea + */ +public class MQTTDropHandler extends IoHandlerAdapter { + + private static final Logger LOG = LoggerFactory.getLogger(MQTTDropHandler.class); + + @Override + public void messageReceived(IoSession session, Object message) throws Exception { + AbstractMessage msg = (AbstractMessage) message; + + try { + switch (msg.getMessageType()) { + case CONNECT: + case SUBSCRIBE: + case UNSUBSCRIBE: + case PUBLISH: + case PINGREQ: + case DISCONNECT: + LOG.debug("Received a message of type " + msg.getMessageType()); + break; + } + } catch (Exception ex) { + LOG.error("Bad error in processing the message", ex); + } + } + + @Override + public void sessionIdle(IoSession session, IdleStatus status) { + session.close(false); + } +} diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropServer.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropServer.java new file mode 100644 index 00000000..44783f3f --- /dev/null +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropServer.java @@ -0,0 +1,97 @@ +package org.dna.mqtt.bechnmark.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.moquette.proto.*; +import org.dna.mqtt.moquette.proto.messages.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class MQTTDropServer { + + private static final Logger LOG = LoggerFactory.getLogger(MQTTDropServer.class); + + public static final int PORT = 1883; + public static final int DEFAULT_CONNECT_TIMEOUT = 10; + private IoAcceptor m_acceptor; + + public static void main(String[] args) throws IOException { + new MQTTDropServer().startServer(); + } + + protected void startServer() throws IOException { + DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); + decoder.addMessageDecoder(new ConnectDecoder()); + decoder.addMessageDecoder(new PublishDecoder()); + 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(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)); + + MQTTDropHandler handler = new MQTTDropHandler(); + + 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, DEFAULT_CONNECT_TIMEOUT ); + m_acceptor.getStatistics().setThroughputCalculationInterval(10); + m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis()); + m_acceptor.bind( new InetSocketAddress(PORT) ); + LOG.info("Server binded"); + + //Bind a shutdown hook + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + stopServer(); + } + }); + } + + protected void stopServer() { + LOG.info("Server stopping..."); + + //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(); + LOG.info("Server stopped"); + } +}