diff --git a/net/src/main/java/com/zfoo/net/core/http/HttpServer.java b/net/src/main/java/com/zfoo/net/core/http/HttpServer.java index 2f0e4a7a..c7a5afea 100644 --- a/net/src/main/java/com/zfoo/net/core/http/HttpServer.java +++ b/net/src/main/java/com/zfoo/net/core/http/HttpServer.java @@ -14,9 +14,9 @@ package com.zfoo.net.core.http; import com.zfoo.net.core.AbstractServer; -import com.zfoo.net.handler.ServerDispatcherHandler; +import com.zfoo.net.handler.HttpDispatcherHandler; import com.zfoo.net.handler.codec.http.HttpCodecHandler; -import com.zfoo.protocol.IPacket; +import com.zfoo.net.packet.model.DecodedPacketInfo; import com.zfoo.util.net.HostAndPort; import io.netty.channel.ChannelInitializer; import io.netty.channel.socket.SocketChannel; @@ -36,9 +36,9 @@ public class HttpServer extends AbstractServer { /** * http的地址解析器 */ - private Function uriResolver; + private Function uriResolver; - public HttpServer(HostAndPort host, Function uriResolver) { + public HttpServer(HostAndPort host, Function uriResolver) { super(host); this.uriResolver = uriResolver; } @@ -56,7 +56,7 @@ public class HttpServer extends AbstractServer { channel.pipeline().addLast(new ChunkedWriteHandler()); channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024)); channel.pipeline().addLast(new HttpCodecHandler(uriResolver)); - channel.pipeline().addLast(new ServerDispatcherHandler()); + channel.pipeline().addLast(new HttpDispatcherHandler()); } } } diff --git a/net/src/main/java/com/zfoo/net/handler/HttpDispatcherHandler.java b/net/src/main/java/com/zfoo/net/handler/HttpDispatcherHandler.java new file mode 100644 index 00000000..e5cd3c23 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/handler/HttpDispatcherHandler.java @@ -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.handler; + +import com.zfoo.net.NetContext; +import com.zfoo.net.packet.model.DecodedPacketInfo; +import com.zfoo.net.packet.model.HttpPacketAttachment; +import com.zfoo.net.util.SessionUtils; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; + +/** + * @author jaysunxiao + * @version 3.0 + */ +@ChannelHandler.Sharable +public class HttpDispatcherHandler extends ServerDispatcherHandler { + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + var session = SessionUtils.getSession(ctx); + if (session == null) { + return; + } + DecodedPacketInfo decodedPacketInfo = (DecodedPacketInfo) msg; + + var httpPacketAttachment = (HttpPacketAttachment) decodedPacketInfo.getPacketAttachment(); + if (httpPacketAttachment.getUid() <= 0) { + httpPacketAttachment.useExecutorConsistentHash(session.getSid()); + } + + NetContext.getDispatcher().receive(session, decodedPacketInfo.getPacket(), decodedPacketInfo.getPacketAttachment()); + } +} diff --git a/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java index 57cce4aa..6f400fc6 100644 --- a/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java @@ -23,13 +23,15 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToMessageCodec; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.FullHttpRequest; -import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpHeaderValues; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; import java.util.function.Function; + /** * @author jaysunxiao * @version 3.0 @@ -38,9 +40,9 @@ public class HttpCodecHandler extends MessageToMessageCodec uriResolver; + private Function uriResolver; - public HttpCodecHandler(Function uriResolver) { + public HttpCodecHandler(Function uriResolver) { super(); this.uriResolver = uriResolver; } @@ -48,9 +50,7 @@ public class HttpCodecHandler extends MessageToMessageCodec list) { try { - var packet = uriResolver.apply(fullHttpRequest); - var attachment = HttpPacketAttachment.valueOf(fullHttpRequest, HttpResponseStatus.OK); - var decodedPacketInfo = DecodedPacketInfo.valueOf(packet, attachment); + var decodedPacketInfo = uriResolver.apply(fullHttpRequest); list.add(decodedPacketInfo); } catch (Exception e) { logger.error("exception异常", e); @@ -63,7 +63,6 @@ public class HttpCodecHandler extends MessageToMessageCodec list) { - try { var packet = (IPacket) out.getPacket(); var attachment = (HttpPacketAttachment) out.getPacketAttachment(); @@ -71,13 +70,31 @@ public class HttpCodecHandler extends MessageToMessageCodec implements LpMap { checkKey(key); return map.get(key); } + + @Override + public long getMaxIndex() { + return maxIndexAtomic.get(); + } } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java index f00f8fa9..e542d860 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java @@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap; import com.zfoo.protocol.IPacket; +import java.util.function.BiConsumer; + /** * @author jaysunxiao * @version 3.0 @@ -55,6 +57,11 @@ public class FileChannelHeapMap implements LpMap { return heapMap.get(key); } + @Override + public long getMaxIndex() { + return heapMap.getMaxIndex(); + } + private void load() { var maxIndex = fileChannelMap.getMaxIndex(); if (maxIndex <= 0) { @@ -68,7 +75,10 @@ public class FileChannelHeapMap implements LpMap { } heapMap.put(key, value); } - } + @Override + public void forEach(BiConsumer biConsumer) { + heapMap.forEach(biConsumer); + } } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java index 6e3f7723..8349a22d 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java @@ -145,6 +145,19 @@ public class FileChannelMap implements LpMap, Closeable { } } + @Override + public long getMaxIndex() { + try { + clearByteBuf(); + indexBuffer.writeBytes(indexFileChannel, 0, 8); + return indexBuffer.readLong(); + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + clearByteBuf(); + } + } + protected void setKeyValue(long key, V value) { try { clearByteBuf(); @@ -194,18 +207,6 @@ public class FileChannelMap implements LpMap, Closeable { } } - protected long getMaxIndex() { - try { - clearByteBuf(); - indexBuffer.writeBytes(indexFileChannel, 0, 8); - return indexBuffer.readLong(); - } catch (Exception e) { - throw new RuntimeException(e); - } finally { - clearByteBuf(); - } - } - protected void clearByteBuf() { indexBuffer.clear(); dbBuffer.clear(); diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java index 6feb52a9..2fdf9764 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java @@ -77,6 +77,11 @@ public class FileHeapMap implements LpMap { return heapMap.get(key); } + @Override + public long getMaxIndex() { + return heapMap.getMaxIndex(); + } + private void load() { FileInputStream fileInputStream = null; FileChannel fileChannel = null; diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java index ebf2a37b..918f692d 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java @@ -17,6 +17,7 @@ import io.netty.util.collection.LongObjectHashMap; import java.util.LinkedList; import java.util.Queue; +import java.util.function.BiConsumer; /** * @author jaysunxiao @@ -81,4 +82,14 @@ public class HeapMap implements LpMap { checkKey(key); return map.get(key); } + + @Override + public long getMaxIndex() { + return maxIndex; + } + + @Override + public void forEach(BiConsumer biConsumer) { + map.forEach(biConsumer); + } } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java index 610adc8b..a3db838b 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java @@ -15,6 +15,8 @@ package com.zfoo.orm.lpmap; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.exception.RunException; +import java.util.function.BiConsumer; + /** * 类型固定的map,key为long,value为IPacket * 其中long必须大于0,value可以为null @@ -41,10 +43,16 @@ public interface LpMap { V get(long key); + long getMaxIndex(); + default void checkKey(long key) { if (key <= 0) { throw new RunException("key[{}]不能为负数", key); } } + default void forEach(BiConsumer biConsumer) { + + } + }