Added brute force MINA testing client and server
This commit is contained in:
selva.andre
@@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user