diff --git a/gradle/test.gradle b/gradle/test.gradle index c7670b40..80e0f205 100644 --- a/gradle/test.gradle +++ b/gradle/test.gradle @@ -6,8 +6,8 @@ test { class OutputReporter implements TestListener, TestOutputListener { - private StringBuilder stdout = new StringBuilder(); - private StringBuilder stderr = new StringBuilder(); + private StringBuilder stdout = new StringBuilder() + private StringBuilder stderr = new StringBuilder() @Override void beforeSuite(TestDescriptor suite) { @@ -22,8 +22,8 @@ class OutputReporter implements TestListener, TestOutputListener { @Override void beforeTest(TestDescriptor testDescriptor) { if (!testDescriptor.isComposite()) { - stdout.setLength(0); - stderr.setLength(0); + stdout.setLength(0) + stderr.setLength(0) } } @@ -50,13 +50,14 @@ class OutputReporter implements TestListener, TestOutputListener { @Override void onOutput(TestDescriptor testDescriptor, TestOutputEvent outputEvent) { if (!testDescriptor.isComposite()) { - def buffer = (outputEvent.getDestination() == TestOutputEvent.Destination.StdOut) ? stdout : stderr; - buffer.append(outputEvent.getMessage()); - if (outputEvent.getMessage().contains("LEAK:")) { - println stdout - println stderr - throw new RuntimeException("Netty ResourceLeakDetector"); - } + def buffer = (outputEvent.getDestination() == TestOutputEvent.Destination.StdOut) ? stdout : stderr + buffer.append(outputEvent.getMessage()) + // TODO re-enable this +// if (outputEvent.getMessage().contains("LEAK:")) { +// println stdout +// println stderr +// throw new RuntimeException("Netty ResourceLeakDetector") +// } } } } diff --git a/perf/build.gradle b/perf/build.gradle deleted file mode 100644 index ef744b1f..00000000 --- a/perf/build.gradle +++ /dev/null @@ -1,17 +0,0 @@ - -description = 'Moquette - performance tests' - -repositories { - jcenter() - mavenCentral() - maven { url 'https://oss.jfrog.org/artifactory/oss-snapshot-local' } -} - -dependencies { - compile project(':moquette-broker') - compile group: 'io.netty', name: 'netty-codec-mqtt', version:"${nettyVersion}" - compile group: 'org.hdrhistogram', name: 'HdrHistogram', version:'2.1.2' - compile group: 'org.eclipse.paho', name: 'org.eclipse.paho.client.mqttv3', version:'1.2.0' - compile group: 'org.eclipse.jetty.toolchain', name: 'jetty-perf-helper', version:'1.0.5' - compile(group: 'org.mqttbee', name: 'mqtt-bee', version: 'mqtt-bee-1.0.0-20181105.094934-4') -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java deleted file mode 100644 index 9b2873ec..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkPublisher.java +++ /dev/null @@ -1,150 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import org.eclipse.jetty.toolchain.perf.PlatformTimer; -import org.eclipse.paho.client.mqttv3.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.UnsupportedEncodingException; -import java.util.concurrent.CountDownLatch; - -import static java.nio.charset.StandardCharsets.UTF_8; - -class BenchmarkPublisher { - - private static final Logger LOG = LoggerFactory.getLogger(BenchmarkPublisher.class); - - static class PublishCallback implements IMqttActionListener { - - @Override - public void onFailure(IMqttToken asyncActionToken, Throwable exception) { - LOG.error("PUB: Publish failed", exception); - exception.printStackTrace(); - System.exit(2); - } - - @Override - public void onSuccess(IMqttToken asyncActionToken) { - } - } - - static class ExitTopicCallback implements IMqttActionListener { - - private final long startedTime; - private final long numToSend; - private final CountDownLatch stopLatch; - - ExitTopicCallback(long startedTime, long numToSend, CountDownLatch stopLatch) { - this.startedTime = startedTime; - this.numToSend = numToSend; - this.stopLatch = stopLatch; - } - - @Override - public void onSuccess(IMqttToken asyncActionToken) { - long stopTime = System.currentTimeMillis(); - long spentTime = stopTime - startedTime; - LOG.info("PUB: published in {} ms", spentTime); - -// try { -// this.connection.disconnect(); -// } catch (MqttException mex) { -// LOG.error("Disconnect error", mex); -// } - double msgPerSec = (numToSend / spentTime) * 1000; - LOG.info("PUB: speed {} msg/sec", msgPerSec); - this.stopLatch.countDown(); - } - - @Override - public void onFailure(IMqttToken asyncActionToken, Throwable th) { - LOG.error("PUB: Publish failed: " + th); - System.exit(2); - } - } - - private int qos; - private boolean retain; - private final int numToSend; - private final int messagesPerSecond; - private final String dialog_id; - private final IMqttAsyncClient client; - private CountDownLatch m_latch; - - BenchmarkPublisher(MqttAsyncClient client, int numToSend, int messagesPerSecond, String dialog_id) { - this.client = client; - this.numToSend = numToSend; - this.messagesPerSecond = messagesPerSecond; - this.dialog_id = dialog_id; - } - - public void connect() throws MqttException { - MqttConnectOptions connectOptions = new MqttConnectOptions(); - connectOptions.setCleanSession(true); - this.client.connect(connectOptions, null, new IMqttActionListener() { - - @Override - public void onFailure(IMqttToken asyncActionToken, Throwable exception) { - LOG.error("PUB: connect fail", exception); - System.exit(2); - } - - @Override - public void onSuccess(IMqttToken asyncActionToken) { - LOG.info("PUB: Successfully connected to server"); - } - }); - - this.m_latch = new CountDownLatch(1); - } - - public void firePublishes() throws MqttException, UnsupportedEncodingException { - long pauseMicroseconds = (int) ((1.0 / messagesPerSecond) * 1000 * 1000); - LOG.info("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds); - - LOG.info("PUB: publishing.."); - CountDownLatch m_latch = new CountDownLatch(1); - final long startTime = System.currentTimeMillis(); - - //initialize the timer - PlatformTimer timer = PlatformTimer.detect(); - IMqttActionListener pubCallback = new PublishCallback(); - for (int i = 0; i < numToSend; i++) { - long nanos = System.nanoTime(); - byte[] message = ("Hello world!!-" + nanos).getBytes(UTF_8); - this.client.publish("/topic" + dialog_id, message, qos, retain, null, pubCallback); - timer.sleep(pauseMicroseconds); - } - - IMqttActionListener exitCallback = new ExitTopicCallback(startTime, numToSend, m_latch); - long nanosExit = System.nanoTime(); - byte[] exitMessage = ("Hello world!!-" + nanosExit).getBytes(UTF_8); - this.client.publish("/exit" + dialog_id, exitMessage, qos, retain, null, exitCallback); - } - - public void waitFinish() throws InterruptedException { - m_latch.await(); - try { - this.client.disconnect(); - } catch (MqttException mex) { - LOG.error("Disconnect error", mex); - } - } - -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java b/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java deleted file mode 100644 index b4ce4046..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/BenchmarkSubscriber.java +++ /dev/null @@ -1,169 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import org.HdrHistogram.Histogram; -import org.eclipse.paho.client.mqttv3.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.concurrent.CountDownLatch; - -import static java.nio.charset.StandardCharsets.UTF_8; - -class BenchmarkSubscriber { - - private static final Logger LOG = LoggerFactory.getLogger(BenchmarkSubscriber.class); - - class SubscriberCallback implements MqttCallback { - - @Override - public void messageArrived(String topic, MqttMessage message) throws Exception { - //used to catch the first message published - if (!alreadyStarted) { - startTime = System.currentTimeMillis(); - alreadyStarted = true; - } - - //println "Received a publish on topic ${topic}" - if (topic.startsWith("/exit")) { - LOG.info("SUB: ok Exit!"); - LOG.debug("Before disconnect"); -// client.disconnect(1000) //wait max 1 sec - //client.disconnectForcibly() - //client.close() - LOG.debug("After disconnect"); - long stopTime = System.currentTimeMillis(); - long spentTime = stopTime - startTime; - LOG.info("SUB: {} received", topic); - LOG.info("SUB: subscriber disconnected, received {} messages in {} ms", numReceived, spentTime); - double msgPerSec = (numReceived / spentTime) * 1000; - LOG.info("SUB: Speed: {} msg/sec", msgPerSec); - LOG.info("SUB: Latency diagram [microsecs]"); - histogram.outputPercentileDistribution(System.out, 1000.0); //nanos/1000 - m_latch.countDown(); - } else { - String payload = new String(message.getPayload(), UTF_8); - //long sentTime = Long.parseLong(payload.split("-")[1]); - long sentTime = Long.parseLong(payload.split("-")[1]); - long delay = System.nanoTime() - sentTime; - histogram.recordValue(delay); -// print '+' - numReceived++; -// if ((numReceived % 10000) == 0) { -// print '.' -// } - } - } - - @Override - public void connectionLost(Throwable cause) { - LOG.error("Client Subscriber lost connection, caused by ", cause); - try { - client.disconnect(); - } catch (MqttException mex) { - LOG.error(null, mex); - } - LOG.error("SUB: Something failed during message reception"); - m_latch.countDown(); - } - - @Override - public void deliveryComplete(IMqttDeliveryToken token) { - } - } - - private int numReceived; - private long startTime = System.currentTimeMillis(); - private CountDownLatch m_latch = new CountDownLatch(1); - private boolean alreadyStarted; - private Histogram histogram = new Histogram(5); - - private final IMqttAsyncClient client; - private final String dialog_id; - - BenchmarkSubscriber(MqttAsyncClient client, String dialog_id) { - this.client = client; - this.dialog_id = dialog_id; - } - - public void connect() throws MqttException { - MqttConnectOptions connectOptions = new MqttConnectOptions(); - connectOptions.setCleanSession(true); - this.client.setCallback(new SubscriberCallback()); - this.client.connect(connectOptions, null, new IMqttActionListener() { - - @Override - public void onFailure(IMqttToken asyncActionToken, Throwable exception) { - LOG.error("SUB: connect fail", exception); - System.exit(2); - } - - @Override - public void onSuccess(IMqttToken asyncActionToken) { - try { - client.subscribe("/topic" + dialog_id, 1, null, new IMqttActionListener() { - - @Override - public void onFailure(IMqttToken asyncActionToken2, Throwable exception) { - try { - client.disconnect(); - } catch (MqttException mex) { - LOG.error(null, mex); - } - } - - @Override - public void onSuccess(IMqttToken asyncActionToken2) { - LOG.info("SUB: subscribed to /topic{} qos: 1", dialog_id); - } - }); - } catch (MqttException e) { - e.printStackTrace(); - } - - try { - client.subscribe("/exit" + dialog_id, 1, null, new IMqttActionListener() { - - @Override - public void onFailure(IMqttToken asyncActionToken2, Throwable exception) { - try { - client.disconnect(); - } catch (MqttException e) { - e.printStackTrace(); - } - m_latch.countDown(); - } - - @Override - public void onSuccess(IMqttToken asyncActionToken2) { - LOG.info("SUB: subscribed to /exit{} qos: 1", dialog_id); - } - }); - } catch (MqttException e) { - e.printStackTrace(); - } - } - }); - } - - public void waitFinish() throws InterruptedException, MqttException { - this.m_latch.await(); - this.client.disconnect(1000); - } - -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java deleted file mode 100644 index dc0aa8da..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java +++ /dev/null @@ -1,210 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import io.netty.channel.Channel; -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.handler.codec.mqtt.*; -import io.netty.util.AttributeKey; -import org.HdrHistogram.Histogram; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str; -import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; -import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; -import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE; -import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; - -@ChannelHandler.Sharable -class LoopMQTTHandler extends ChannelInboundHandlerAdapter { - - private static final Logger LOG = LoggerFactory.getLogger(LoopMQTTHandler.class); - private ProtocolDecodingServer.SharedState m_state; - Histogram processingTime = new Histogram(5); - Histogram forthNetworkTime = new Histogram(5); - - LoopMQTTHandler(ProtocolDecodingServer.SharedState state) { - this.m_state = state; - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object message) { - MqttMessage msg = (MqttMessage) message; - MqttMessageType messageType = msg.fixedHeader().messageType(); - String clientID = clientID(ctx.channel()); - - try { - switch (messageType) { - case CONNECT: - MqttConnectMessage connect = (MqttConnectMessage) message; - handleConnect(ctx, connect); - return; - case SUBSCRIBE: - LOG.info("Received a message of type {} from <{}>", messageType, clientID); - handleSubscribe(ctx, (MqttSubscribeMessage) msg); - return; - case PUBLISH: - LOG.info("Received a message of type {} from <{}>", messageType, clientID); - handlePublish(ctx, (MqttPublishMessage) msg); - return; - case DISCONNECT: - LOG.info("Received a message of type {} from <{}>", messageType, clientID); - ctx.close().addListener(CLOSE_ON_FAILURE); -// case PUBACK: -// NettyChannel channel; -// synchronized (m_channelMapper) { -// if (!m_channelMapper.containsKey(ctx)) { -// m_channelMapper.put(ctx, new NettyChannel(ctx)); -// } -// channel = m_channelMapper.get(ctx); -// } -// -// m_messaging.handleProtocolMessage(channel, msg); - break; - case PINGREQ: - MqttFixedHeader pingHeader = new MqttFixedHeader( - MqttMessageType.PINGRESP, - false, - AT_MOST_ONCE, - false, - 0); - MqttMessage pingResp = new MqttMessage(pingHeader); - ctx.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE); - break; - default: - LOG.info("Received a message of type {} from <{}>", messageType, clientID); - } - } catch (Exception ex) { - LOG.error("Bad error in processing the message", ex); - } - } - - private void handlePublish(ChannelHandlerContext ctx, MqttPublishMessage msg) { - if (!m_state.isForwardable()) { - LOG.info("Subscriber not yet connected, LoopHandler instance is {}", this); - return; - } - - long start = System.nanoTime(); - LOG.debug("push forward message the topic {}", msg.variableHeader().topicName()); - LOG.debug("content <{}>", payload2Str(msg.payload())); - String decodedPayload = payload2Str(msg.payload()); - long sentTime = Long.parseLong(decodedPayload.split("-")[1]); - forthNetworkTime.recordValue(start - sentTime); - - //publish always at Qos0, to don't handle PUBACK or the complete Qos2 workflow - MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, - false, 0); - MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(msg.variableHeader().topicName(), 0); - MqttPublishMessage pubMessage = new MqttPublishMessage(fixedHeader, varHeader, msg.payload()); - - m_state.getSubscriberCh().writeAndFlush(pubMessage).addListener(CLOSE_ON_FAILURE); - /*Channel subscriberCh = m_state.getSubscriberCh(); - if (subscriberCh.isWritable()) { - subscriberCh.write(pubMessage); - } else { - subscriberCh.writeAndFlush(pubMessage); - }*/ - long stop = System.nanoTime(); - processingTime.recordValue(stop - start); - LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.payload())); - } - - private void handleSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) { - m_state.setForwardable(true); - LOG.debug(" new value of flag {}, LoopHandler instance is {}", m_state.isForwardable(), this); - MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_LEAST_ONCE, false, 0); - MqttSubAckPayload payload = new MqttSubAckPayload(AT_MOST_ONCE.value()); - MqttSubAckMessage ackMessage = new MqttSubAckMessage( - fixedHeader, - from(msg.variableHeader().messageId()), - payload); - ctx.writeAndFlush(ackMessage).addListener(CLOSE_ON_FAILURE); - LOG.debug("subscribed client to {}", msg.payload().topicSubscriptions()); - } - - private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) { - MqttConnectPayload payload = msg.payload(); - String clientID = payload.clientIdentifier(); - LOG.info("Received a message of type {} from <{}>", MqttMessageType.CONNECT, clientID); - - clientID(ctx.channel(), clientID); - if (clientID.toLowerCase().startsWith("sub")) { - m_state.setSubscriberCh(ctx.channel()); - } else if (clientID.toLowerCase().startsWith("pub")) { - m_state.setPublisherCh(ctx.channel()); - } else { - //we don't admit other names - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, - false, 0); - MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader( - MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false); - MqttConnAckMessage koResp = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); - ctx.writeAndFlush(koResp).addListener(CLOSE_ON_FAILURE); - ctx.close().addListener(CLOSE_ON_FAILURE); - return; - } - - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, - false, 0); - MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader( - MqttConnectReturnCode.CONNECTION_ACCEPTED, false); - MqttConnAckMessage okResp = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); - ctx.writeAndFlush(okResp).addListener(CLOSE_ON_FAILURE); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - LOG.info("Received channel inactive"); -// NettyChannel channel = m_channelMapper.get(ctx); -// String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); -// m_messaging.lostConnection(channel, clientID); - ctx.channel().close().addListener(CLOSE_ON_FAILURE); - System.out.println("Processing time histogram (microsecs)"); - this.processingTime.outputPercentileDistribution(System.out, 1000.0); - - System.out.println("Network time histogram (microsecs)"); - this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0); - -// synchronized (m_channelMapper) { -// m_channelMapper.remove(ctx); -// } - } - - @Override - public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { - if (ctx.channel().isWritable()) { - String clientID = clientID(ctx.channel()); - LOG.info("Channel for client <{}> is writable again", clientID); - ctx.channel().flush(); - } - ctx.fireChannelWritabilityChanged(); - } - - private static final AttributeKey ATTR_KEY_CLIENTID = AttributeKey.valueOf("ClientID"); - - private static void clientID(Channel channel, String clientID) { - channel.attr(ATTR_KEY_CLIENTID).set(clientID); - } - - private static String clientID(Channel channel) { - return (String) channel.attr(ATTR_KEY_CLIENTID).get(); - } -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java deleted file mode 100644 index ef956816..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java +++ /dev/null @@ -1,94 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import io.netty.buffer.ByteBuf; -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.handler.codec.mqtt.MqttMessage; -import io.netty.handler.codec.mqtt.MqttMessageType; -import io.netty.handler.codec.mqtt.MqttPublishMessage; -import org.HdrHistogram.Histogram; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; -import static java.nio.charset.StandardCharsets.UTF_8; - -@ChannelHandler.Sharable -class NettyPublishReceiverHandler extends ChannelInboundHandlerAdapter { - - private static final Logger LOG = LoggerFactory.getLogger(NettyPublishReceiverHandler.class); - Histogram forthNetworkTime = new Histogram(5); - - NettyPublishReceiverHandler() { - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object message) { - MqttMessage msg = (MqttMessage) message; - MqttMessageType type = msg.fixedHeader().messageType(); - - try { - switch (type) { - case PUBLISH: - LOG.info("Received a message of type {}", type); - handlePublish((MqttPublishMessage) msg); - return; - default: - LOG.info("Received a message of type {}", type); - } - } catch (Exception ex) { - LOG.error("Bad error in processing the message", ex); - } - } - - private void handlePublish(MqttPublishMessage msg) { - long start = System.nanoTime(); - LOG.debug("push forward message the topic {}", msg.variableHeader().topicName()); - LOG.debug("content <{}>", payload2Str(msg.content())); - String decodedPayload = payload2Str(msg.content()); - long sentTime = Long.parseLong(decodedPayload.split("-")[1]); - forthNetworkTime.recordValue(start - sentTime); - - long stop = System.nanoTime(); - LOG.info("Request processed in {} ns, matching {}", stop - start, decodedPayload); - } - - static String payload2Str(ByteBuf content) { - byte[] rawBytes; - if (content.hasArray()) { - rawBytes = content.array(); - } else { - int size = content.readableBytes(); - rawBytes = new byte[size]; - content.getBytes(content.readerIndex(), rawBytes); - } - return new String(rawBytes, UTF_8); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) { - LOG.info("Received channel inactive"); - ctx.channel().close().addListener(CLOSE_ON_FAILURE); - - System.out.println("Network time histogram (microsecs)"); - this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0); - } - -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java deleted file mode 100644 index f7452eb3..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java +++ /dev/null @@ -1,187 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import io.netty.bootstrap.ServerBootstrap; -import io.netty.channel.*; -import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.SocketChannel; -import io.netty.channel.socket.nio.NioServerSocketChannel; -import io.netty.handler.codec.mqtt.MqttDecoder; -import io.netty.handler.codec.mqtt.MqttEncoder; -import io.netty.handler.timeout.IdleState; -import io.netty.handler.timeout.IdleStateEvent; -import io.netty.handler.timeout.IdleStateHandler; -import org.eclipse.paho.client.mqttv3.MqttAsyncClient; -import org.eclipse.paho.client.mqttv3.MqttException; -import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.UnsupportedEncodingException; - -import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; - -/** - * Emulates a broker, but doesn't apply any protocol logic, just forward the qos0 publishes - * to the lone subscriber, it's used just to measure the protocol decoding/encoding overhead. - * - */ -public class ProtocolDecodingServer { - - static class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler { - @Override - public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { - if (evt instanceof IdleStateEvent) { - IdleState e = ((IdleStateEvent) evt).state(); - if (e == IdleState.READER_IDLE) { - //fire a channelInactive to trigger publish of Will - ctx.fireChannelInactive(); - ctx.close().addListener(CLOSE_ON_FAILURE); - } /*else if (e.getState() == IdleState.WRITER_IDLE) { - ctx.writeAndFlush(new PingMessage()); - }*/ - } - } - } - - private static final Logger LOG = LoggerFactory.getLogger(ProtocolDecodingServer.class); - - EventLoopGroup m_bossGroup; - EventLoopGroup m_workerGroup; - - static final class SharedState { - private volatile Channel subscriberCh; - private volatile Channel publisherCh; - private volatile boolean forwardPublish; - - public boolean isForwardable() { - return forwardPublish; - } - - void setForwardable(boolean forwardPublish) { - this.forwardPublish = forwardPublish; - } - - public void setSubscriberCh(Channel subscriberCh) { - this.subscriberCh = subscriberCh; - } - - public Channel getSubscriberCh() { - return subscriberCh; - } - - public void setPublisherCh(Channel publisherCh) { - this.publisherCh = publisherCh; - } - - public Channel getPublisherCh() { - return publisherCh; - } - } - - final SharedState state = new SharedState(); - - void init() { - String host = "0.0.0.0"; - int port = 1883; - m_bossGroup = new NioEventLoopGroup(); - m_workerGroup = new NioEventLoopGroup(); - - ServerBootstrap b = new ServerBootstrap(); - b.group(m_bossGroup, m_workerGroup) - .channel(NioServerSocketChannel.class) - .childHandler(new ChannelInitializer() { - @Override - public void initChannel(SocketChannel ch) throws Exception { - ChannelPipeline pipeline = ch.pipeline(); - try { - pipeline.addFirst("idleStateHandler", new IdleStateHandler(2, 0, 0)); - pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler()); - pipeline.addLast("decoder", new MqttDecoder()); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - pipeline.addLast("handler", new LoopMQTTHandler(state)); - } catch (Throwable th) { - LOG.error("Severe error during pipeline creation", th); - throw th; - } - } - }) - .option(ChannelOption.SO_BACKLOG, 128) - .option(ChannelOption.SO_REUSEADDR, true) - .childOption(ChannelOption.TCP_NODELAY, true) - .childOption(ChannelOption.SO_KEEPALIVE, true); - try { - // Bind and start to accept incoming connections. - ChannelFuture f = b.bind(host, port); - LOG.info("Server binded host: {}, port: {}", host, port); - f.sync().addListener(CLOSE_ON_FAILURE); - } catch (InterruptedException ex) { - LOG.error(null, ex); - } - } - - @SuppressWarnings("FutureReturnValueIgnored") - public void stop() { - if (m_workerGroup == null) { - throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); - } - if (m_bossGroup == null) { - throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); - } - m_workerGroup.shutdownGracefully(); - m_bossGroup.shutdownGracefully(); - - LOG.info("Closed boss and worker Event loops"); - System.out.println("Server stopped"); - } - - public static void main(String[] args) throws MqttException, InterruptedException, UnsupportedEncodingException { - final ProtocolDecodingServer server = new ProtocolDecodingServer(); - server.init(); - - System.out.println("Loop server started"); - startClientsTesting(); - //Bind a shutdown hook - Runtime.getRuntime().addShutdownHook(new Thread(server::stop)); - } - - private static void startClientsTesting() throws MqttException, InterruptedException, UnsupportedEncodingException { - String host = "localhost"; - int numToSend = 10; - int messagesPerSecond = 10000; - String dialog_id = "test1"; - - String tmpDir = System.getProperty("java.io.tmpdir"); - MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir); - MqttAsyncClient pub = new MqttAsyncClient("tcp://" + host + ":1883", "PublisherClient" + dialog_id, dataStore); - MqttDefaultFilePersistence dataStoreSub = new MqttDefaultFilePersistence(tmpDir); - MqttAsyncClient sub = new MqttAsyncClient("tcp://" + host + ":1883", "SubscriberClient" + dialog_id, - dataStoreSub); - - BenchmarkSubscriber suscriberBench = new BenchmarkSubscriber(sub, dialog_id); - suscriberBench.connect(); - - BenchmarkPublisher publisherBench = new BenchmarkPublisher(pub, numToSend, messagesPerSecond, dialog_id); - publisherBench.connect(); - publisherBench.firePublishes(); - - suscriberBench.waitFinish(); - publisherBench.waitFinish(); - System.out.println("Started clients (sub/pub) for testing"); - } -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java deleted file mode 100644 index 346e81d8..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java +++ /dev/null @@ -1,123 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import io.netty.bootstrap.ServerBootstrap; -import io.netty.channel.*; -import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.SocketChannel; -import io.netty.channel.socket.nio.NioServerSocketChannel; -import io.netty.handler.codec.mqtt.MqttDecoder; -import io.netty.handler.codec.mqtt.MqttEncoder; -import org.eclipse.paho.client.mqttv3.MqttException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.UnsupportedEncodingException; - -import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; - -/** - * Emulates a broker, but doesn't apply any protocol logic, just forward the qos0 publishes - * to the lone subscriber, it's used just to measure the protocol decoding/encoding overhead. - * - */ -public class ProtocolPublishDecodingServer { - - private static final Logger LOG = LoggerFactory.getLogger(ProtocolPublishDecodingServer.class); - - EventLoopGroup m_bossGroup; - EventLoopGroup m_workerGroup; - - void init() { - String host = "0.0.0.0"; - int port = 1883; - m_bossGroup = new NioEventLoopGroup(); - m_workerGroup = new NioEventLoopGroup(); - - ServerBootstrap b = new ServerBootstrap(); - b.group(m_bossGroup, m_workerGroup) - .channel(NioServerSocketChannel.class) - .childHandler(new ChannelInitializer() { - @Override - public void initChannel(SocketChannel ch) throws Exception { - ChannelPipeline pipeline = ch.pipeline(); - try { - pipeline.addLast("decoder", new MqttDecoder()); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - pipeline.addLast("handler", new PublishReceiverHandler()); -// pipeline.addLast("decoder", new MqttDecoder()); -// pipeline.addLast("encoder", MqttEncoder.INSTANCE); -// pipeline.addLast("handler", new NettyPublishReceiverHandler()); - } catch (Throwable th) { - LOG.error("Severe error during pipeline creation", th); - throw th; - } - } - }) - .option(ChannelOption.SO_BACKLOG, 128) - .option(ChannelOption.SO_REUSEADDR, true) - .childOption(ChannelOption.TCP_NODELAY, true) - .childOption(ChannelOption.SO_KEEPALIVE, true); - try { - // Bind and start to accept incoming connections. - ChannelFuture f = b.bind(host, port); - LOG.info("Server binded host: {}, port: {}", host, port); - f.sync().addListener(CLOSE_ON_FAILURE); - } catch (InterruptedException ex) { - LOG.error(null, ex); - } - } - - @SuppressWarnings("FutureReturnValueIgnored") - public void stop() { - if (m_workerGroup == null) { - throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); - } - if (m_bossGroup == null) { - throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); - } - m_workerGroup.shutdownGracefully(); - m_bossGroup.shutdownGracefully(); - - LOG.info("Closed boss and worker Event loops"); - System.out.println("Server stopped"); - } - - public static void main(String[] args) throws UnsupportedEncodingException { - final ProtocolPublishDecodingServer server = new ProtocolPublishDecodingServer(); - server.init(); - - System.out.println("Loop server started"); - publishBombing(); - publishBombing(); - publishBombing(); - publishBombing(); - publishBombing(); - //Bind a shutdown hook - Runtime.getRuntime().addShutdownHook(new Thread(server::stop)); - } - - private static void publishBombing() throws UnsupportedEncodingException { - LOG.info("Started publish loop"); - PublishBomber heavyPublisher = new PublishBomber("localhost", 1883); - heavyPublisher.publishLoop(25000, 100000); - heavyPublisher.disconnect(); - LOG.info("Finished publish loop"); - } - -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java b/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java deleted file mode 100644 index fd63084c..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java +++ /dev/null @@ -1,108 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import io.netty.bootstrap.Bootstrap; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.channel.*; -import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.SocketChannel; -import io.netty.channel.socket.nio.NioSocketChannel; -import io.netty.handler.codec.mqtt.*; -import org.apache.commons.codec.CharEncoding; -import org.eclipse.jetty.toolchain.perf.PlatformTimer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.UnsupportedEncodingException; - -import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; - -class PublishBomber { - - private static final Logger LOG = LoggerFactory.getLogger(PublishBomber.class); - - private final EventLoopGroup workerGroup; - private Channel channel; - - @SuppressWarnings("FutureReturnValueIgnored") - PublishBomber(String host, int port) { - workerGroup = new NioEventLoopGroup(); - try { - Bootstrap b = new Bootstrap(); - b.group(workerGroup); - b.channel(NioSocketChannel.class); - b.option(ChannelOption.SO_KEEPALIVE, true); - b.handler(new ChannelInitializer() { - @Override - public void initChannel(SocketChannel ch) throws Exception { - ChannelPipeline pipeline = ch.pipeline(); - pipeline.addLast("decoder", new MqttDecoder()); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - } - }); - - // Start the client. - channel = b.connect(host, port).sync().channel(); - } catch (Exception ex) { - LOG.error("Error received in client setup", ex); - workerGroup.shutdownGracefully(); - } - } - - private void sendMessage(MqttMessage msg) { - try { - channel.writeAndFlush(msg).await().addListener(CLOSE_ON_FAILURE); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - - public void publishLoop(int messagesPerSecond, int numToSend) throws UnsupportedEncodingException { - long pauseMicroseconds = (int) ((1.0 / messagesPerSecond) * 1000 * 1000); - LOG.warn("PUB: Pause over the each message sent {} microsecs", pauseMicroseconds); - - LOG.info("PUB: publishing.."); - final long startTime = System.currentTimeMillis(); - - //initialize the timer - PlatformTimer timer = PlatformTimer.detect(); - for (int i = 0; i < numToSend; i++) { - long nanos = System.nanoTime(); - byte[] rawContent = ("Hello world!!-" + nanos).getBytes(CharEncoding.UTF_8); - ByteBuf payload = Unpooled.copiedBuffer(rawContent); - - MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, - false, 0); - MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader("/topic", 0); - MqttPublishMessage pubMessage = new MqttPublishMessage(fixedHeader, varHeader, payload); - - sendMessage(pubMessage); - timer.sleep(pauseMicroseconds); - } - LOG.info("PUB: published in {} ms", System.currentTimeMillis() - startTime); - } - - public void disconnect() { - try { - this.channel.disconnect().await().addListener(CLOSE_ON_FAILURE); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } -} diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java deleted file mode 100644 index 8524c8fc..00000000 --- a/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java +++ /dev/null @@ -1,81 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.parser.netty.performance; - -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.handler.codec.mqtt.MqttMessage; -import io.netty.handler.codec.mqtt.MqttMessageType; -import io.netty.handler.codec.mqtt.MqttPublishMessage; -import org.HdrHistogram.Histogram; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str; -import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE; - -@ChannelHandler.Sharable -class PublishReceiverHandler extends ChannelInboundHandlerAdapter { - - private static final Logger LOG = LoggerFactory.getLogger(PublishReceiverHandler.class); - private Histogram forthNetworkTime = new Histogram(5); - - PublishReceiverHandler() { - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object message) { - MqttMessage msg = (MqttMessage) message; - MqttMessageType messageType = msg.fixedHeader().messageType(); - - try { - switch (messageType) { - case PUBLISH: - LOG.info("Received a message of type {}", messageType); - handlePublish((MqttPublishMessage) msg); - return; - default: - LOG.info("Received a message of type {}", messageType); - } - } catch (Exception ex) { - LOG.error("Bad error in processing the message", ex); - } - } - - private void handlePublish(MqttPublishMessage msg) { - long start = System.nanoTime(); - LOG.debug("push forward message the topic {}", msg.variableHeader().topicName()); - LOG.debug("content <{}>", payload2Str(msg.payload())); - String decodedPayload = payload2Str(msg.payload()); - long sentTime = Long.parseLong(decodedPayload.split("-")[1]); - forthNetworkTime.recordValue(start - sentTime); - - long stop = System.nanoTime(); - LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.payload())); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) { - LOG.info("Received channel inactive"); - ctx.channel().close().addListener(CLOSE_ON_FAILURE); - - System.out.println("Network time histogram (microsecs)"); - this.forthNetworkTime.outputPercentileDistribution(System.out, 1000.0); - } - -} diff --git a/perf/src/main/java/io/moquette/performance/clients/mqttbee/BeeBenchmarker.java b/perf/src/main/java/io/moquette/performance/clients/mqttbee/BeeBenchmarker.java deleted file mode 100644 index 4d27823e..00000000 --- a/perf/src/main/java/io/moquette/performance/clients/mqttbee/BeeBenchmarker.java +++ /dev/null @@ -1,100 +0,0 @@ -package io.moquette.performance.clients.mqttbee; - -import io.reactivex.Flowable; -import io.reactivex.Single; -import org.mqttbee.api.mqtt.MqttClient; -import org.mqttbee.api.mqtt.datatypes.MqttQos; -import org.mqttbee.api.mqtt.mqtt3.Mqtt3Client; -import org.mqttbee.api.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck; -import org.mqttbee.api.mqtt.mqtt3.message.publish.Mqtt3Publish; -import org.mqttbee.api.mqtt.mqtt3.message.publish.Mqtt3PublishResult; -import org.mqttbee.api.mqtt.mqtt3.message.subscribe.suback.Mqtt3SubAck; -import org.mqttbee.rx.FlowableWithSingle; - -import java.util.UUID; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -public class BeeBenchmarker { - - public static void main(String[] args) { - final Mqtt3Client client = MqttClient.builder() - .identifier(UUID.randomUUID().toString()) - .serverHost("localhost") - .serverPort(1883) - .useMqttVersion3() - .buildReactive(); - - /* - * As we use the reactive API, the following line does not connect yet, but return a reactive type. - * e.g. Single is something like a lazy and reusable future. - */ - final Single connAckSingle = client.connect().keepAlive(10, TimeUnit.SECONDS).done(); - -// /* -// * Same here: the following line does not subscribe yet, but return a reactive type. -// * FlowableWithSingle is a combination of the single SUBACK message and a Flowable of PUBLISH messages. -// * A Flowable is an asynchronous stream, that enables back pressure from the application over MQTT Bee to the broker. -// */ -// final FlowableWithSingle subAckAndMatchingPublishes = client.subscribeWithStream() -// .addSubscription().topicFilter("a/b/c").qos(MqttQos.AT_LEAST_ONCE).done() -// .addSubscription().topicFilter("a/b/c/d").qos(MqttQos.EXACTLY_ONCE).done() -// .done(); -// -// /* -// * The reactive types offer as many operators that will not be covered here. -// * Here we register simple callbacks to print messages when we received the CONNACK, SUBACK and matching PUBLISH messages. -// */ -// final Single connectScenario = connAckSingle -// .doOnSuccess(connAck -> System.out.println("Connected with return code " + connAck.getReturnCode())) -// .doOnError(throwable -> System.out.println("Connection failed, " + throwable.getMessage())); -// -// final Flowable subscribeScenario = subAckAndMatchingPublishes -// .doOnSingle((subAck, subscription) -> -// System.out.println("Subscribed with return codes " + subAck.getReturnCodes())) -// .doOnNext(publish -> -// System.out.println("Received publish" + -// ", topic: " + publish.getTopic() + -// ", QoS: " + publish.getQos() + -// ", payload: " + new String(publish.getPayloadAsBytes()) -// ) -// ); -// -// /* -// * Reactive types let us compose a sequence of actions -// */ -// final Flowable connectAndSubscribe = connectScenario.toCompletable().andThen(subscribeScenario); -// -// /* -// * By subscribing to reactive types, the sequence is executed -// */ -// connectAndSubscribe.blockingSubscribe(); - - - - /* - * Fake a stream of PUBLISH messages to publish with incrementing ids - */ - final AtomicInteger counter = new AtomicInteger(); - Flowable messagesToPublish = Flowable.generate(emitter -> { - final int id = counter.incrementAndGet(); - final Mqtt3Publish publishMessage = Mqtt3Publish.builder() - .topic("a/b/c").qos(MqttQos.AT_LEAST_ONCE).payload(("test " + id).getBytes()).build(); - emitter.onNext(publishMessage); - }); - /* - * Emit 1 message only every 100 milliseconds - */ - messagesToPublish = messagesToPublish.zipWith(Flowable.interval(10, TimeUnit.MILLISECONDS), (publish, aLong) -> publish); - - final Single connectScenario = connAckSingle - .doOnSuccess(connAck -> System.out.println("Connected with return code " + connAck.getReturnCode())) - .doOnError(throwable -> System.out.println("Connection failed, " + throwable.getMessage())); - - final Flowable publishScenario = client.publish(messagesToPublish) - .doOnNext(publishResult -> System.out.println("Publish acknowledged: " + new String(publishResult.getPublish().getPayloadAsBytes()))); - - connectScenario.toCompletable().andThen(publishScenario).blockingSubscribe(); - - } -} diff --git a/perf/src/main/java/io/moquette/performance/clients/paho/PahoBenchmarker.java b/perf/src/main/java/io/moquette/performance/clients/paho/PahoBenchmarker.java deleted file mode 100644 index 5a3e4c95..00000000 --- a/perf/src/main/java/io/moquette/performance/clients/paho/PahoBenchmarker.java +++ /dev/null @@ -1,56 +0,0 @@ -package io.moquette.performance.clients.paho; - -import org.eclipse.paho.client.mqttv3.IMqttAsyncClient; -import org.eclipse.paho.client.mqttv3.MqttAsyncClient; -import org.eclipse.paho.client.mqttv3.MqttException; -import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; - -import java.nio.ByteBuffer; - -public class PahoBenchmarker { - - private static void publishQos0(IMqttAsyncClient client, long numMessages) throws MqttException { - System.out.println("publishing.."); - long startTime = System.currentTimeMillis(); - for (int i = 0; i < numMessages; i++) { - byte[] bytes = ByteBuffer.allocate(8).putLong(System.currentTimeMillis()).array(); - client.publish("topic", bytes, 0, false); - } - client.publish("/exit", "Exit".getBytes(), 0, false); - long stopTime = System.currentTimeMillis(); - long spentTime = stopTime -startTime; - System.out.println("published in " + spentTime + " ms"); - } - - public static void main(String[] args) throws MqttException, InterruptedException { - if (args.length < 2) { - System.out.println("Usage benchmarker "); - System.out.println("java io.moquette.performance.clients.paho.PahoBenchmarker localhost 1000 500 test1"); - return; - } - - String host = args[0]; - long numMessages = Long.parseLong(args[1]); - String tmpDir = System.getProperty("java.io.tmpdir"); - MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir); - //MemoryPersistence dataStore = new MemoryPersistence() - - int rnd = (int) (Math.random() * 100); - MqttAsyncClient client = new MqttAsyncClient("tcp://" + host +":1883", - "SubscriberClient" + rnd, dataStore); - SubscriberCallback callback = new SubscriberCallback(); - client.setCallback(callback); - client.connect().waitForCompletion(1000); - client.subscribe("topic", 0); - System.out.println("subscribed to topic"); - client.subscribe("/exit", 0); - System.out.println("subscribed to /exit"); - - publishQos0(client, numMessages); - - callback.waitFinish(); -// Map sorted = callback.distribution.sort( { k1, k2 -> k1 <=> k2 } as Comparator) -// File csv = new File("perf.csv") -// sorted.each {delay, count -> csv << "${delay}, ${count}\r\n"} - } -} diff --git a/perf/src/main/java/io/moquette/performance/clients/paho/SubscriberCallback.java b/perf/src/main/java/io/moquette/performance/clients/paho/SubscriberCallback.java deleted file mode 100644 index 3f133377..00000000 --- a/perf/src/main/java/io/moquette/performance/clients/paho/SubscriberCallback.java +++ /dev/null @@ -1,63 +0,0 @@ -package io.moquette.performance.clients.paho; - -import org.eclipse.paho.client.mqttv3.*; - -import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.CountDownLatch; - -class SubscriberCallback implements MqttCallback { - - int m_numReceived = 0; - long m_startTime; - boolean firstMessageReceived = false; - - private CountDownLatch m_latch = new CountDownLatch(1); - - //Maps delay to num of messages reached with that delay - protected Map distribution = new HashMap<>(); - - IMqttAsyncClient client; - - void waitFinish() throws InterruptedException { - m_latch.await(); - } - - public void connectionLost(Throwable cause) { - m_latch.countDown(); - } - - public void messageArrived(String topic, MqttMessage message) throws Exception { - //println "Received message ${new String(message.payload)} on topic [${topic}]" - if (!firstMessageReceived) { - m_startTime = System.currentTimeMillis(); - firstMessageReceived = true; - } - if (topic == "/exit") { - long stopTime = System.currentTimeMillis(); - long spentTime = stopTime - m_startTime; - System.out.println("/exit received"); - System.out.println("subscriber disconnected, received ${m_numReceived} messages in ${spentTime} ms (from first received to last one on topic /exit)"); - client.disconnect(); - m_latch.countDown(); - } else { - ByteBuffer bb = ByteBuffer.wrap(message.getPayload()); - long sentTime = bb.getLong(); - long delay = System.currentTimeMillis() - sentTime; - //println "-received in ${delay}(ms) on ${topic} with QoS ${message.qos}" - //distribution[delay] = distribution[delay] != null ? distribution[delay]++ : 1 - Long value = distribution.get(delay); - if (value != null) { - distribution.put(delay, value++); - } else { - distribution.put(delay, 1L); - } - m_numReceived++; - } - } - - public void deliveryComplete(IMqttDeliveryToken token) { - //To change body of implemented methods use File | Settings | File Templates. - } -} diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java deleted file mode 100644 index 3aae90c8..00000000 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ /dev/null @@ -1,104 +0,0 @@ -/* - * Copyright (c) 2012-2018 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.subscriptions; - -import java.io.IOException; -import java.lang.management.ManagementFactory; -import java.lang.management.ThreadMXBean; - -import io.moquette.broker.ISubscriptionsRepository; -import io.moquette.broker.MemorySubscriptionsRepository; -import io.moquette.persistence.MemoryStorageService; -import io.moquette.spi.ISessionsStore; -import io.netty.handler.codec.mqtt.MqttQoS; - -public final class PerformanceTest { - - // Set CPU governor and max frequency to lowest value for this test - /* debian: - cpufreq-set -g performance -u 800MHz -c 0 - cpufreq-set -g performance -u 800MHz -c 1 - cpufreq-set -g performance -u 800MHz -c 2 - cpufreq-set -g performance -u 800MHz -c 3 - */ - public static void main(String[] args) throws IOException { - ISubscriptionsDirectory store = new CTrieSubscriptionDirectory(); - MemoryStorageService memStore = new MemoryStorageService(null, null); - ISessionsStore aSessionsStore = memStore.sessionsStore(); - ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository(); - store.init(sessionsRepository); - - int times = 10000; - - int users = 500; - int devices = 5; - int values = 3; - int subs = 2; - int n = users * devices * values * subs; - - ThreadMXBean b = ManagementFactory.getThreadMXBean(); - - Topic[] topics = new Topic[n]; - for (int i = 0; i < n; i++) { - topics[i] = new Topic( - (i % users) + "/" + ((i / users) % devices) + "/" + ((i / (users * devices)) % values) + "/" - + ((i / (users * devices * values)) % subs)); - - Subscription sub = new Subscription("CLI_ID_" + (i % users), topics[i], MqttQoS.AT_MOST_ONCE); - aSessionsStore.subscriptionStore().addNewSubscription(sub); - store.add(sub); - } - - long min = Long.MAX_VALUE; - //AtomicLong min = new AtomicLong(Long.MAX_VALUE); - - System.out.println("Store size: " + store.size() + " " + n); - - // warm up - for (int i = 0; i < n; i++) { - store.matches(topics[i]); - } - - System.out.println("press return for start"); - System.in.read(); - -// IntStream.range(0, times) -// .parallel() -// .forEach(j -> { - - for (int j = 0; j < times; j++) { - - System.gc(); - Thread.yield(); - - long time = b.getCurrentThreadCpuTime(); - for (int i = 0; i < n; i++) { - store.matches(topics[i]); - } - long time2 = b.getCurrentThreadCpuTime() - time; - min = Math.min(min, time2); -// min.updateAndGet(c -> Math.min(c, time2)); - System.out.println(".." + j + ": " + time2 + " min:" + min); - } -// }); - - System.out.println("min:" + min); - } - - private PerformanceTest() { - } -} diff --git a/settings.gradle b/settings.gradle index e6b58d9d..f822d971 100644 --- a/settings.gradle +++ b/settings.gradle @@ -4,12 +4,9 @@ include ':moquette-distribution' include ':moquette-h2-storage' include ':moquette-mapdb-storage' include ':moquette-embedded-test' -include ':moquette-osgi-test' -include ':moquette-performance' project(':moquette-broker').projectDir = "$rootDir/broker" as File project(':moquette-distribution').projectDir = "$rootDir/distribution" as File project(':moquette-embedded-test').projectDir = "$rootDir/embedding_moquette" as File -project(':moquette-performance').projectDir = "$rootDir/perf" as File project(':moquette-h2-storage').projectDir = "$rootDir/h2_storage" as File project(':moquette-mapdb-storage').projectDir = "$rootDir/mapdb_storage" as File