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

Removed Moquette parser module

Browse Source
This commit is contained in:
andsel
2017-03-05 12:37:32 +01:00
parent e5c46e7a55
commit 46d27e3bea
83 changed files with 83 additions and 5557 deletions
Download Patch File Download Diff File Expand all files Collapse all files
perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java
+47 -47
View File
@@ -1,23 +1,21 @@
package io.moquette.parser.netty.performance;
import io.moquette.parser.proto.messages.*;
import static io.moquette.parser.proto.messages.AbstractMessage.*;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.moquette.parser.proto.Utils;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import org.HdrHistogram.Histogram;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str;
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;
/**
* Created by andrea on 6/2/15.
*/
@ChannelHandler.Sharable
class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
@@ -32,27 +30,26 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
AbstractMessage msg = (AbstractMessage) message;
MqttMessage msg = (MqttMessage) message;
MqttMessageType messageType = msg.fixedHeader().messageType();
String clientID = clientID(ctx.channel());
try {
switch (msg.getMessageType()) {
switch (messageType) {
case CONNECT:
ConnectMessage connect = (ConnectMessage) message;
clientID = connect.getClientID();
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
MqttConnectMessage connect = (MqttConnectMessage) message;
handleConnect(ctx, connect);
return;
case SUBSCRIBE:
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
handleSubscribe(ctx, (SubscribeMessage) msg);
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 <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
handlePublish(ctx, (PublishMessage) msg);
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 <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
LOG.info("Received a message of type {} from <{}>", messageType, clientID);
ctx.close();
// case PUBACK:
// NettyChannel channel;
@@ -66,36 +63,36 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
// m_messaging.handleProtocolMessage(channel, msg);
break;
case PINGREQ:
PingRespMessage pingResp = new PingRespMessage();
MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, AT_MOST_ONCE, false, 0);
MqttMessage pingResp = new MqttMessage(pingHeader);
ctx.writeAndFlush(pingResp);
break;
default:
LOG.info("Received a message of type {} from <{}>", Utils.msgType2String(msg.getMessageType()), clientID);
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, PublishMessage msg) {
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.getTopicName());
LOG.debug("content <{}>", payload2Str(msg.getPayload()));
String decodedPayload = payload2Str(msg.getPayload());
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
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(msg.getTopicName());
pubMessage.setQos(QOSType.MOST_ONE);
pubMessage.setPayload(msg.getPayload());
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);
/*Channel subscriberCh = m_state.getSubscriberCh();
@@ -106,28 +103,24 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
}*/
long stop = System.nanoTime();
processingTime.recordValue(stop - start);
LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.getPayload()));
LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.payload()));
}
private void handleSubscribe(ChannelHandlerContext ctx, SubscribeMessage msg) {
private void handleSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {
m_state.setForwardable(true);
LOG.debug(" new value of flag {}, LoopHandler instance is {}", m_state.isForwardable(), this);
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
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);
LOG.debug("subscribed client to {}", msg.subscriptions());
LOG.debug("subscribed client to {}", msg.payload().topicSubscriptions());
}
static String payload2Str(ByteBuffer content) {
byte[] b = new byte[content.remaining()];
content.mark();
content.get(b);
content.reset();
return new String(b);
}
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);
private void handleConnect(ChannelHandlerContext ctx, ConnectMessage msg) {
String clientID = msg.getClientID();
clientID(ctx.channel(), clientID);
if (clientID.toLowerCase().startsWith("sub")) {
m_state.setSubscriberCh(ctx.channel());
@@ -135,14 +128,21 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter {
m_state.setPublisherCh(ctx.channel());
} else {
//we don't admit other names
ConnAckMessage koResp = new ConnAckMessage();
koResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
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);
ctx.close();
return;
}
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
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);
}
perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java
+5 -6
View File
@@ -1,13 +1,12 @@
package io.moquette.parser.netty.performance;
import io.moquette.parser.commons.Constants;
import io.moquette.parser.netty.MQTTDecoder;
import io.moquette.parser.netty.MQTTEncoder;
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;
@@ -87,10 +86,10 @@ public class ProtocolDecodingServer {
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
try {
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, 2));
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler());
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
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);
perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java
+2 -4
View File
@@ -1,7 +1,5 @@
package io.moquette.parser.netty.performance;
import io.moquette.parser.netty.MQTTDecoder;
import io.moquette.parser.netty.MQTTEncoder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
@@ -39,8 +37,8 @@ public class ProtocolPublishDecodingServer {
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
try {
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
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);
perf/src/main/java/io/moquette/parser/netty/performance/PublishBomber.java
+13 -15
View File
@@ -1,21 +1,18 @@
package io.moquette.parser.netty.performance;
import io.moquette.parser.netty.MQTTDecoder;
import io.moquette.parser.netty.MQTTEncoder;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.PublishMessage;
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.eclipse.jetty.toolchain.perf.PlatformTimer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
class PublishBomber {
private static final Logger LOG = LoggerFactory.getLogger(PublishBomber.class);
@@ -34,8 +31,8 @@ class PublishBomber {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
}
});
@@ -47,7 +44,7 @@ class PublishBomber {
}
}
private void sendMessage(AbstractMessage msg) {
private void sendMessage(MqttMessage msg) {
try {
channel.writeAndFlush(msg).await();
} catch (InterruptedException e) {
@@ -67,13 +64,14 @@ class PublishBomber {
PlatformTimer timer = PlatformTimer.detect();
for (int i=0; i < numToSend; i++) {
long nanos = System.nanoTime();
PublishMessage pubMessage = new PublishMessage();
pubMessage.setQos(AbstractMessage.QOSType.MOST_ONE);
pubMessage.setTopicName("/topic");
byte[] rawContent = ("Hello world!!-" + nanos).getBytes();
ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(rawContent.length).put(rawContent).flip();
pubMessage.setPayload(payload);
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);
}
perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java
+15 -26
View File
@@ -1,19 +1,16 @@
package io.moquette.parser.netty.performance;
import io.moquette.parser.proto.Utils;
import io.moquette.parser.proto.messages.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;
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 java.nio.ByteBuffer;
import static io.moquette.parser.proto.messages.AbstractMessage.*;
import static io.moquette.parser.netty.performance.NettyPublishReceiverHandler.payload2Str;
@ChannelHandler.Sharable
class PublishReceiverHandler extends ChannelInboundHandlerAdapter {
@@ -26,43 +23,35 @@ class PublishReceiverHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
AbstractMessage msg = (AbstractMessage) message;
MqttMessage msg = (MqttMessage) message;
MqttMessageType messageType = msg.fixedHeader().messageType();
try {
switch (msg.getMessageType()) {
switch (messageType) {
case PUBLISH:
LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
handlePublish((PublishMessage) msg);
LOG.info("Received a message of type {}", messageType);
handlePublish((MqttPublishMessage) msg);
return;
default:
LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
LOG.info("Received a message of type {}", messageType);
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
private void handlePublish(PublishMessage msg) {
private void handlePublish(MqttPublishMessage msg) {
long start = System.nanoTime();
LOG.debug("push forward message the topic {}", msg.getTopicName());
LOG.debug("content <{}>", payload2Str(msg.getPayload()));
String decodedPayload = payload2Str(msg.getPayload());
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.getPayload()));
LOG.info("Request processed in {} ns, matching {}", stop - start, payload2Str(msg.payload()));
}
static String payload2Str(ByteBuffer content) {
byte[] b = new byte[content.remaining()];
content.mark();
content.get(b);
content.reset();
return new String(b);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
LOG.info("Received channel inactive");