This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
The following commit(s) were added to refs/heads/main by this push:
new 9259af6 [ISSUE #71] Rename WebsocketEncoder to WebSocketEncoder and
add tests
9259af6 is described below
commit 9259af6069760d8bcb7a337c4eb67f956877948d
Author: YongXing <[email protected]>
AuthorDate: Mon Apr 18 11:11:47 2022 +0800
[ISSUE #71] Rename WebsocketEncoder to WebSocketEncoder and add tests
Co-authored-by: AhaThinking <[email protected]>
---
...WebsocketEncoder.java => WebSocketEncoder.java} | 2 +-
.../rocketmq/mqtt/cs/starter/MqttServer.java | 4 +-
.../cs/test/protocol/ws/TestWebSocketEncoder.java | 51 +++++++++++++++
.../protocol/ws/TestWebSocketServerHandler.java | 72 ++++++++++++++++++++++
4 files changed, 126 insertions(+), 3 deletions(-)
diff --git
a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ws/WebsocketEncoder.java
b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ws/WebSocketEncoder.java
similarity index 95%
rename from
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ws/WebsocketEncoder.java
rename to
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ws/WebSocketEncoder.java
index 602da73..d61b7f5 100644
---
a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ws/WebsocketEncoder.java
+++
b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ws/WebSocketEncoder.java
@@ -24,7 +24,7 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
-public class WebsocketEncoder extends MessageToMessageEncoder<ByteBuf> {
+public class WebSocketEncoder extends MessageToMessageEncoder<ByteBuf> {
@Override
protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object>
out) throws Exception {
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 adaa39b..9022940 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
@@ -36,7 +36,7 @@ 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.ws.WebSocketServerHandler;
-import org.apache.rocketmq.mqtt.cs.protocol.ws.WebsocketEncoder;
+import org.apache.rocketmq.mqtt.cs.protocol.ws.WebSocketEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@@ -117,7 +117,7 @@ public class MqttServer {
pipeline.addLast("aggregator", new
HttpObjectAggregator(connectConf.getMaxPacketSizeInByte() * 2));
pipeline.addLast("http-chunked", new
ChunkedWriteHandler());
pipeline.addLast("websocket-handler",
webSocketServerHandler);
- pipeline.addLast("websocket-encoder", new
WebsocketEncoder());
+ pipeline.addLast("websocket-encoder", new
WebSocketEncoder());
pipeline.addLast("decoder", new
MqttDecoder(connectConf.getMaxPacketSizeInByte()));
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast("dispatcher", mqttPacketDispatcher);
diff --git
a/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/protocol/ws/TestWebSocketEncoder.java
b/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/protocol/ws/TestWebSocketEncoder.java
new file mode 100644
index 0000000..6e005f6
--- /dev/null
+++
b/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/protocol/ws/TestWebSocketEncoder.java
@@ -0,0 +1,51 @@
+/*
+ * 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.test.protocol.ws;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import org.apache.rocketmq.mqtt.cs.protocol.ws.WebSocketEncoder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestWebSocketEncoder {
+
+ private WebSocketEncoder webSocketEncoder = new WebSocketEncoder();
+ private final int content = 666;
+
+ @Test
+ public void test() {
+ ByteBuf byteMsg = Unpooled.buffer();
+ byteMsg.writeByte(content);
+
+ EmbeddedChannel channel = new EmbeddedChannel(webSocketEncoder);
+ assertTrue(channel.writeOutbound(byteMsg));
+ assertTrue(channel.finish());
+
+ assertEquals(new BinaryWebSocketFrame(byteMsg),
channel.readOutbound());
+ assertNull(channel.readOutbound());
+ }
+}
diff --git
a/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/protocol/ws/TestWebSocketServerHandler.java
b/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/protocol/ws/TestWebSocketServerHandler.java
new file mode 100644
index 0000000..440aecc
--- /dev/null
+++
b/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/protocol/ws/TestWebSocketServerHandler.java
@@ -0,0 +1,72 @@
+/*
+ * 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.test.protocol.ws;
+
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http.DefaultFullHttpRequest;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
+import org.apache.rocketmq.mqtt.cs.protocol.ws.WebSocketServerHandler;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static io.netty.handler.codec.DecoderResult.UNFINISHED;
+import static io.netty.handler.codec.http.HttpMethod.CONNECT;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.junit.Assert.assertNull;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestWebSocketServerHandler {
+
+ private WebSocketServerHandler webSocketServerHandler = new
WebSocketServerHandler();
+ private EmbeddedChannel channel = new EmbeddedChannel();
+
+ @Before
+ public void setUp() throws Exception {
+ ChannelPipeline pipeline = channel.pipeline();
+ pipeline.addLast(webSocketServerHandler);
+ }
+
+ @Test
+ public void testHttpRequestUpgradeNull() {
+ FullHttpRequest httpRequest = new DefaultFullHttpRequest(HTTP_1_1,
CONNECT, "ws://localhost:8888/mqtt");
+ channel.writeInbound(httpRequest);
+
+ assertNull(channel.readInbound());
+ }
+
+ @Test
+ public void testSendHttpResponse() {
+ FullHttpRequest httpRequest = new DefaultFullHttpRequest(HTTP_1_1,
CONNECT, "ws://localhost:8888/mqtt");
+ httpRequest.setDecoderResult(UNFINISHED);
+ channel.writeInbound(httpRequest);
+
+ assertNull(channel.readInbound());
+ }
+
+ @Test
+ public void testPongWebSocketFrame() {
+ PongWebSocketFrame socketFrame = new PongWebSocketFrame();
+ channel.writeInbound(socketFrame);
+
+ assertNull(channel.readInbound());
+ }
+}
