You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2024/05/10 13:15:58 UTC
(rocketmq-mqtt) branch develop updated (95fbf79 -> 90cfe7c)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
from 95fbf79 feat: add metrics for topic dimension read and write msg size (#271)
new 6466344 feat: use openssl SSL provider to create TLS/SSL engine, adding netty-tcnative to cs module
new c768b77 fix: socket option keep_alive shall be a child option in netty
new 0643cd3 feat: support WebSocket over TLS transport
new 318841e feat: use shared netty event loop for TCP/TCP over TLS/Web Socket
new 90cfe7c feat: add experimental support for QUIC transport
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/quic.md | 16 +++
mqtt-common/pom.xml | 5 +
.../util/{SpringUtils.java => NetworkHelper.java} | 30 ++--
mqtt-cs/pom.xml | 11 +-
.../mqtt/cs/channel/AdaptiveTlsHandler.java | 93 ++++++++++++
.../rocketmq/mqtt/cs/config/ConnectConf.java | 23 +++
.../rocketmq/mqtt/cs/protocol/ssl/SslFactory.java | 30 +++-
.../rocketmq/mqtt/cs/starter/MqttServer.java | 158 +++++++++++++++++----
pom.xml | 21 ++-
9 files changed, 339 insertions(+), 48 deletions(-)
create mode 100644 docs/quic.md
copy mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/{SpringUtils.java => NetworkHelper.java} (56%)
create mode 100644 mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/AdaptiveTlsHandler.java
(rocketmq-mqtt) 04/05: feat: use shared netty event loop for TCP/TCP over TLS/Web Socket
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
commit 318841e9acb0e6a1c0d846573e6efbd612984970
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Tue Apr 30 10:07:32 2024 +0800
feat: use shared netty event loop for TCP/TCP over TLS/Web Socket
Signed-off-by: Li Zhanhui <li...@gmail.com>
---
.../rocketmq/mqtt/cs/starter/MqttServer.java | 56 +++++++++++++++-------
1 file changed, 38 insertions(+), 18 deletions(-)
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
index 94d67ff..133c08f 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
@@ -32,6 +32,7 @@ import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
+import javax.annotation.PreDestroy;
import org.apache.rocketmq.mqtt.cs.channel.AdaptiveTlsHandler;
import org.apache.rocketmq.mqtt.cs.channel.ConnectHandler;
import org.apache.rocketmq.mqtt.cs.config.ConnectConf;
@@ -50,11 +51,11 @@ import java.net.InetSocketAddress;
@Service
public class MqttServer {
- private static Logger logger = LoggerFactory.getLogger(MqttServer.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(MqttServer.class);
- private ServerBootstrap serverBootstrap = new ServerBootstrap();
- private ServerBootstrap wsServerBootstrap = new ServerBootstrap();
- private ServerBootstrap tlsServerBootstrap = new ServerBootstrap();
+ private final ServerBootstrap serverBootstrap = new ServerBootstrap();
+ private final ServerBootstrap wsServerBootstrap = new ServerBootstrap();
+ private final ServerBootstrap tlsServerBootstrap = new ServerBootstrap();
@Resource
private ConnectHandler connectHandler;
@@ -71,10 +72,17 @@ public class MqttServer {
@Resource
private SslFactory sslFactory;
+ private NioEventLoopGroup acceptorEventLoopGroup;
+
+ private NioEventLoopGroup workerEventLoopGroup;
+
private AdaptiveTlsHandler adaptiveTlsHandler;
@PostConstruct
public void init() throws Exception {
+ acceptorEventLoopGroup = new NioEventLoopGroup(connectConf.getNettySelectorThreadNum());
+ workerEventLoopGroup = new NioEventLoopGroup(connectConf.getNettyWorkerThreadNum());
+
adaptiveTlsHandler = new AdaptiveTlsHandler(TlsMode.PERMISSIVE, sslFactory);
start();
@@ -82,10 +90,21 @@ public class MqttServer {
startTls();
}
+ @PreDestroy
+ public void shutdown() {
+ if (null != acceptorEventLoopGroup) {
+ acceptorEventLoopGroup.shutdownGracefully();
+ }
+
+ if (null != workerEventLoopGroup) {
+ workerEventLoopGroup.shutdownGracefully();
+ }
+ }
+
private void start() {
int port = connectConf.getMqttPort();
serverBootstrap
- .group(new NioEventLoopGroup(connectConf.getNettySelectorThreadNum()), new NioEventLoopGroup(connectConf.getNettyWorkerThreadNum()))
+ .group(acceptorEventLoopGroup, workerEventLoopGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 8 * 1024)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
@@ -96,14 +115,11 @@ public class MqttServer {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
- pipeline.addLast("connectHandler", connectHandler);
- pipeline.addLast("decoder", new MqttDecoder(connectConf.getMaxPacketSizeInByte()));
- pipeline.addLast("encoder", MqttEncoder.INSTANCE);
- pipeline.addLast("dispatcher", mqttPacketDispatcher);
+ assembleHandlerPipeline(pipeline);
}
});
serverBootstrap.bind();
- logger.warn("start mqtt server , port:{}", port);
+ LOGGER.info("MQTT server for TCP started, listening: {}", port);
}
private void startTls() {
@@ -113,7 +129,7 @@ public class MqttServer {
int tlsPort = connectConf.getMqttTlsPort();
tlsServerBootstrap
- .group(new NioEventLoopGroup(connectConf.getNettySelectorThreadNum()), new NioEventLoopGroup(connectConf.getNettyWorkerThreadNum()))
+ .group(acceptorEventLoopGroup, workerEventLoopGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 8 * 1024)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
@@ -125,14 +141,18 @@ public class MqttServer {
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("sslHandler", new SslHandler(sslFactory.buildSslEngine(ch)));
- pipeline.addLast("connectHandler", connectHandler);
- pipeline.addLast("decoder", new MqttDecoder(connectConf.getMaxPacketSizeInByte()));
- pipeline.addLast("encoder", MqttEncoder.INSTANCE);
- pipeline.addLast("dispatcher", mqttPacketDispatcher);
+ assembleHandlerPipeline(pipeline);
}
});
tlsServerBootstrap.bind();
- logger.warn("start mqtt tls server , port:{}", tlsPort);
+ LOGGER.info("MQTT server for TCP over TLS started, listening: {}", tlsPort);
+ }
+
+ private void assembleHandlerPipeline(ChannelPipeline pipeline) {
+ pipeline.addLast("connectHandler", connectHandler);
+ pipeline.addLast("decoder", new MqttDecoder(connectConf.getMaxPacketSizeInByte()));
+ pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+ pipeline.addLast("dispatcher", mqttPacketDispatcher);
}
/**
@@ -151,7 +171,7 @@ public class MqttServer {
private void startWs() {
int port = connectConf.getMqttWsPort();
wsServerBootstrap
- .group(new NioEventLoopGroup(connectConf.getNettySelectorThreadNum()), new NioEventLoopGroup(connectConf.getNettyWorkerThreadNum()))
+ .group(acceptorEventLoopGroup, workerEventLoopGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 8 * 1024)
.childOption(ChannelOption.SO_KEEPALIVE, true)
@@ -176,7 +196,7 @@ public class MqttServer {
}
});
wsServerBootstrap.bind();
- logger.warn("start mqtt ws server , port:{}", port);
+ LOGGER.info("MQTT server for WebSocket started, listening: {}", port);
}
}
(rocketmq-mqtt) 01/05: feat: use openssl SSL provider to create TLS/SSL engine, adding netty-tcnative to cs module
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
commit 6466344c9171d17908bd3801dae2bd24430bd5f9
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Apr 11 23:56:22 2024 +0800
feat: use openssl SSL provider to create TLS/SSL engine, adding netty-tcnative to cs module
Signed-off-by: Li Zhanhui <li...@gmail.com>
---
mqtt-cs/pom.xml | 6 +++++-
.../java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java | 8 +++++++-
pom.xml | 7 ++++---
3 files changed, 16 insertions(+), 5 deletions(-)
diff --git a/mqtt-cs/pom.xml b/mqtt-cs/pom.xml
index e3924a1..c1c4521 100644
--- a/mqtt-cs/pom.xml
+++ b/mqtt-cs/pom.xml
@@ -30,6 +30,10 @@
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-tcnative-boringssl-static</artifactId>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
@@ -78,4 +82,4 @@
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
index d2cfbd8..b9cc770 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
@@ -56,6 +56,12 @@ public class SslFactory {
return;
}
+ SslProvider sslProvider = OpenSsl.isAvailable() ? SslProvider.OPENSSL : SslProvider.JDK;
+ if (OpenSsl.isAvailable()) {
+ LOG.info("OpenSSL is available");
+ } else {
+ LOG.warn("OpenSSL is NOT available, falling back to JDK SslEngine");
+ }
try {
File sslCertFile = new File(connectConf.getSslServerCertFile());
File sslKeyFile = new File(connectConf.getSslServerKeyFile());
@@ -64,7 +70,7 @@ public class SslFactory {
SslContextBuilder contextBuilder = SslContextBuilder.forServer(sslCertFile, sslKeyFile,
StringUtils.isBlank(password) ? null : password);
contextBuilder.clientAuth(ClientAuth.OPTIONAL);
- contextBuilder.sslProvider(OpenSsl.isAvailable() ? SslProvider.OPENSSL : SslProvider.JDK);
+ contextBuilder.sslProvider(sslProvider);
if (connectConf.isNeedClientAuth()) {
LOG.info("client tls authentication is required.");
contextBuilder.clientAuth(ClientAuth.REQUIRE);
diff --git a/pom.xml b/pom.xml
index c99cc65..6826ab6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -47,7 +47,8 @@
<rpc-grpc-impl.version>1.3.8</rpc-grpc-impl.version>
<guava.version>32.0.0-jre</guava.version>
<jraft-core.version>1.3.11</jraft-core.version>
-
+ <netty.version>4.1.43.Final</netty.version>
+ <netty.tcnative.version>2.0.26.Final</netty.tcnative.version>
</properties>
<dependencyManagement>
@@ -95,12 +96,12 @@
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
- <version>4.1.43.Final</version>
+ <version>${netty.version}</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
- <version>2.0.26.Final</version>
+ <version>${netty.tcnative.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
(rocketmq-mqtt) 03/05: feat: support WebSocket over TLS transport
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
commit 0643cd3a7ee95bbd32cb9f7320c08cbe5429360f
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Tue Apr 30 11:40:50 2024 +0800
feat: support WebSocket over TLS transport
Signed-off-by: Li Zhanhui <li...@gmail.com>
---
.../mqtt/cs/channel/AdaptiveTlsHandler.java | 93 ++++++++++++++++++++++
.../rocketmq/mqtt/cs/protocol/ssl/SslFactory.java | 4 +
.../rocketmq/mqtt/cs/starter/MqttServer.java | 20 +++++
3 files changed, 117 insertions(+)
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/AdaptiveTlsHandler.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/AdaptiveTlsHandler.java
new file mode 100644
index 0000000..d2271ac
--- /dev/null
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/AdaptiveTlsHandler.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.mqtt.cs.channel;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.ssl.SslHandler;
+import java.util.NoSuchElementException;
+import org.apache.rocketmq.mqtt.cs.protocol.ssl.SslFactory;
+import org.apache.rocketmq.remoting.common.TlsMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@ChannelHandler.Sharable
+public class AdaptiveTlsHandler extends SimpleChannelInboundHandler<ByteBuf> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(AdaptiveTlsHandler.class);
+
+ private final TlsMode tlsMode;
+
+ private static final byte HANDSHAKE_MAGIC_CODE = 0x16;
+
+ private final SslFactory sslFactory;
+
+ public AdaptiveTlsHandler(TlsMode tlsMode, SslFactory sslFactory) {
+ this.tlsMode = tlsMode;
+ this.sslFactory = sslFactory;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
+
+ // Peek the current read index byte to determine if the content is starting with TLS handshake
+ byte b = msg.getByte(msg.readerIndex());
+
+ if (b == HANDSHAKE_MAGIC_CODE) {
+ switch (tlsMode) {
+ case DISABLED:
+ ctx.close();
+ LOGGER.warn("Clients intend to establish an SSL connection while this server is running in SSL disabled mode");
+ throw new UnsupportedOperationException("The NettyRemotingServer in SSL disabled mode doesn't support ssl client");
+ case PERMISSIVE:
+ case ENFORCING:
+ if (null != sslFactory.getSslContext()) {
+ ctx.pipeline()
+ .addAfter(AdaptiveTlsHandler.class.getSimpleName(), SslHandler.class.getSimpleName(),
+ sslFactory.getSslContext().newHandler(ctx.alloc()));
+ LOGGER.info("Handlers prepended to channel pipeline to establish SSL connection");
+ } else {
+ ctx.close();
+ LOGGER.error("Trying to establish an SSL connection but sslContext is null");
+ }
+ break;
+
+ default:
+ LOGGER.warn("Unknown TLS mode");
+ break;
+ }
+ } else if (tlsMode == TlsMode.ENFORCING) {
+ ctx.close();
+ LOGGER.warn("Clients intend to establish an insecure connection while this server is running in SSL enforcing mode");
+ }
+
+ try {
+ // Remove this handler
+ ctx.pipeline().remove(this);
+ } catch (NoSuchElementException e) {
+ LOGGER.error("Error while removing TlsModeHandler", e);
+ }
+
+ // Hand over this message to the next handler.
+ ctx.fireChannelRead(msg.retain());
+ }
+
+
+}
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
index b9cc770..08d776e 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
@@ -53,6 +53,7 @@ public class SslFactory {
@PostConstruct
private void initSslContext() {
if (!connectConf.isEnableTlsSever()) {
+ LOG.info("SSL/TLS is disabled");
return;
}
@@ -139,5 +140,8 @@ public class SslFactory {
return sslEngine;
}
+ public SslContext getSslContext() {
+ return sslContext;
+ }
}
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
index f4d0183..94d67ff 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
@@ -32,12 +32,14 @@ import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
+import org.apache.rocketmq.mqtt.cs.channel.AdaptiveTlsHandler;
import org.apache.rocketmq.mqtt.cs.channel.ConnectHandler;
import org.apache.rocketmq.mqtt.cs.config.ConnectConf;
import org.apache.rocketmq.mqtt.cs.protocol.mqtt.MqttPacketDispatcher;
import org.apache.rocketmq.mqtt.cs.protocol.ssl.SslFactory;
import org.apache.rocketmq.mqtt.cs.protocol.ws.WebSocketServerHandler;
import org.apache.rocketmq.mqtt.cs.protocol.ws.WebSocketEncoder;
+import org.apache.rocketmq.remoting.common.TlsMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@@ -69,8 +71,12 @@ public class MqttServer {
@Resource
private SslFactory sslFactory;
+ private AdaptiveTlsHandler adaptiveTlsHandler;
+
@PostConstruct
public void init() throws Exception {
+ adaptiveTlsHandler = new AdaptiveTlsHandler(TlsMode.PERMISSIVE, sslFactory);
+
start();
startWs();
startTls();
@@ -129,6 +135,19 @@ public class MqttServer {
logger.warn("start mqtt tls server , port:{}", tlsPort);
}
+ /**
+ * Support Web Socket Transport.
+ * </p>
+ *
+ * Both WebSocket and WebSocket-Over-TLS/SSL are supported.
+ * </p>
+ *
+ * Clients are supposed to use one of the following Server URIs to connect:
+ * <ul>
+ * <li>ws://host:port/mqtt</li>
+ * <li>wss://host:port/mqtt</li>
+ * </ul>
+ */
private void startWs() {
int port = connectConf.getMqttWsPort();
wsServerBootstrap
@@ -144,6 +163,7 @@ public class MqttServer {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
+ pipeline.addLast(AdaptiveTlsHandler.class.getSimpleName(), adaptiveTlsHandler);
pipeline.addLast("connectHandler", connectHandler);
pipeline.addLast("http-codec", new HttpServerCodec(1024, 32 * 1024, connectConf.getMaxPacketSizeInByte() * 2, true));
pipeline.addLast("aggregator", new HttpObjectAggregator(connectConf.getMaxPacketSizeInByte() * 2));
(rocketmq-mqtt) 05/05: feat: add experimental support for QUIC transport
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
commit 90cfe7c90647e738963fe1e21489aff19865742d
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Apr 12 19:13:03 2024 +0800
feat: add experimental support for QUIC transport
---
docs/quic.md | 16 +++++
mqtt-common/pom.xml | 5 ++
.../rocketmq/mqtt/common/util/NetworkHelper.java | 35 ++++++++++
mqtt-cs/pom.xml | 5 ++
.../rocketmq/mqtt/cs/config/ConnectConf.java | 23 +++++++
.../rocketmq/mqtt/cs/protocol/ssl/SslFactory.java | 18 ++++-
.../rocketmq/mqtt/cs/starter/MqttServer.java | 80 ++++++++++++++++++++--
pom.xml | 14 ++++
8 files changed, 190 insertions(+), 6 deletions(-)
diff --git a/docs/quic.md b/docs/quic.md
new file mode 100644
index 0000000..025d71a
--- /dev/null
+++ b/docs/quic.md
@@ -0,0 +1,16 @@
+## Setup
+Download emqtt-bench with QUIC feature
+```shell
+wget https://github.com/emqx/emqtt-bench/releases/download/0.4.18/emqtt-bench-0.4.18-macos13-amd64-quic.zip
+```
+
+## Test Connect
+```shell
+./bin/emqtt_bench conn -u user0 -P secret0 -V 4 --quic true -h localhost -p 14567 -c 10
+```
+
+### Pub
+
+```shell
+./bin/emqtt_bench pub -u user0 -P secret0 -V 4 --quic true -h localhost -p 14567 -c 1 -t T_Event/test
+```
diff --git a/mqtt-common/pom.xml b/mqtt-common/pom.xml
index 1edfa50..47d1c0d 100644
--- a/mqtt-common/pom.xml
+++ b/mqtt-common/pom.xml
@@ -68,6 +68,11 @@
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty.incubator</groupId>
+ <artifactId>netty-incubator-codec-native-quic</artifactId>
+ <classifier>${os.detected.classifier}</classifier>
+ </dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
diff --git a/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/NetworkHelper.java b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/NetworkHelper.java
new file mode 100644
index 0000000..769cd89
--- /dev/null
+++ b/mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/NetworkHelper.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.mqtt.common.util;
+
+import io.netty.incubator.codec.quic.QuicStreamAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import org.apache.rocketmq.common.utils.NetworkUtil;
+
+public class NetworkHelper {
+ public static String socketAddressToString(SocketAddress address) {
+ if (address instanceof InetSocketAddress) {
+ return NetworkUtil.socketAddress2String(address);
+ } else if (address instanceof QuicStreamAddress) {
+ QuicStreamAddress quicAddress = (QuicStreamAddress) address;
+ return String.format("quic[%d]", quicAddress.streamId());
+ }
+ throw new RuntimeException("Unsupported SocketAddress type");
+ }
+}
diff --git a/mqtt-cs/pom.xml b/mqtt-cs/pom.xml
index c1c4521..da12f23 100644
--- a/mqtt-cs/pom.xml
+++ b/mqtt-cs/pom.xml
@@ -34,6 +34,11 @@
<groupId>io.netty</groupId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty.incubator</groupId>
+ <artifactId>netty-incubator-codec-native-quic</artifactId>
+ <classifier>${os.detected.classifier}</classifier>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/config/ConnectConf.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/config/ConnectConf.java
index f558023..dcab351 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/config/ConnectConf.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/config/ConnectConf.java
@@ -32,9 +32,24 @@ public class ConnectConf {
private File confFile;
private int nettySelectorThreadNum = 1;
private int nettyWorkerThreadNum = Runtime.getRuntime().availableProcessors() * 2;
+
+ /**
+ * TCP ports 8883 and 1883 are registered with IANA for MQTT TLS and non-TLS communication respectively.
+ */
private int mqttPort = 1883;
+
+ /**
+ * TCP ports 8883 and 1883 are registered with IANA for MQTT TLS and non-TLS communication respectively.
+ */
private int mqttTlsPort = 8883;
+
+ /**
+ * Uses this port for both normal WebSocket connections and WebSocket connections over TLS as well.
+ */
private int mqttWsPort = 8888;
+
+ private int quicPort = 14567;
+
private boolean enableTlsSever = false;
private boolean needClientAuth = false;
private String sslCaCertFile;
@@ -276,4 +291,12 @@ public class ConnectConf {
public void setMaxTransferCountOnMessageInDisk(int maxTransferCountOnMessageInDisk) {
this.maxTransferCountOnMessageInDisk = maxTransferCountOnMessageInDisk;
}
+
+ public int getQuicPort() {
+ return quicPort;
+ }
+
+ public void setQuicPort(int quicPort) {
+ this.quicPort = quicPort;
+ }
}
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
index 08d776e..f214eae 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
@@ -24,6 +24,10 @@ import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslProvider;
import org.apache.commons.lang3.StringUtils;
+import io.netty.handler.ssl.util.SelfSignedCertificate;
+import io.netty.incubator.codec.quic.QuicSslContext;
+import io.netty.incubator.codec.quic.QuicSslContextBuilder;
+import java.security.cert.CertificateException;
import org.apache.rocketmq.mqtt.cs.config.ConnectConf;
import org.apache.rocketmq.srvutil.FileWatchService;
import org.slf4j.Logger;
@@ -50,6 +54,8 @@ public class SslFactory {
private FileWatchService fileWatchService;
+ private QuicSslContext quicSslContext;
+
@PostConstruct
private void initSslContext() {
if (!connectConf.isEnableTlsSever()) {
@@ -78,7 +84,13 @@ public class SslFactory {
contextBuilder.trustManager(new File(connectConf.getSslCaCertFile()));
}
sslContext = contextBuilder.build();
- } catch (IOException e) {
+
+ SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
+ quicSslContext = QuicSslContextBuilder.forServer(
+ selfSignedCertificate.privateKey(), null, selfSignedCertificate.certificate())
+ .applicationProtocols("mqtt")
+ .build();
+ } catch (IOException | CertificateException e) {
throw new RuntimeException("failed to initialize ssl context.", e);
}
}
@@ -143,5 +155,9 @@ public class SslFactory {
public SslContext getSslContext() {
return sslContext;
}
+
+ public QuicSslContext getQuicSslContext() {
+ return quicSslContext;
+ }
}
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
index 133c08f..5bf16ad 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
@@ -17,14 +17,19 @@
package org.apache.rocketmq.mqtt.cs.starter;
+import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
@@ -33,8 +38,13 @@ import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import javax.annotation.PreDestroy;
-import org.apache.rocketmq.mqtt.cs.channel.AdaptiveTlsHandler;
+import io.netty.incubator.codec.quic.InsecureQuicTokenHandler;
+import io.netty.incubator.codec.quic.QuicChannel;
+import io.netty.incubator.codec.quic.QuicServerCodecBuilder;
+import io.netty.incubator.codec.quic.QuicStreamChannel;
+import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.mqtt.cs.channel.ConnectHandler;
+import org.apache.rocketmq.mqtt.cs.channel.AdaptiveTlsHandler;
import org.apache.rocketmq.mqtt.cs.config.ConnectConf;
import org.apache.rocketmq.mqtt.cs.protocol.mqtt.MqttPacketDispatcher;
import org.apache.rocketmq.mqtt.cs.protocol.ssl.SslFactory;
@@ -57,6 +67,8 @@ public class MqttServer {
private final ServerBootstrap wsServerBootstrap = new ServerBootstrap();
private final ServerBootstrap tlsServerBootstrap = new ServerBootstrap();
+ private final Bootstrap quicBootstrap = new Bootstrap();
+
@Resource
private ConnectHandler connectHandler;
@@ -86,8 +98,12 @@ public class MqttServer {
adaptiveTlsHandler = new AdaptiveTlsHandler(TlsMode.PERMISSIVE, sslFactory);
start();
- startWs();
startTls();
+
+ startWs();
+
+ // QUIC over DTLS
+ startQuic();
}
@PreDestroy
@@ -113,7 +129,7 @@ public class MqttServer {
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
- public void initChannel(SocketChannel ch) throws Exception {
+ public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
assembleHandlerPipeline(pipeline);
}
@@ -138,7 +154,7 @@ public class MqttServer {
.localAddress(new InetSocketAddress(tlsPort))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
- public void initChannel(SocketChannel ch) throws Exception {
+ public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("sslHandler", new SslHandler(sslFactory.buildSslEngine(ch)));
assembleHandlerPipeline(pipeline);
@@ -181,7 +197,7 @@ public class MqttServer {
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
- public void initChannel(SocketChannel ch) throws Exception {
+ public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(AdaptiveTlsHandler.class.getSimpleName(), adaptiveTlsHandler);
pipeline.addLast("connectHandler", connectHandler);
@@ -199,4 +215,58 @@ public class MqttServer {
LOGGER.info("MQTT server for WebSocket started, listening: {}", port);
}
+ private void startQuic() throws InterruptedException {
+ ChannelHandler channelHandler = new QuicServerCodecBuilder()
+ .sslContext(sslFactory.getQuicSslContext())
+ .maxIdleTimeout(5000, TimeUnit.SECONDS)
+ // Configure some limits for the maximal number of streams (and the data) that we want to handle.
+ .initialMaxData(10000000)
+ .initialMaxStreamDataBidirectionalLocal(1000000)
+ .initialMaxStreamDataBidirectionalRemote(1000000)
+ .initialMaxStreamsBidirectional(100)
+ .initialMaxStreamsUnidirectional(100)
+ .activeMigration(true)
+
+ // Setup a token handler. In a production system you would want to implement and provide your custom
+ // one.
+ .tokenHandler(InsecureQuicTokenHandler.INSTANCE)
+ // ChannelHandler that is added into QuicChannel pipeline.
+ .handler(new ChannelInboundHandlerAdapter() {
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) {
+ QuicChannel channel = (QuicChannel) ctx.channel();
+ LOGGER.debug("QUIC Connection Established: remote={}", channel.remoteAddress());
+ // Create streams etc..
+ }
+
+ public void channelInactive(ChannelHandlerContext ctx) {
+ ((QuicChannel) ctx.channel()).collectStats().addListener(f -> {
+ if (f.isSuccess()) {
+ LOGGER.info("Connection closed: {}", f.getNow());
+ }
+ });
+ }
+
+ @Override
+ public boolean isSharable() {
+ return true;
+ }
+ })
+ .streamHandler(new ChannelInitializer<QuicStreamChannel>() {
+ @Override
+ protected void initChannel(QuicStreamChannel ch) {
+ ChannelPipeline pipeline = ch.pipeline();
+ assembleHandlerPipeline(pipeline);
+ }
+ })
+ .build();
+
+ quicBootstrap.group(workerEventLoopGroup)
+ .channel(NioDatagramChannel.class)
+ .handler(channelHandler)
+ .bind(new InetSocketAddress(connectConf.getQuicPort()))
+ .sync();
+ LOGGER.info("MQTT server for QUIC over DTLS started, listening: {}", connectConf.getQuicPort());
+ }
+
}
diff --git a/pom.xml b/pom.xml
index 6826ab6..e863aa3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -49,6 +49,7 @@
<jraft-core.version>1.3.11</jraft-core.version>
<netty.version>4.1.43.Final</netty.version>
<netty.tcnative.version>2.0.26.Final</netty.tcnative.version>
+ <netty.quic.version>0.0.62.Final</netty.quic.version>
</properties>
<dependencyManagement>
@@ -103,6 +104,12 @@
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${netty.tcnative.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.netty.incubator</groupId>
+ <artifactId>netty-incubator-codec-native-quic</artifactId>
+ <version>${netty.quic.version}</version>
+ <classifier>${os.detected.classifier}</classifier>
+ </dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
@@ -262,6 +269,13 @@
</dependencyManagement>
<build>
+ <extensions>
+ <extension>
+ <groupId>kr.motd.maven</groupId>
+ <artifactId>os-maven-plugin</artifactId>
+ <version>1.7.0</version>
+ </extension>
+ </extensions>
<plugins>
<plugin>
<artifactId>maven-resources-plugin</artifactId>
(rocketmq-mqtt) 02/05: fix: socket option keep_alive shall be a child option in netty
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
commit c768b77f9245a56005156156627df9c289cbd38a
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Wed Apr 17 13:38:22 2024 +0800
fix: socket option keep_alive shall be a child option in netty
Signed-off-by: Li Zhanhui <li...@gmail.com>
---
.../src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
index 81a77b2..f4d0183 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
@@ -135,7 +135,7 @@ public class MqttServer {
.group(new NioEventLoopGroup(connectConf.getNettySelectorThreadNum()), new NioEventLoopGroup(connectConf.getNettyWorkerThreadNum()))
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 8 * 1024)
- .option(ChannelOption.SO_KEEPALIVE, true)
+ .childOption(ChannelOption.SO_KEEPALIVE, true)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,new WriteBufferWaterMark(connectConf.getLowWater(), connectConf.getHighWater()))
.childOption(ChannelOption.TCP_NODELAY, true)
@@ -159,4 +159,4 @@ public class MqttServer {
logger.warn("start mqtt ws server , port:{}", port);
}
-}
\ No newline at end of file
+}