From 5df83b26498a2b57b33776f1bb590ad5e5f06683 Mon Sep 17 00:00:00 2001 From: redkale Date: Mon, 6 Feb 2023 19:08:20 +0800 Subject: [PATCH] =?UTF-8?q?=E7=A7=BB=E9=99=A4ChannelContext?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../org/redkale/convert/ArrayEncoder.java | 2 +- .../org/redkale/convert/ObjectEncoder.java | 2 +- src/main/java/org/redkale/convert/Writer.java | 50 +++++++++----- .../convert/bson/BsonByteBufferWriter.java | 2 +- .../org/redkale/convert/bson/BsonWriter.java | 2 +- .../convert/json/JsonByteBufferReader.java | 4 +- .../convert/json/JsonByteBufferWriter.java | 3 +- .../redkale/convert/json/JsonBytesWriter.java | 21 +++--- .../redkale/convert/json/JsonCharsWriter.java | 2 +- .../org/redkale/convert/json/JsonConvert.java | 16 ++--- .../redkale/mq/HttpMessageClusterClient.java | 8 +-- .../java/org/redkale/net/AsyncConnection.java | 7 +- .../java/org/redkale/net/ChannelContext.java | 31 --------- src/main/java/org/redkale/net/Request.java | 4 -- .../java/org/redkale/net/client/Client.java | 65 ++++--------------- .../org/redkale/net/client/ClientCodec.java | 25 +++++-- .../org/redkale/net/http/HttpContext.java | 16 +---- .../net/http/HttpDispatcherServlet.java | 19 ++++-- .../org/redkale/net/http/HttpRequest.java | 6 +- .../java/org/redkale/net/http/HttpScope.java | 6 -- src/main/java/org/redkale/net/http/Rest.java | 23 +++---- .../java/org/redkale/net/http/RestLocale.java | 4 +- .../org/redkale/net/http/RestWebSocket.java | 9 +-- .../org/redkale/net/http/WebSocketEngine.java | 8 +-- .../redkale/net/http/WebSocketServlet.java | 8 +-- 25 files changed, 130 insertions(+), 213 deletions(-) delete mode 100644 src/main/java/org/redkale/net/ChannelContext.java diff --git a/src/main/java/org/redkale/convert/ArrayEncoder.java b/src/main/java/org/redkale/convert/ArrayEncoder.java index a8fdca3f9..5e87eaf81 100644 --- a/src/main/java/org/redkale/convert/ArrayEncoder.java +++ b/src/main/java/org/redkale/convert/ArrayEncoder.java @@ -111,7 +111,7 @@ public class ArrayEncoder implements Encodeable { final Type comp = this.componentType; for (int i = 0;; i++) { Object v = value[i]; - writeMemberValue(out, member, ((v != null && (v.getClass() == comp || out.specify() == comp)) ? itemEncoder : anyEncoder), v, i); + writeMemberValue(out, member, ((v != null && (v.getClass() == comp || out.specificObjectType() == comp)) ? itemEncoder : anyEncoder), v, i); if (i == iMax) { break; } diff --git a/src/main/java/org/redkale/convert/ObjectEncoder.java b/src/main/java/org/redkale/convert/ObjectEncoder.java index 2ea97c22e..ecefa0076 100644 --- a/src/main/java/org/redkale/convert/ObjectEncoder.java +++ b/src/main/java/org/redkale/convert/ObjectEncoder.java @@ -279,7 +279,7 @@ public class ObjectEncoder implements Encodeable { lock.unlock(); } } - if (value.getClass() != this.typeClass && !this.type.equals(out.specify())) { + if (value.getClass() != this.typeClass && !this.type.equals(out.specificObjectType())) { final Class clz = value.getClass(); if (out.needWriteClassName()) { out.writeClassName(factory.getEntityAlias(clz)); diff --git a/src/main/java/org/redkale/convert/Writer.java b/src/main/java/org/redkale/convert/Writer.java index 42b1d91f1..2378c1408 100644 --- a/src/main/java/org/redkale/convert/Writer.java +++ b/src/main/java/org/redkale/convert/Writer.java @@ -23,7 +23,7 @@ public abstract class Writer { protected boolean comma; //convertTo时是否以指定Type的ObjectEncoder进行处理 - protected Type specify; + protected Type specificObjectType; //对某个key值进行动态处理，仅供MapEncoder使用 protected BiFunction mapFieldFunc; @@ -35,32 +35,36 @@ public abstract class Writer { protected Function objExtFunc; /** - * 设置specify + * 设置specificObjectType * * @param value Type */ - public void specify(Type value) { + public void specificObjectType(Type value) { if (value instanceof GenericArrayType) { - this.specify = ((GenericArrayType) value).getGenericComponentType(); + this.specificObjectType = ((GenericArrayType) value).getGenericComponentType(); } else if (value instanceof Class && ((Class) value).isArray()) { - this.specify = ((Class) value).getComponentType(); + this.specificObjectType = ((Class) value).getComponentType(); } else { - this.specify = value; + this.specificObjectType = value; } } protected boolean recycle() { + this.comma = false; + this.specificObjectType = null; + this.mapFieldFunc = null; this.objFieldFunc = null; + this.objExtFunc = null; return true; } /** - * 返回specify + * 返回specificObjectType * * @return int */ - public Type specify() { - return this.specify; + public Type specificObjectType() { + return this.specificObjectType; } /** @@ -127,12 +131,18 @@ public abstract class Writer { } else { value = objFieldFunc.apply(member.attribute, obj); } - if (value == null) return; + if (value == null) { + return; + } if (tiny()) { if (member.string) { - if (((CharSequence) value).length() == 0) return; + if (((CharSequence) value).length() == 0) { + return; + } } else if (member.bool) { - if (!((Boolean) value)) return; + if (!((Boolean) value)) { + return; + } } } Attribute attr = member.getAttribute(); @@ -153,14 +163,22 @@ public abstract class Writer { */ @SuppressWarnings("unchecked") public void writeObjectField(final String fieldName, Type fieldType, int fieldPos, Encodeable anyEncoder, Object value) { - if (value == null) return; - if (fieldType == null) fieldType = value.getClass(); + if (value == null) { + return; + } + if (fieldType == null) { + fieldType = value.getClass(); + } if (tiny() && fieldType instanceof Class) { Class clazz = (Class) fieldType; if (CharSequence.class.isAssignableFrom(clazz)) { - if (((CharSequence) value).length() == 0) return; + if (((CharSequence) value).length() == 0) { + return; + } } else if (clazz == boolean.class || clazz == Boolean.class) { - if (!((Boolean) value)) return; + if (!((Boolean) value)) { + return; + } } } this.writeFieldName(null, fieldName, fieldType, fieldPos); diff --git a/src/main/java/org/redkale/convert/bson/BsonByteBufferWriter.java b/src/main/java/org/redkale/convert/bson/BsonByteBufferWriter.java index 49a523a38..eeb9e718b 100644 --- a/src/main/java/org/redkale/convert/bson/BsonByteBufferWriter.java +++ b/src/main/java/org/redkale/convert/bson/BsonByteBufferWriter.java @@ -139,7 +139,7 @@ public class BsonByteBufferWriter extends BsonWriter { protected boolean recycle() { super.recycle(); this.index = 0; - this.specify = null; + this.specificObjectType = null; this.buffers = null; return false; } diff --git a/src/main/java/org/redkale/convert/bson/BsonWriter.java b/src/main/java/org/redkale/convert/bson/BsonWriter.java index df901caad..b834b5375 100644 --- a/src/main/java/org/redkale/convert/bson/BsonWriter.java +++ b/src/main/java/org/redkale/convert/bson/BsonWriter.java @@ -153,7 +153,7 @@ public class BsonWriter extends Writer implements ByteTuple { protected boolean recycle() { super.recycle(); this.count = 0; - this.specify = null; + this.specificObjectType = null; if (this.content != null && this.content.length > defaultSize) { this.content = new byte[defaultSize]; } diff --git a/src/main/java/org/redkale/convert/json/JsonByteBufferReader.java b/src/main/java/org/redkale/convert/json/JsonByteBufferReader.java index 8a919f17b..683edc5e9 100644 --- a/src/main/java/org/redkale/convert/json/JsonByteBufferReader.java +++ b/src/main/java/org/redkale/convert/json/JsonByteBufferReader.java @@ -42,10 +42,10 @@ public class JsonByteBufferReader extends JsonReader { @Override protected boolean recycle() { super.recycle(); // this.position 初始化值为-1 - this.currentIndex = 0; this.currentChar = 0; - this.currentBuffer = null; this.buffers = null; + this.currentIndex = 0; + this.currentBuffer = null; this.mask = null; return false; } diff --git a/src/main/java/org/redkale/convert/json/JsonByteBufferWriter.java b/src/main/java/org/redkale/convert/json/JsonByteBufferWriter.java index 21da23b67..80ce0fd22 100644 --- a/src/main/java/org/redkale/convert/json/JsonByteBufferWriter.java +++ b/src/main/java/org/redkale/convert/json/JsonByteBufferWriter.java @@ -53,10 +53,9 @@ public class JsonByteBufferWriter extends JsonWriter { @Override protected boolean recycle() { super.recycle(); - this.index = 0; - this.specify = null; this.charset = null; this.buffers = null; + this.index = 0; return false; } diff --git a/src/main/java/org/redkale/convert/json/JsonBytesWriter.java b/src/main/java/org/redkale/convert/json/JsonBytesWriter.java index 87e76236c..340228edc 100644 --- a/src/main/java/org/redkale/convert/json/JsonBytesWriter.java +++ b/src/main/java/org/redkale/convert/json/JsonBytesWriter.java @@ -72,6 +72,16 @@ public class JsonBytesWriter extends JsonWriter implements ByteTuple { return newdata; } + @Override + public boolean recycle() { + super.recycle(); + this.count = 0; + if (this.content != null && this.content.length > defaultSize * 100) { + this.content = new byte[defaultSize]; + } + return true; + } + @Override public byte[] content() { return content; @@ -333,17 +343,6 @@ public class JsonBytesWriter extends JsonWriter implements ByteTuple { return this; } - @Override - public boolean recycle() { - super.recycle(); - this.count = 0; - this.specify = null; - if (this.content != null && this.content.length > defaultSize * 100) { - this.content = new byte[defaultSize]; - } - return true; - } - /** * 直接获取全部数据, 实际数据需要根据count长度来截取 * diff --git a/src/main/java/org/redkale/convert/json/JsonCharsWriter.java b/src/main/java/org/redkale/convert/json/JsonCharsWriter.java index d04d1217b..d6ea0f3b6 100644 --- a/src/main/java/org/redkale/convert/json/JsonCharsWriter.java +++ b/src/main/java/org/redkale/convert/json/JsonCharsWriter.java @@ -293,7 +293,7 @@ public class JsonCharsWriter extends JsonWriter { protected boolean recycle() { super.recycle(); this.count = 0; - this.specify = null; + this.specificObjectType = null; if (this.content != null && this.content.length > defaultSize) { this.content = new char[defaultSize]; } diff --git a/src/main/java/org/redkale/convert/json/JsonConvert.java b/src/main/java/org/redkale/convert/json/JsonConvert.java index 168e2b98f..c396defc2 100644 --- a/src/main/java/org/redkale/convert/json/JsonConvert.java +++ b/src/main/java/org/redkale/convert/json/JsonConvert.java @@ -304,7 +304,7 @@ public class JsonConvert extends TextConvert { this.lastConvertEncodeable = encoder; } if (encoder.specifyable()) { - writer.specify(type); + writer.specificObjectType(type); } encoder.convertTo(writer, value); @@ -336,7 +336,7 @@ public class JsonConvert extends TextConvert { this.lastConvertEncodeable = encoder; } if (encoder.specifyable()) { - writer.specify(type); + writer.specificObjectType(type); } encoder.convertTo(writer, value); @@ -362,7 +362,7 @@ public class JsonConvert extends TextConvert { this.lastConvertEncodeable = encoder; } if (encoder.specifyable()) { - writer.specify(type); + writer.specificObjectType(type); } encoder.convertTo(writer, value); } @@ -386,7 +386,7 @@ public class JsonConvert extends TextConvert { this.lastConvertEncodeable = encoder; } if (encoder.specifyable()) { - writer.specify(type); + writer.specificObjectType(type); } encoder.convertTo(writer, value); } @@ -415,7 +415,7 @@ public class JsonConvert extends TextConvert { this.lastConvertEncodeable = encoder; } if (encoder.specifyable()) { - writer.specify(type); + writer.specificObjectType(type); } encoder.convertTo(writer, value); } @@ -444,7 +444,7 @@ public class JsonConvert extends TextConvert { if (value == null) { out.writeNull(); } else { - out.specify(type); + out.specificObjectType(type); factory.loadEncoder(type).convertTo(out, value); } return out.toBuffers(); @@ -462,7 +462,7 @@ public class JsonConvert extends TextConvert { this.lastConvertEncodeable = encoder; } if (encoder.specifyable()) { - writer.specify(type); + writer.specificObjectType(type); } encoder.convertTo(writer, value); } @@ -482,7 +482,7 @@ public class JsonConvert extends TextConvert { this.lastConvertEncodeable = encoder; } if (encoder.specifyable()) { - writer.specify(type); + writer.specificObjectType(type); } encoder.convertTo(writer, value); } diff --git a/src/main/java/org/redkale/mq/HttpMessageClusterClient.java b/src/main/java/org/redkale/mq/HttpMessageClusterClient.java index 1a54012b6..ddcf8ba8e 100644 --- a/src/main/java/org/redkale/mq/HttpMessageClusterClient.java +++ b/src/main/java/org/redkale/mq/HttpMessageClusterClient.java @@ -106,7 +106,7 @@ public class HttpMessageClusterClient extends HttpMessageClient { clientHeaders.put(Rest.REST_HEADER_RESP_CONVERT_TYPE, req.getRespConvertType().toString()); } if (userid != null) { - clientHeaders.put(Rest.REST_HEADER_CURRUSERID_NAME, "" + userid); + clientHeaders.put(Rest.REST_HEADER_CURRUSERID, "" + userid); } if (headers != null) { headers.forEach((n, v) -> { @@ -186,7 +186,7 @@ public class HttpMessageClusterClient extends HttpMessageClient { clientHeaders.put(Rest.REST_HEADER_RESP_CONVERT_TYPE, req.getRespConvertType().toString()); } if (userid != null) { - clientHeaders.put(Rest.REST_HEADER_CURRUSERID_NAME, "" + userid); + clientHeaders.put(Rest.REST_HEADER_CURRUSERID, "" + userid); } if (headers != null) { boolean ws = headers.containsKey("Sec-WebSocket-Key"); @@ -265,7 +265,7 @@ public class HttpMessageClusterClient extends HttpMessageClient { // if (req.isFrombody()) builder.header(Rest.REST_HEADER_PARAM_FROM_BODY, "true"); // if (req.getReqConvertType() != null) builder.header(Rest.REST_HEADER_REQ_CONVERT_TYPE, req.getReqConvertType().toString()); // if (req.getRespConvertType() != null) builder.header(Rest.REST_HEADER_RESP_CONVERT_TYPE, req.getRespConvertType().toString()); -// if (userid != 0) builder.header(Rest.REST_HEADER_CURRUSERID_NAME, "" + userid); +// if (userid != 0) builder.header(Rest.REST_HEADER_CURRUSERID, "" + userid); // if (headers != null) headers.forEach((n, v) -> { // if (!DISALLOWED_HEADERS_SET.contains(n.toLowerCase())) builder.header(n, v); // }); @@ -313,7 +313,7 @@ public class HttpMessageClusterClient extends HttpMessageClient { // if (req.isFrombody()) builder.header(Rest.REST_HEADER_PARAM_FROM_BODY, "true"); // if (req.getReqConvertType() != null) builder.header(Rest.REST_HEADER_REQ_CONVERT_TYPE, req.getReqConvertType().toString()); // if (req.getRespConvertType() != null) builder.header(Rest.REST_HEADER_RESP_CONVERT_TYPE, req.getRespConvertType().toString()); -// if (userid != 0) builder.header(Rest.REST_HEADER_CURRUSERID_NAME, "" + userid); +// if (userid != 0) builder.header(Rest.REST_HEADER_CURRUSERID, "" + userid); // if (headers != null) headers.forEach((n, v) -> { // if (!DISALLOWED_HEADERS_SET.contains(n.toLowerCase())) builder.header(n, v); // }); diff --git a/src/main/java/org/redkale/net/AsyncConnection.java b/src/main/java/org/redkale/net/AsyncConnection.java index 9b5a42c78..bccd24de6 100644 --- a/src/main/java/org/redkale/net/AsyncConnection.java +++ b/src/main/java/org/redkale/net/AsyncConnection.java @@ -26,7 +26,7 @@ import org.redkale.util.*; * * @author zhangjx */ -public abstract class AsyncConnection implements ChannelContext, Channel, AutoCloseable { +public abstract class AsyncConnection implements Channel, AutoCloseable { private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0); @@ -814,7 +814,6 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl this.subobject = value; } - @Override public void setAttribute(String name, Object value) { if (this.attributes == null) { this.attributes = new HashMap<>(); @@ -822,25 +821,21 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl this.attributes.put(name, value); } - @Override @SuppressWarnings("unchecked") public final T getAttribute(String name) { return (T) (this.attributes == null ? null : this.attributes.get(name)); } - @Override public final void removeAttribute(String name) { if (this.attributes != null) { this.attributes.remove(name); } } - @Override public final Map getAttributes() { return this.attributes; } - @Override public final void clearAttribute() { if (this.attributes != null) { this.attributes.clear(); diff --git a/src/main/java/org/redkale/net/ChannelContext.java b/src/main/java/org/redkale/net/ChannelContext.java deleted file mode 100644 index acd990347..000000000 --- a/src/main/java/org/redkale/net/ChannelContext.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * To change this license header, choose License Headers in Project Properties. - * To change this template file, choose Tools | Templates - * and open the template in the editor. - */ -package org.redkale.net; - -import java.util.*; - -/** - * 当前一个Request绑定的AsyncConnection， 类似Session，但概念上不同于sessionid对应的对象 - * - *

- * 详情见: https://redkale.org - * - * @author zhangjx - * @since 2.4.0 - */ -public interface ChannelContext { - - public void setAttribute(String name, Object value); - - @SuppressWarnings("unchecked") - public T getAttribute(String name); - - public void removeAttribute(String name); - - public Map getAttributes(); - - public void clearAttribute(); -} diff --git a/src/main/java/org/redkale/net/Request.java b/src/main/java/org/redkale/net/Request.java index b572d2ee6..edbf80498 100644 --- a/src/main/java/org/redkale/net/Request.java +++ b/src/main/java/org/redkale/net/Request.java @@ -151,10 +151,6 @@ public abstract class Request { return attributes; } - public ChannelContext getChannelContext() { - return channel; - } - public C getContext() { return this.context; } diff --git a/src/main/java/org/redkale/net/client/Client.java b/src/main/java/org/redkale/net/client/Client.java index f8642a907..53d51658e 100644 --- a/src/main/java/org/redkale/net/client/Client.java +++ b/src/main/java/org/redkale/net/client/Client.java @@ -206,28 +206,14 @@ public abstract class Client, R extends ClientR if (request.workThread == null) { request.workThread = WorkThread.currWorkThread(); } - return connect(null).thenCompose(conn -> writeChannel(conn, request)); + return connect().thenCompose(conn -> writeChannel(conn, request)); } public final CompletableFuture sendAsync(R request, Function respTransfer) { if (request.workThread == null) { request.workThread = WorkThread.currWorkThread(); } - return connect(null).thenCompose(conn -> writeChannel(conn, request, respTransfer)); - } - - public final CompletableFuture

sendAsync(ChannelContext context, R request) { - if (request.workThread == null) { - request.workThread = WorkThread.currWorkThread(); - } - return connect(context).thenCompose(conn -> writeChannel(conn, request)); - } - - public final CompletableFuture sendAsync(ChannelContext context, R request, Function respTransfer) { - if (request.workThread == null) { - request.workThread = WorkThread.currWorkThread(); - } - return connect(context).thenCompose(conn -> writeChannel(conn, request, respTransfer)); + return connect().thenCompose(conn -> writeChannel(conn, request, respTransfer)); } protected CompletableFuture

writeChannel(ClientConnection conn, R request) { @@ -239,38 +225,16 @@ public abstract class Client, R extends ClientR } protected CompletableFuture connect() { - return connect(null); - } - - protected CompletableFuture connect(final ChannelContext context) { - final boolean cflag = context != null && connectionContextName != null; - if (cflag) { - C cc = context.getAttribute(connectionContextName); - if (cc != null && cc.isOpen()) { - return CompletableFuture.completedFuture(cc); - } - } final int size = this.connArray.length; - int connIndex = (int) Math.abs(connIndexSeq.getAndIncrement()) % size; -// WorkThread workThread = WorkThread.currWorkThread(); -// if (workThread != null && workThread.threads() == size) { -// connIndex = workThread.index(); -// } else { -// connIndex = (int) Math.abs(Thread.currentThread().getId() % size); -// } -// if (connIndex >= 0) { + final int connIndex = (int) Math.abs(connIndexSeq.getAndIncrement()) % size; C cc = (C) this.connArray[connIndex]; if (cc != null && cc.isOpen()) { - if (cflag) { - context.setAttribute(connectionContextName, cc); - } return CompletableFuture.completedFuture(cc); } - final int index = connIndex; - final Queue> waitQueue = this.connAcquireWaitings[index]; - if (this.connOpenStates[index].compareAndSet(false, true)) { + final Queue> waitQueue = this.connAcquireWaitings[connIndex]; + if (this.connOpenStates[connIndex].compareAndSet(false, true)) { CompletableFuture future = address.createClient(tcp, group, readTimeoutSeconds, writeTimeoutSeconds) - .thenApply(c -> (C) createClientConnection(index, c).setMaxPipelines(maxPipelines)); + .thenApply(c -> (C) createClientConnection(connIndex, c).setMaxPipelines(maxPipelines)); R virtualReq = createVirtualRequestAfterConnect(); if (virtualReq != null) { future = future.thenCompose(conn -> conn.writeVirtualRequest(virtualReq).thenApply(v -> conn)); @@ -282,18 +246,15 @@ public abstract class Client, R extends ClientR } return future.thenApply(c -> { c.setAuthenticated(true); - this.connArray[index] = c; + this.connArray[connIndex] = c; CompletableFuture f; - if (cflag) { - context.setAttribute(connectionContextName, c); - } while ((f = waitQueue.poll()) != null) { f.complete(c); } return c; }).whenComplete((r, t) -> { if (t != null) { - this.connOpenStates[index].set(false); + this.connOpenStates[connIndex].set(false); } }); } else { @@ -327,13 +288,9 @@ public abstract class Client, R extends ClientR // if (minRunningConn != null) { // && minRunningConn.runningCount() < maxPipelines // return CompletableFuture.completedFuture(minRunningConn); // } -// return waitClientConnection(); - } - - protected CompletableFuture waitClientConnection() { - CompletableFuture rs = Utility.orTimeout(new CompletableFuture(), 6, TimeUnit.SECONDS); - connAcquireWaitings[connSeqno.getAndIncrement() % this.connLimit].offer(rs); - return rs; +// CompletableFuture rs = Utility.orTimeout(new CompletableFuture(), 6, TimeUnit.SECONDS); +// connAcquireWaitings[connSeqno.getAndIncrement() % this.connLimit].offer(rs); +// return rs; } protected long getRespWaitingCount() { diff --git a/src/main/java/org/redkale/net/client/ClientCodec.java b/src/main/java/org/redkale/net/client/ClientCodec.java index 54a559a7f..b195727f7 100644 --- a/src/main/java/org/redkale/net/client/ClientCodec.java +++ b/src/main/java/org/redkale/net/client/ClientCodec.java @@ -121,28 +121,43 @@ public abstract class ClientCodec implements Complet respFuture.cancelTimeout(); //if (client.finest) client.logger.log(Level.FINEST, Utility.nowMillis() + ": " + Thread.currentThread().getName() + ": " + ClientConnection.this + ", 回调处理, req=" + request + ", message=" + rs.message); connection.preComplete(message, (R) request, exc); + boolean reqInIO = workThread != null && workThread.inIO(); if (workThread == null || workThread.getWorkExecutor() == null) { workThread = connection.channel.getReadIOThread(); } if (exc != null) { - workThread.runWork(() -> { + if (reqInIO) { //request在IO线程中发送请求，说明request是在异步模式中 if (request != null) { Traces.currTraceid(request.traceid); } respFuture.completeExceptionally(exc); - }); + } else { + workThread.runWork(() -> { + if (request != null) { + Traces.currTraceid(request.traceid); + } + respFuture.completeExceptionally(exc); + }); + } } else { final Object rs = request == null || request.respTransfer == null ? message : request.respTransfer.apply(message); - workThread.runWork(() -> { + if (reqInIO) { //request在IO线程中发送请求，说明request是在异步模式中 if (request != null) { Traces.currTraceid(request.traceid); } ((ClientFuture) respFuture).complete(rs); - }); + } else { + workThread.runWork(() -> { + if (request != null) { + Traces.currTraceid(request.traceid); + } + ((ClientFuture) respFuture).complete(rs); + }); + } } } catch (Throwable t) { - if (workThread == null) { + if (workThread == null || workThread.inIO()) { if (request != null) { Traces.currTraceid(request.traceid); } diff --git a/src/main/java/org/redkale/net/http/HttpContext.java b/src/main/java/org/redkale/net/http/HttpContext.java index 23db82030..f173273f1 100644 --- a/src/main/java/org/redkale/net/http/HttpContext.java +++ b/src/main/java/org/redkale/net/http/HttpContext.java @@ -41,12 +41,11 @@ public class HttpContext extends Context { protected final AnyValue rpcAuthenticatorConfig; - //所有Servlet方法都不需要读取http-header，lazyHeaders=true + //所有Servlet方法都不需要读取http-header且不存在HttpFilter的情况下，lazyHeaders=true protected boolean lazyHeaders; //存在动态改值 Function webSocketWriterIOThreadFunc; -// protected RequestURINode[] uriCacheNodes; public HttpContext(HttpContextConfig config) { super(config); this.remoteAddrHeader = config.remoteAddrHeader; @@ -57,19 +56,6 @@ public class HttpContext extends Context { random.setSeed(Math.abs(System.nanoTime())); } -// protected RequestURINode[] getUriCacheNodes() { -// return uriCacheNodes; -// } -// -// protected void addRequestURINode(String path) { -// RequestURINode node = new RequestURINode(path); -// if (this.uriCacheNodes != null) { -// for (int i = 0; i < uriCacheNodes.length; i++) { -// if (uriCacheNodes[i].path.equals(path)) return; -// } -// } -// this.uriCacheNodes = Utility.append(this.uriCacheNodes, node); -// } @Override protected void updateReadIOThread(AsyncConnection conn, AsyncIOThread ioReadThread) { super.updateReadIOThread(conn, ioReadThread); diff --git a/src/main/java/org/redkale/net/http/HttpDispatcherServlet.java b/src/main/java/org/redkale/net/http/HttpDispatcherServlet.java index 24405128a..3ac6caf6e 100644 --- a/src/main/java/org/redkale/net/http/HttpDispatcherServlet.java +++ b/src/main/java/org/redkale/net/http/HttpDispatcherServlet.java @@ -12,7 +12,8 @@ import java.util.function.*; import java.util.logging.*; import java.util.regex.Pattern; import java.util.stream.Stream; -import org.redkale.net.DispatcherServlet; +import org.redkale.net.*; +import org.redkale.net.Filter; import org.redkale.net.http.Rest.RestDynSourceType; import org.redkale.service.Service; import org.redkale.util.AnyValue.DefaultAnyValue; @@ -47,6 +48,7 @@ public class HttpDispatcherServlet extends DispatcherServlet> forbidURIMaps; //禁用的URL的正则表达式, 必须与 forbidURIPredicates 保持一致 @@ -363,6 +365,15 @@ public class HttpDispatcherServlet extends DispatcherServlet filter, AnyValue conf) { + super.addFilter(filter, conf); + this.lazyHeaders = false; + if (context != null) { + context.lazyHeaders = this.lazyHeaders; //启动后运行过程中执行addFilter + } + } + /** * 添加HttpServlet * @@ -385,10 +396,10 @@ public class HttpDispatcherServlet extends DispatcherServlet { req.setBody(array.length() == 0 ? null : array.getBytes()); if (!getHeaders().isEmpty()) { if (headers.containsKey(Rest.REST_HEADER_RPC) - || headers.containsKey(Rest.REST_HEADER_CURRUSERID_NAME)) { //外部request不能包含RPC的header信息 + || headers.containsKey(Rest.REST_HEADER_CURRUSERID)) { //外部request不能包含RPC的header信息 req.setHeaders(new HashMap<>(headers)); req.removeHeader(Rest.REST_HEADER_RPC); - req.removeHeader(Rest.REST_HEADER_CURRUSERID_NAME); + req.removeHeader(Rest.REST_HEADER_CURRUSERID); } else { req.setHeaders(headers); } @@ -745,7 +745,7 @@ public class HttpRequest extends Request { this.rpc = "true".equalsIgnoreCase(value); headers.put(name, value); break; - case Rest.REST_HEADER_CURRUSERID_NAME: + case Rest.REST_HEADER_CURRUSERID: value = bytes.toString(charset); this.hashid = value.hashCode(); this.currentUserid = value; diff --git a/src/main/java/org/redkale/net/http/HttpScope.java b/src/main/java/org/redkale/net/http/HttpScope.java index e1d3a23c1..8a2631704 100644 --- a/src/main/java/org/redkale/net/http/HttpScope.java +++ b/src/main/java/org/redkale/net/http/HttpScope.java @@ -6,15 +6,12 @@ package org.redkale.net.http; import java.io.Serializable; -import java.lang.reflect.Type; import java.net.HttpCookie; import java.util.*; -import java.util.concurrent.CompletableFuture; import java.util.function.*; import org.redkale.convert.*; import org.redkale.convert.json.JsonConvert; import org.redkale.persistence.Transient; -import org.redkale.util.TypeToken; /** * HTTP输出引擎的对象域
@@ -44,9 +41,6 @@ public class HttpScope { public static final Object NIL = new Object(); - static final Type FUTRU_TYPE = new TypeToken>() { - }.getType(); - @ConvertColumn(index = 1) protected String referid; diff --git a/src/main/java/org/redkale/net/http/Rest.java b/src/main/java/org/redkale/net/http/Rest.java index bd2008b9f..f5c3bc50f 100644 --- a/src/main/java/org/redkale/net/http/Rest.java +++ b/src/main/java/org/redkale/net/http/Rest.java @@ -38,16 +38,22 @@ import org.redkale.util.*; @SuppressWarnings("unchecked") public final class Rest { + //请求所需的RestService的资源名，值类型: 字符串 public static final String REST_HEADER_RESOURCE_NAME = "rest-resource-name"; + //请求是否为rpc协议，值类型: 布尔，取值为true、false public static final String REST_HEADER_RPC = "rest-rpc"; - public static final String REST_HEADER_CURRUSERID_NAME = "rest-curruserid-name"; + //当前用户ID值，值类型: 字符串 + public static final String REST_HEADER_CURRUSERID = "rest-curruserid"; - public static final String REST_HEADER_PARAM_FROM_BODY = "rest-paramfrombody"; + //参数是否从body中获取，值类型: 布尔，取值为true、false + public static final String REST_HEADER_PARAM_FROM_BODY = "rest-param-from-body"; + //请求参数的反序列化种类，值类型: 字符串，取值为ConvertType枚举值名 public static final String REST_HEADER_REQ_CONVERT_TYPE = "rest-req-convert-type"; + //响应结果的序列化种类，值类型: 字符串，取值为ConvertType枚举值名 public static final String REST_HEADER_RESP_CONVERT_TYPE = "rest-resp-convert-type"; static final String REST_TOSTRINGOBJ_FIELD_NAME = "_redkale_tostringsupplier"; @@ -574,10 +580,6 @@ public final class Rest { MethodDebugVisitor.pushInt(mv, rws.wsmaxbody()); mv.visitFieldInsn(PUTFIELD, newDynName, "wsmaxbody", "I"); - mv.visitVarInsn(ALOAD, 0); - mv.visitInsn(rws.mergemsg() ? ICONST_1 : ICONST_0); - mv.visitFieldInsn(PUTFIELD, newDynName, "mergemsg", "Z"); - mv.visitVarInsn(ALOAD, 0); mv.visitInsn(rws.single() ? ICONST_1 : ICONST_0); mv.visitFieldInsn(PUTFIELD, newDynName, "single", "Z"); @@ -1053,7 +1055,6 @@ public final class Rest { final String httpScopeDesc = Type.getDescriptor(HttpScope.class); final String stageDesc = Type.getDescriptor(CompletionStage.class); final String flipperDesc = Type.getDescriptor(Flipper.class); - final String channelDesc = Type.getDescriptor(ChannelContext.class); final String httpServletName = HttpServlet.class.getName().replace('.', '/'); final String actionEntryName = HttpServlet.ActionEntry.class.getName().replace('.', '/'); final String attrDesc = Type.getDescriptor(org.redkale.util.Attribute.class); @@ -1330,7 +1331,6 @@ public final class Rest { } else if ("&".equals(pname) && ptype == userType) { //当前用户对象的类名 } else if (ptype.isPrimitive()) { } else if (ptype == String.class) { - } else if (ptype == ChannelContext.class) { } else if (ptype == Flipper.class) { } else { //其他Json对象 //构建 RestHeader、RestCookie、RestAddress 等赋值操作 @@ -1561,7 +1561,7 @@ public final class Rest { av0.visit("value", Type.getType(Type.getDescriptor(serviceType))); av0.visitEnd(); } - boolean dynsimple = true; + boolean dynsimple = baseServletType != HttpServlet.class; //有自定义的BaseServlet会存在读取header的操作 //获取所有可以转换成HttpMapping的方法 int methodidex = 0; final MessageMultiConsumer mmc = serviceType.getAnnotation(MessageMultiConsumer.class); @@ -2729,11 +2729,6 @@ public final class Rest { mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, iscookie ? "getCookie" : (ishead ? "getHeader" : "getParameter"), "(Ljava/lang/String;Ljava/lang/String;)Ljava/lang/String;", false); mv.visitVarInsn(ASTORE, maxLocals); varInsns.add(new int[]{ALOAD, maxLocals}); - } else if (ptype == ChannelContext.class) { - mv.visitVarInsn(ALOAD, 1); - mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getChannelContext", "()" + channelDesc, false); - mv.visitVarInsn(ASTORE, maxLocals); - varInsns.add(new int[]{ALOAD, maxLocals}); } else if (ptype == Flipper.class) { mv.visitVarInsn(ALOAD, 1); mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getFlipper", "()" + flipperDesc, false); diff --git a/src/main/java/org/redkale/net/http/RestLocale.java b/src/main/java/org/redkale/net/http/RestLocale.java index d9b581a39..89e9d4b56 100644 --- a/src/main/java/org/redkale/net/http/RestLocale.java +++ b/src/main/java/org/redkale/net/http/RestLocale.java @@ -2,9 +2,9 @@ */ package org.redkale.net.http; +import static java.lang.annotation.RetentionPolicy.RUNTIME; import java.lang.annotation.*; import static java.lang.annotation.ElementType.*; -import static java.lang.annotation.RetentionPolicy.RUNTIME; /** * 只能注解于Service类的方法的String参数或参数内的String字段 @@ -24,7 +24,7 @@ import static java.lang.annotation.RetentionPolicy.RUNTIME; public @interface RestLocale { /** - * 备注描述, 对应@HttpParam.comment + * 备注描述 * * @return String */ diff --git a/src/main/java/org/redkale/net/http/RestWebSocket.java b/src/main/java/org/redkale/net/http/RestWebSocket.java index 0cdc18b44..34209adc9 100644 --- a/src/main/java/org/redkale/net/http/RestWebSocket.java +++ b/src/main/java/org/redkale/net/http/RestWebSocket.java @@ -5,9 +5,9 @@ */ package org.redkale.net.http; -import java.lang.annotation.*; import static java.lang.annotation.ElementType.TYPE; import static java.lang.annotation.RetentionPolicy.RUNTIME; +import java.lang.annotation.*; import org.redkale.net.Cryptor; /** @@ -60,13 +60,6 @@ public @interface RestWebSocket { */ boolean anyuser() default false; - /** - * 接收客户端的分包(last=false)消息时是否自动合并包 - * - * @return 默认true - */ - boolean mergemsg() default true; - /** * WebScoket服务器给客户端进行ping操作的间隔时间, 单位: 秒， 默认值：15秒 * diff --git a/src/main/java/org/redkale/net/http/WebSocketEngine.java b/src/main/java/org/redkale/net/http/WebSocketEngine.java index c0aa2c4e7..fcf95c6a6 100644 --- a/src/main/java/org/redkale/net/http/WebSocketEngine.java +++ b/src/main/java/org/redkale/net/http/WebSocketEngine.java @@ -76,14 +76,11 @@ public class WebSocketEngine { @Comment("最大消息体长度, 小于1表示无限制") protected int wsMaxBody; - @Comment("接收客户端的分包(last=false)消息时是否自动合并包") - protected boolean mergeMode = true; - @Comment("加密解密器") protected Cryptor cryptor; - protected WebSocketEngine(String engineid, boolean single, HttpContext context, int liveInterval, int wsMaxConns, - int wsThreads, int wsMaxBody, boolean mergeMode, Cryptor cryptor, WebSocketNode node, Convert sendConvert, Logger logger) { + protected WebSocketEngine(String engineid, boolean single, HttpContext context, int liveInterval, + int wsMaxConns, int wsThreads, int wsMaxBody, Cryptor cryptor, WebSocketNode node, Convert sendConvert, Logger logger) { this.engineid = engineid; this.single = single; this.context = context; @@ -93,7 +90,6 @@ public class WebSocketEngine { this.wsMaxConns = wsMaxConns; this.wsThreads = wsThreads; this.wsMaxBody = wsMaxBody; - this.mergeMode = mergeMode; this.cryptor = cryptor; this.logger = logger; this.index = sequence.getAndIncrement(); diff --git a/src/main/java/org/redkale/net/http/WebSocketServlet.java b/src/main/java/org/redkale/net/http/WebSocketServlet.java index 1089565b0..756444687 100644 --- a/src/main/java/org/redkale/net/http/WebSocketServlet.java +++ b/src/main/java/org/redkale/net/http/WebSocketServlet.java @@ -60,9 +60,6 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl @Comment("最大消息体长度, 小于1表示无限制") public static final String WEBPARAM__WSMAXBODY = "wsmaxbody"; - @Comment("接收客户端的分包(last=false)消息时是否自动合并包") - public static final String WEBPARAM__WSMERGEMSG = "wsmergemsg"; - @Comment("加密解密器") public static final String WEBPARAM__CRYPTOR = "cryptor"; @@ -93,9 +90,6 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl //同RestWebSocket.anyuser protected boolean anyuser = false; - //同RestWebSocket.mergemsg - protected boolean mergemsg = true; - //同RestWebSocket.cryptor, 变量名不可改， 被Rest.createRestWebSocketServlet用到 protected Cryptor cryptor; @@ -197,7 +191,7 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl } //存在WebSocketServlet，则此WebSocketNode必须是本地模式Service this.webSocketNode.localEngine = new WebSocketEngine("WebSocketEngine-" + addr.getHostString() + ":" + addr.getPort() + "-[" + resourceName() + "]", - this.single, context, liveinterval, wsmaxconns, wsthreads, wsmaxbody, mergemsg, this.cryptor, this.webSocketNode, this.sendConvert, logger); + this.single, context, liveinterval, wsmaxconns, wsthreads, wsmaxbody, this.cryptor, this.webSocketNode, this.sendConvert, logger); this.webSocketNode.init(conf); this.webSocketNode.localEngine.init(conf);