feat[json]: 支持json协议格式的服务器

This commit is contained in:
godotg
2022-09-12 09:03:04 +08:00
parent 524e759d41
commit df210f968d
8 changed files with 416 additions and 0 deletions
@@ -0,0 +1,61 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.json;
import com.zfoo.net.core.AbstractClient;
import com.zfoo.net.handler.ClientRouteHandler;
import com.zfoo.net.handler.codec.json.JsonWebSocketCodecHandler;
import com.zfoo.protocol.util.IOUtils;
import com.zfoo.util.net.HostAndPort;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolConfig;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
/**
* @author godotg
* @version 3.0
*/
public class JsonWebsocketClient extends AbstractClient {
private WebSocketClientProtocolConfig webSocketClientProtocolConfig;
public JsonWebsocketClient(HostAndPort host, WebSocketClientProtocolConfig webSocketClientProtocolConfig) {
super(host);
this.webSocketClientProtocolConfig = webSocketClientProtocolConfig;
}
@Override
public ChannelInitializer<? extends Channel> channelChannelInitializer() {
return new ChannelHandlerInitializer();
}
public class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel channel) {
channel.pipeline().addLast(new HttpClientCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
channel.pipeline().addLast(new WebSocketClientProtocolHandler(webSocketClientProtocolConfig));
channel.pipeline().addLast(new ChunkedWriteHandler());
channel.pipeline().addLast(new JsonWebSocketCodecHandler());
channel.pipeline().addLast(new ClientRouteHandler());
}
}
}
@@ -0,0 +1,64 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.json;
import com.zfoo.net.core.AbstractServer;
import com.zfoo.net.handler.ServerRouteHandler;
import com.zfoo.net.handler.codec.json.JsonWebSocketCodecHandler;
import com.zfoo.net.handler.codec.websocket.WebSocketCodecHandler;
import com.zfoo.protocol.util.IOUtils;
import com.zfoo.util.net.HostAndPort;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
/**
* @author godotg
* @version 3.0
*/
public class JsonWebsocketServer extends AbstractServer {
public JsonWebsocketServer(HostAndPort host) {
super(host);
}
@Override
public ChannelInitializer<SocketChannel> channelChannelInitializer() {
return new ChannelHandlerInitializer();
}
public static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel channel) {
// 编解码 http 请求
channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
// 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
// 保证接收的 Http 请求的完整性
channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
// 处理其他的 WebSocketFrame
channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
// 写文件内容,支持异步发送大的码流,一般用于发送文件流
channel.pipeline().addLast(new ChunkedWriteHandler());
// 编解码WebSocketFrame二进制协议
channel.pipeline().addLast(new JsonWebSocketCodecHandler());
channel.pipeline().addLast(new ServerRouteHandler());
}
}
}
@@ -0,0 +1,50 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed 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 com.zfoo.net.handler.codec.json;
import com.zfoo.protocol.IPacket;
/**
* @author godotg
* @version 3.0
*/
public class JsonPacket {
private short protocolId;
private IPacket packet;
public static JsonPacket valueOf(short protocolId, IPacket packet) {
var jsonPacket = new JsonPacket();
jsonPacket.protocolId = protocolId;
jsonPacket.packet = packet;
return jsonPacket;
}
public short getProtocolId() {
return protocolId;
}
public void setProtocolId(short protocolId) {
this.protocolId = protocolId;
}
public IPacket getPacket() {
return packet;
}
public void setPacket(IPacket packet) {
this.packet = packet;
}
}
@@ -0,0 +1,61 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.handler.codec.json;
import com.zfoo.net.packet.model.DecodedPacketInfo;
import com.zfoo.net.packet.model.EncodedPacketInfo;
import com.zfoo.protocol.IPacket;
import com.zfoo.protocol.ProtocolManager;
import com.zfoo.protocol.buffer.ByteBufUtils;
import com.zfoo.protocol.util.JsonUtils;
import com.zfoo.protocol.util.StringUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import java.util.List;
/**
* @author godotg
* @version 3.0
*/
public class JsonWebSocketCodecHandler extends MessageToMessageCodec<WebSocketFrame, EncodedPacketInfo> {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, WebSocketFrame webSocketFrame, List<Object> list) {
var byteBuf = webSocketFrame.content();
var bytes = ByteBufUtils.readAllBytes(byteBuf);
var jsonStr = StringUtils.bytesToString(bytes);
var jsonMap = JsonUtils.getJsonMap(jsonStr);
var protocolId = Short.parseShort(jsonMap.get("protocolId"));
var packetStr = jsonMap.get("packet");
var protocolClass = ProtocolManager.getProtocol(protocolId).protocolConstructor().getDeclaringClass();
var packet = JsonUtils.string2Object(packetStr, protocolClass);
list.add(DecodedPacketInfo.valueOf((IPacket) packet, null));
}
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List<Object> list) {
var byteBuf = channelHandlerContext.alloc().ioBuffer();
var packet = out.getPacket();
var jsonPacket = JsonPacket.valueOf(packet.protocolId(), packet);
var bytes = StringUtils.bytes(JsonUtils.object2String(jsonPacket));
byteBuf.writeBytes(bytes);
list.add(new BinaryWebSocketFrame(byteBuf));
}
}
@@ -0,0 +1,57 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.json.client;
import com.zfoo.net.NetContext;
import com.zfoo.net.core.json.JsonWebsocketClient;
import com.zfoo.net.core.websocket.WebsocketClient;
import com.zfoo.net.packet.websocket.WebsocketHelloRequest;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolConfig;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author godotg
* @version 3.0
*/
@Ignore
public class JsonWebsocketClientTest {
@Test
public void startClient() {
var context = new ClassPathXmlApplicationContext("config.xml");
var webSocketClientProtocolConfig = WebSocketClientProtocolConfig.newBuilder()
.webSocketUri("http://127.0.0.1:9000/websocket")
.build();
var client = new JsonWebsocketClient(HostAndPort.valueOf("127.0.0.1:9000"), webSocketClientProtocolConfig);
var session = client.start();
var request = new WebsocketHelloRequest();
request.setMessage("Hello, this is the websocket client!");
for (int i = 0; i < 1000; i++) {
ThreadUtils.sleep(2000);
NetContext.getRouter().send(session, request);
}
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
@@ -0,0 +1,38 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed 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 com.zfoo.net.core.json.client;
import com.zfoo.net.packet.websocket.WebsocketHelloResponse;
import com.zfoo.net.router.receiver.PacketReceiver;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.util.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* @author godotg
* @version 3.0
*/
@Component
public class JsonWsClientController {
private static final Logger logger = LoggerFactory.getLogger(JsonWsClientController.class);
@PacketReceiver
public void atWebsocketHelloResponse(Session session, WebsocketHelloResponse response) {
logger.info("websocket client receive [packet:{}] from server", JsonUtils.object2String(response));
}
}
@@ -0,0 +1,40 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.json.server;
import com.zfoo.net.core.json.JsonWebsocketServer;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author godotg
* @version 3.0
*/
@Ignore
public class JsonWebsocketServerTest {
@Test
public void startServer() {
var context = new ClassPathXmlApplicationContext("config.xml");
var server = new JsonWebsocketServer(HostAndPort.valueOf("127.0.0.1:9000"));
server.start();
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
@@ -0,0 +1,45 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed 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 com.zfoo.net.core.json.server;
import com.zfoo.net.NetContext;
import com.zfoo.net.packet.websocket.WebsocketHelloRequest;
import com.zfoo.net.packet.websocket.WebsocketHelloResponse;
import com.zfoo.net.router.receiver.PacketReceiver;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.util.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* @author godotg
* @version 3.0
*/
@Component
public class JsonWsServerController {
private static final Logger logger = LoggerFactory.getLogger(JsonWsServerController.class);
@PacketReceiver
public void atWebsocketHelloRequest(Session session, WebsocketHelloRequest request) {
logger.info("receive [packet:{}] from browser", JsonUtils.object2String(request));
var response = new WebsocketHelloResponse();
response.setMessage("Hello, this is the websocket server!");
NetContext.getRouter().send(session, response);
}
}