lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加HttpHeader

Browse Source
This commit is contained in:
redkale
2023-11-20 18:17:48 +08:00
parent ae91e47a73
commit 753d8b020a
26 changed files with 1369 additions and 636 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/redkale/annotation/ResourceListener.java
View File

@@ -26,7 +26,7 @@ import static java.lang.annotation.RetentionPolicy.RUNTIME;
* @ResourceListener
* private void changeResource(ResourceEvent[] events) {
* for(ResourceEvent event : events) {
* System.out.println("@Resource = " + event.name() + " 资源变更: newVal = " + event.newValue() + ", oldVal = " + event.oldValue());
* System.out .println("@Resource = " + event.name() + " 资源变更: newVal = " + event.newValue() + ", oldVal = " + event.oldValue());
* }
* }
*

367
src/main/java/org/redkale/boot/Application.java
View File

@@ -25,6 +25,7 @@ import org.redkale.cluster.*;
import org.redkale.convert.Convert;
import org.redkale.convert.bson.BsonFactory;
import org.redkale.convert.json.*;
import org.redkale.convert.protobuf.ProtobufFactory;
import org.redkale.mq.*;
import org.redkale.net.*;
import org.redkale.net.http.*;
@@ -268,40 +269,55 @@ public final class Application {
this.config = config;
this.configFromCache = "true".equals(config.getValue("[config-from-cache]"));
this.environment = new Environment(this.envProperties);
//设置APP_HOME、APP_NAME
//设置APP_NAME
this.name = checkName(config.getValue("name", ""));
this.resourceFactory.register(RESNAME_APP_NAME, name);
System.setProperty(RESNAME_APP_NAME, name);
final File root = new File(System.getProperty(RESNAME_APP_HOME));
this.resourceFactory.register(RESNAME_APP_TIME, long.class, this.startTime);
this.resourceFactory.register(RESNAME_APP_HOME, Path.class, root.toPath());
this.resourceFactory.register(RESNAME_APP_HOME, File.class, root);
this.resourceFactory.register(RESNAME_APP_HOME, URI.class, root.toURI());
File confFile = null;
try { //设置APP_HOME
this.resourceFactory.register(RESNAME_APP_HOME, root.getCanonicalPath());
{ //设置APP_HOME
final File root = new File(System.getProperty(RESNAME_APP_HOME));
final String rootPath = getCanonicalPath(root);
this.resourceFactory.register(RESNAME_APP_TIME, long.class, this.startTime);
this.resourceFactory.register(RESNAME_APP_HOME, Path.class, root.toPath());
this.resourceFactory.register(RESNAME_APP_HOME, File.class, root);
this.resourceFactory.register(RESNAME_APP_HOME, URI.class, root.toURI());
File confFile = null;
this.resourceFactory.register(RESNAME_APP_HOME, rootPath);
if (System.getProperty(RESNAME_APP_HOME) == null) {
System.setProperty(RESNAME_APP_HOME, root.getCanonicalPath());
System.setProperty(RESNAME_APP_HOME, rootPath);
}
this.home = root.getCanonicalFile();
this.home = new File(rootPath);
this.homePath = this.home.getPath();
String confDir = System.getProperty(RESNAME_APP_CONF_DIR, "conf");
if (confDir.contains("://") || confDir.startsWith("file:") || confDir.startsWith("resource:") || confDir.contains("!")) { //graalvm native-image startwith resource:META-INF
this.confPath = URI.create(confDir);
if (confDir.startsWith("file:")) {
confFile = new File(this.confPath.getPath()).getCanonicalFile();
confFile = getCanonicalFile(new File(this.confPath.getPath()));
}
} else if (confDir.charAt(0) == '/' || confDir.indexOf(':') > 0) {
confFile = new File(confDir).getCanonicalFile();
} else if (confDir.charAt(0) == '/' || confDir.indexOf(':') > -1) {
confFile = getCanonicalFile(new File(confDir));
this.confPath = confFile.toURI();
} else {
confFile = new File(this.home, confDir).getCanonicalFile();
confFile = new File(getCanonicalPath(new File(this.home, confDir)));
this.confPath = confFile.toURI();
}
} catch (IOException e) {
throw new RedkaleException(e);
if (confFile != null) {
this.resourceFactory.register(RESNAME_APP_CONF_DIR, File.class, confFile);
this.resourceFactory.register(RESNAME_APP_CONF_DIR, Path.class, confFile.toPath());
}
String localaddr = getPropertyValue(config.getValue("address", "").trim());
InetAddress addr = localaddr.isEmpty() ? Utility.localInetAddress() : new InetSocketAddress(localaddr, config.getIntValue("port")).getAddress();
this.localAddress = new InetSocketAddress(addr, config.getIntValue("port"));
this.resourceFactory.register(RESNAME_APP_ADDR, addr.getHostAddress());
this.resourceFactory.register(RESNAME_APP_ADDR, InetAddress.class, addr);
this.resourceFactory.register(RESNAME_APP_ADDR, InetSocketAddress.class, this.localAddress);
this.resourceFactory.register(RESNAME_APP_CONF_DIR, URI.class, this.confPath);
this.resourceFactory.register(RESNAME_APP_CONF_DIR, String.class, this.confPath.toString());
}
{ //设置系统变量
System.setProperty("redkale.version", Redkale.getDotedVersion());
int nid = config.getIntValue("nodeid", 0);
@@ -335,23 +351,9 @@ public final class Application {
sysProperties.forEach((key, value) -> {
System.setProperty(key.toString(), getPropertyValue(value.toString(), sysProperties));
});
this.resourceFactory.register(Environment.class, environment);
}
String localaddr = getPropertyValue(config.getValue("address", "").trim());
InetAddress addr = localaddr.isEmpty() ? Utility.localInetAddress() : new InetSocketAddress(localaddr, config.getIntValue("port")).getAddress();
this.localAddress = new InetSocketAddress(addr, config.getIntValue("port"));
this.resourceFactory.register(RESNAME_APP_ADDR, addr.getHostAddress());
this.resourceFactory.register(RESNAME_APP_ADDR, InetAddress.class, addr);
this.resourceFactory.register(RESNAME_APP_ADDR, InetSocketAddress.class, this.localAddress);
this.resourceFactory.register(RESNAME_APP_CONF_DIR, URI.class, this.confPath);
this.resourceFactory.register(RESNAME_APP_CONF_DIR, String.class, this.confPath.toString());
if (confFile != null) {
this.resourceFactory.register(RESNAME_APP_CONF_DIR, File.class, confFile);
this.resourceFactory.register(RESNAME_APP_CONF_DIR, Path.class, confFile.toPath());
}
this.resourceFactory.register(Environment.class, environment);
{ //初始化ClassLoader
ClassLoader currClassLoader = Thread.currentThread().getContextClassLoader();
if (currClassLoader instanceof RedkaleClassLoader) {
@@ -364,7 +366,7 @@ public final class Application {
if (in != null) {
BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8), 1024);
List<String> list = new ArrayList<>();
reader.lines().forEach(v -> list.add(v));
reader.lines().forEach(list::add);
Collections.sort(list);
if (!list.isEmpty()) {
cacheClasses = new LinkedHashSet<>(list);
@@ -382,8 +384,14 @@ public final class Application {
}
Thread.currentThread().setContextClassLoader(this.classLoader);
}
if (compileMode || this.classLoader instanceof RedkaleClassLoader.RedkaleCacheClassLoader) {
this.serverClassLoader = this.classLoader;
} else {
this.serverClassLoader = new RedkaleClassLoader(this.classLoader);
}
}
{ //以下是初始化日志配置
{ //初始化日志配置
URI logConfURI;
File logConfFile = null;
if (configFromCache) {
@@ -420,13 +428,12 @@ public final class Application {
RedkaleClassLoader.putReflectionPublicConstructors(LoggingFileHandler.LoggingConsoleHandler.class, LoggingFileHandler.LoggingConsoleHandler.class.getName());
RedkaleClassLoader.putReflectionPublicConstructors(LoggingFileHandler.LoggingSncpFileHandler.class, LoggingFileHandler.LoggingSncpFileHandler.class.getName());
}
this.logger = Logger.getLogger(this.getClass().getSimpleName());
this.shutdownLatch = new CountDownLatch(config.getAnyValues("server").length + 1);
logger.log(Level.INFO, colorMessage(logger, 36, 1, "-------------------------------- Redkale " + Redkale.getDotedVersion() + " --------------------------------"));
}
this.logger = Logger.getLogger(this.getClass().getSimpleName());
this.shutdownLatch = new CountDownLatch(config.getAnyValues("server").length + 1);
logger.log(Level.INFO, colorMessage(logger, 36, 1, "-------------------------------- Redkale " + Redkale.getDotedVersion() + " --------------------------------"));
//------------------------------------ 基本设置 ------------------------------------
{
{ //基本设置
final String confDir = this.confPath.toString();
logger.log(Level.INFO, "APP_OS = " + System.getProperty("os.name") + " " + System.getProperty("os.version") + " " + System.getProperty("os.arch") + "\r\n"
+ "APP_JAVA = " + System.getProperty("java.runtime.name", System.getProperty("org.graalvm.nativeimage.kind") != null ? "Nativeimage" : "")
@@ -447,154 +454,179 @@ public final class Application {
this.resourceFactory.register(BsonFactory.root());
this.resourceFactory.register(JsonFactory.root());
this.resourceFactory.register(ProtobufFactory.root());
this.resourceFactory.register(BsonFactory.root().getConvert());
this.resourceFactory.register(JsonFactory.root().getConvert());
this.resourceFactory.register(ProtobufFactory.root().getConvert());
this.resourceFactory.register("bsonconvert", Convert.class, BsonFactory.root().getConvert());
this.resourceFactory.register("jsonconvert", Convert.class, JsonFactory.root().getConvert());
this.resourceFactory.register("protobufconvert", Convert.class, ProtobufFactory.root().getConvert());
}
//------------------------------------ 读取配置 ------------------------------------
try {
try { //读取配置
loadResourceProperties();
} catch (IOException e) {
throw new RedkaleException(e);
}
//------------------------------------ 配置 <xxxagent> 节点 ------------------------------------
ClusterAgent cluster = null;
MessageAgent[] mqs = null;
int bufferCapacity = 32 * 1024;
int bufferPoolSize = Utility.cpus() * 8;
AnyValue executorConf = null;
executorConf = config.getAnyValue("executor");
AnyValue clusterConf = config.getAnyValue("cluster");
if (clusterConf != null) {
try {
String classVal = getPropertyValue(clusterConf.getValue("type", clusterConf.getValue("value"))); //兼容value字段
if (classVal == null || classVal.isEmpty() || classVal.indexOf('.') < 0) { //不包含.表示非类名，比如值: consul, nacos
Iterator<ClusterAgentProvider> it = ServiceLoader.load(ClusterAgentProvider.class, classLoader).iterator();
RedkaleClassLoader.putServiceLoader(ClusterAgentProvider.class);
while (it.hasNext()) {
ClusterAgentProvider provider = it.next();
if (provider != null) {
RedkaleClassLoader.putReflectionPublicConstructors(provider.getClass(), provider.getClass().getName()); //loader class
}
if (provider != null && provider.acceptsConf(clusterConf)) {
cluster = provider.createInstance();
cluster.setConfig(clusterConf);
break;
}
}
if (cluster == null) {
ClusterAgent cacheClusterAgent = new CacheClusterAgent();
if (cacheClusterAgent.acceptsConf(clusterConf)) {
cluster = cacheClusterAgent;
cluster.setConfig(clusterConf);
}
}
if (cluster == null) {
logger.log(Level.SEVERE, "load application cluster resource, but not found name='type' value error: " + clusterConf);
}
} else {
Class type = classLoader.loadClass(classVal);
if (!ClusterAgent.class.isAssignableFrom(type)) {
logger.log(Level.SEVERE, "load application cluster resource, but not found " + ClusterAgent.class.getSimpleName() + " implements class error: " + clusterConf);
} else {
RedkaleClassLoader.putReflectionDeclaredConstructors(type, type.getName());
cluster = (ClusterAgent) type.getDeclaredConstructor().newInstance();
cluster.setConfig(clusterConf);
}
}
//此时不能执行cluster.init，因内置的对象可能依赖config.properties配置项
} catch (Exception e) {
logger.log(Level.SEVERE, "load application cluster resource error: " + clusterConf, e);
}
}
{ //加载XXXAgent
ClusterAgent cluster = null;
MessageAgent[] mqs = null;
AnyValue[] mqConfs = config.getAnyValues("mq");
if (mqConfs != null && mqConfs.length > 0) {
mqs = new MessageAgent[mqConfs.length];
Set<String> mqnames = new HashSet<>();
for (int i = 0; i < mqConfs.length; i++) {
AnyValue mqConf = mqConfs[i];
String names = getPropertyValue(mqConf.getValue("name")); //含,或者;表示多个别名使用同一mq对象
if (names != null && !names.isEmpty()) {
for (String n : names.replace(',', ';').split(";")) {
if (n.trim().isEmpty()) {
continue;
}
if (mqnames.contains(n.trim())) {
throw new RedkaleException("mq.name(" + n.trim() + ") is repeat");
}
mqnames.add(n.trim());
}
} else if (names != null && names.isEmpty()) {
String n = "";
if (mqnames.contains(n.trim())) {
throw new RedkaleException("mq.name(" + n.trim() + ") is repeat");
}
mqnames.add(n);
}
AnyValue clusterConf = config.getAnyValue("cluster");
if (clusterConf != null) {
try {
String classVal = getPropertyValue(mqConf.getValue("type", mqConf.getValue("value"))); //兼容value字段
if (classVal == null || classVal.isEmpty() || classVal.indexOf('.') < 0) { //不包含.表示非类名，比如值: kafka, pulsar
Iterator<MessageAgentProvider> it = ServiceLoader.load(MessageAgentProvider.class, classLoader).iterator();
RedkaleClassLoader.putServiceLoader(MessageAgentProvider.class);
String classVal = getPropertyValue(clusterConf.getValue("type", clusterConf.getValue("value"))); //兼容value字段
if (classVal == null || classVal.isEmpty() || classVal.indexOf('.') < 0) { //不包含.表示非类名，比如值: consul, nacos
Iterator<ClusterAgentProvider> it = ServiceLoader.load(ClusterAgentProvider.class, classLoader).iterator();
RedkaleClassLoader.putServiceLoader(ClusterAgentProvider.class);
while (it.hasNext()) {
MessageAgentProvider provider = it.next();
ClusterAgentProvider provider = it.next();
if (provider != null) {
RedkaleClassLoader.putReflectionPublicConstructors(provider.getClass(), provider.getClass().getName()); //loader class
}
if (provider != null && provider.acceptsConf(mqConf)) {
mqs[i] = provider.createInstance();
mqs[i].setConfig(mqConf);
if (provider != null && provider.acceptsConf(clusterConf)) {
cluster = provider.createInstance();
cluster.setConfig(clusterConf);
break;
}
}
if (mqs[i] == null) {
logger.log(Level.SEVERE, "load application mq resource, but not found name='value' value error: " + mqConf);
if (cluster == null) {
ClusterAgent cacheClusterAgent = new CacheClusterAgent();
if (cacheClusterAgent.acceptsConf(clusterConf)) {
cluster = cacheClusterAgent;
cluster.setConfig(clusterConf);
}
}
if (cluster == null) {
logger.log(Level.SEVERE, "load application cluster resource, but not found name='type' value error: " + clusterConf);
}
} else {
Class type = classLoader.loadClass(classVal);
if (!MessageAgent.class.isAssignableFrom(type)) {
logger.log(Level.SEVERE, "load application mq resource, but not found " + MessageAgent.class.getSimpleName() + " implements class error: " + mqConf);
if (!ClusterAgent.class.isAssignableFrom(type)) {
logger.log(Level.SEVERE, "load application cluster resource, but not found " + ClusterAgent.class.getSimpleName() + " implements class error: " + clusterConf);
} else {
RedkaleClassLoader.putReflectionDeclaredConstructors(type, type.getName());
mqs[i] = (MessageAgent) type.getDeclaredConstructor().newInstance();
mqs[i].setConfig(mqConf);
cluster = (ClusterAgent) type.getDeclaredConstructor().newInstance();
cluster.setConfig(clusterConf);
}
}
//此时不能执行mq.init，因内置的对象可能依赖config.properties配置项
//此时不能执行cluster.init，因内置的对象可能依赖config.properties配置项
} catch (Exception e) {
logger.log(Level.SEVERE, "load application mq resource error: " + mqs[i], e);
logger.log(Level.SEVERE, "load application cluster resource error: " + clusterConf, e);
}
}
}
ExecutorService workExecutor0 = null;
if (executorConf == null) {
executorConf = DefaultAnyValue.create();
}
final int workThreads = executorConf.getIntValue("threads", Utility.cpus() * 4);
if (workThreads > 0) {
//指定threads则不使用虚拟线程池
workExecutor0 = executorConf.getValue("threads") != null ? WorkThread.createExecutor(workThreads, "Redkale-WorkThread-%s") : WorkThread.createWorkExecutor(workThreads, "Redkale-WorkThread-%s");
}
this.workExecutor = workExecutor0;
this.resourceFactory.register(RESNAME_APP_EXECUTOR, Executor.class, this.workExecutor);
this.resourceFactory.register(RESNAME_APP_EXECUTOR, ExecutorService.class, this.workExecutor);
{
ExecutorService clientExecutor = workExecutor0;
if (clientExecutor == null) {
//给所有client给一个默认的ExecutorService
clientExecutor = WorkThread.createWorkExecutor(executorConf.getIntValue("clients", Utility.cpus()), "Redkale-DefaultClient-WorkThread-%s");
AnyValue[] mqConfs = config.getAnyValues("mq");
if (mqConfs != null && mqConfs.length > 0) {
mqs = new MessageAgent[mqConfs.length];
Set<String> mqnames = new HashSet<>();
for (int i = 0; i < mqConfs.length; i++) {
AnyValue mqConf = mqConfs[i];
String names = getPropertyValue(mqConf.getValue("name")); //含,或者;表示多个别名使用同一mq对象
if (names != null && !names.isEmpty()) {
for (String n : names.replace(',', ';').split(";")) {
if (n.trim().isEmpty()) {
continue;
}
if (mqnames.contains(n.trim())) {
throw new RedkaleException("mq.name(" + n.trim() + ") is repeat");
}
mqnames.add(n.trim());
}
} else if (names != null && names.isEmpty()) {
String n = "";
if (mqnames.contains(n.trim())) {
throw new RedkaleException("mq.name(" + n.trim() + ") is repeat");
}
mqnames.add(n);
}
try {
String classVal = getPropertyValue(mqConf.getValue("type", mqConf.getValue("value"))); //兼容value字段
if (classVal == null || classVal.isEmpty() || classVal.indexOf('.') < 0) { //不包含.表示非类名，比如值: kafka, pulsar
Iterator<MessageAgentProvider> it = ServiceLoader.load(MessageAgentProvider.class, classLoader).iterator();
RedkaleClassLoader.putServiceLoader(MessageAgentProvider.class);
while (it.hasNext()) {
MessageAgentProvider provider = it.next();
if (provider != null) {
RedkaleClassLoader.putReflectionPublicConstructors(provider.getClass(), provider.getClass().getName()); //loader class
}
if (provider != null && provider.acceptsConf(mqConf)) {
mqs[i] = provider.createInstance();
mqs[i].setConfig(mqConf);
break;
}
}
if (mqs[i] == null) {
logger.log(Level.SEVERE, "load application mq resource, but not found name='value' value error: " + mqConf);
}
} else {
Class type = classLoader.loadClass(classVal);
if (!MessageAgent.class.isAssignableFrom(type)) {
logger.log(Level.SEVERE, "load application mq resource, but not found " + MessageAgent.class.getSimpleName() + " implements class error: " + mqConf);
} else {
RedkaleClassLoader.putReflectionDeclaredConstructors(type, type.getName());
mqs[i] = (MessageAgent) type.getDeclaredConstructor().newInstance();
mqs[i].setConfig(mqConf);
}
}
//此时不能执行mq.init，因内置的对象可能依赖config.properties配置项
} catch (Exception e) {
logger.log(Level.SEVERE, "load application mq resource error: " + mqs[i], e);
}
}
}
this.clusterAgent = cluster;
this.messageAgents = mqs;
}
{ //设置WorkExecutor
int bufferCapacity = 32 * 1024;
int bufferPoolSize = Utility.cpus() * 8;
final AnyValue executorConf = config.getAnyValue("executor", true);
StringBuilder executorLog = new StringBuilder();
ExecutorService workExecutor0 = null;
final int workThreads = executorConf.getIntValue("threads", Utility.cpus() * 4);
if (workThreads > 0) {
//指定threads则不使用虚拟线程池
workExecutor0 = executorConf.getValue("threads") != null ? WorkThread.createExecutor(workThreads, "Redkale-WorkThread-%s") : WorkThread.createWorkExecutor(workThreads, "Redkale-WorkThread-%s");
String executorName = workExecutor0.getClass().getSimpleName();
executorLog.append("defaultWorkExecutor: {type=" + executorName);
if (executorName.contains("VirtualExecutor") || executorName.contains("PerTaskExecutor")) {
executorLog.append(", threads=[virtual]}");
} else {
executorLog.append(", threads=" + workThreads + "}");
}
}
this.workExecutor = workExecutor0;
this.resourceFactory.register(RESNAME_APP_EXECUTOR, Executor.class, this.workExecutor);
this.resourceFactory.register(RESNAME_APP_EXECUTOR, ExecutorService.class, this.workExecutor);
ExecutorService clientWorkExecutor = workExecutor0;
if (clientWorkExecutor == null) {
//给所有client给一个默认的ExecutorService
int clients = executorConf.getIntValue("clients", Utility.cpus());
clientWorkExecutor = WorkThread.createWorkExecutor(clients, "Redkale-DefaultClient-WorkThread-%s");
String executorName = clientWorkExecutor.getClass().getSimpleName();
executorLog.append("clientWorkExecutor: {type=" + executorName);
if (executorName.contains("VirtualExecutor") || executorName.contains("PerTaskExecutor")) {
executorLog.append(", threads=[virtual]}");
} else {
executorLog.append(", threads=" + workThreads + "}");
}
} else {
executorLog.append(", clientWorkExecutor: [workExecutor]");
}
this.clientAsyncGroup = new AsyncIOGroup("Redkale-DefaultClient-IOThread-%s", clientWorkExecutor, bufferCapacity, bufferPoolSize).skipClose(true);
this.resourceFactory.register(RESNAME_APP_CLIENT_ASYNCGROUP, AsyncGroup.class, this.clientAsyncGroup);
this.resourceFactory.register(RESNAME_APP_CLIENT_ASYNCGROUP, AsyncIOGroup.class, this.clientAsyncGroup);
if (executorLog.length() > 0) {
logger.log(Level.INFO, executorLog.toString());
}
this.clientAsyncGroup = new AsyncIOGroup("Redkale-DefaultClient-IOThread-%s", clientExecutor, bufferCapacity, bufferPoolSize).skipClose(true);
}
this.resourceFactory.register(RESNAME_APP_CLIENT_ASYNCGROUP, AsyncGroup.class, this.clientAsyncGroup);
this.resourceFactory.register(RESNAME_APP_CLIENT_ASYNCGROUP, AsyncIOGroup.class, this.clientAsyncGroup);
this.clusterAgent = cluster;
this.messageAgents = mqs;
{ //加载原生sql解析器
Iterator<DataNativeSqlParserProvider> it = ServiceLoader.load(DataNativeSqlParserProvider.class, classLoader).iterator();
RedkaleClassLoader.putServiceLoader(DataNativeSqlParserProvider.class);
@@ -609,14 +641,9 @@ public final class Application {
for (DataNativeSqlParserProvider provider : InstanceProvider.sort(providers)) {
this.nativeSqlParser = provider.createInstance();
this.resourceFactory.register(DataNativeSqlParser.class, this.nativeSqlParser);
break;
break; //only first provider
}
}
if (compileMode || this.classLoader instanceof RedkaleClassLoader.RedkaleCacheClassLoader) {
this.serverClassLoader = this.classLoader;
} else {
this.serverClassLoader = new RedkaleClassLoader(this.classLoader);
}
}
private void loadResourceProperties() throws IOException {
@@ -1008,6 +1035,22 @@ public final class Application {
return name;
}
private String getCanonicalPath(File file) {
try {
return file.getCanonicalPath();
} catch (IOException e) {
return file.getAbsolutePath();
}
}
private File getCanonicalFile(File file) {
try {
return file.getCanonicalFile();
} catch (IOException e) {
return file;
}
}
public void init() throws Exception {
//只有WatchService才能加载Application、WatchFactory
final Application application = this;
@@ -1585,7 +1628,7 @@ public final class Application {
if (logger != null) {
logger.info("Send: " + cmd + ", Reply: " + rs);
}
System.out.println(rs);
(System.out).println(rs);
} catch (Exception e) {
if (e instanceof PortUnreachableException) {
if ("APIDOC".equalsIgnoreCase(cmd)) {
@@ -1597,7 +1640,7 @@ public final class Application {
if (logger != null) {
logger.info(rs);
}
System.out.println(rs);
(System.out).println(rs);
return;
}
//if ("SHUTDOWN".equalsIgnoreCase(cmd)) {

46
src/main/java/org/redkale/cluster/HttpClusterRpcClient.java
View File

@@ -82,7 +82,7 @@ public class HttpClusterRpcClient extends HttpRpcClient {
String module = req.getRequestURI();
module = module.substring(1); //去掉/
module = module.substring(0, module.indexOf('/'));
Map<String, String> headers = req.getHeaders();
HttpHeader headers = req.getHeaders();
String resname = req.getHeader(Rest.REST_HEADER_RESNAME, "");
final String localModule = module;
if (logger.isLoggable(Level.FINEST)) {
@@ -96,34 +96,34 @@ public class HttpClusterRpcClient extends HttpRpcClient {
}
return new HttpResult<byte[]>().status(404).toFuture();
}
final Map<String, String> clientHeaders = new LinkedHashMap<>();
byte[] clientBody = null;
if (req.isRpc()) {
clientHeaders.put(Rest.REST_HEADER_RPC, "true");
}
if (isNotEmpty(req.getTraceid())) {
clientHeaders.put(Rest.REST_HEADER_TRACEID, req.getTraceid());
}
if (req.getReqConvertType() != null) {
clientHeaders.put(Rest.REST_HEADER_REQ_CONVERT, req.getReqConvertType().toString());
}
if (req.getRespConvertType() != null) {
clientHeaders.put(Rest.REST_HEADER_RESP_CONVERT, req.getRespConvertType().toString());
}
if (userid != null) {
clientHeaders.put(Rest.REST_HEADER_CURRUSERID, "" + userid);
}
final HttpHeader clientHeaders = HttpHeader.create();
if (headers != null) {
boolean ws = headers.containsKey("Sec-WebSocket-Key");
boolean ws = headers.contains("Sec-WebSocket-Key");
headers.forEach((n, v) -> {
if (!DISALLOWED_HEADERS_SET.contains(n.toLowerCase())
&& (!ws || (!"Connection".equals(n) && !"Sec-WebSocket-Key".equals(n)
&& !"Sec-WebSocket-Version".equals(n)))) {
clientHeaders.put(n, v);
clientHeaders.add(n, v);
}
});
}
clientHeaders.put("Content-Type", "x-www-form-urlencoded");
byte[] clientBody = null;
if (req.isRpc()) {
clientHeaders.set(Rest.REST_HEADER_RPC, "true");
}
if (isNotEmpty(req.getTraceid())) {
clientHeaders.set(Rest.REST_HEADER_TRACEID, req.getTraceid());
}
if (req.getReqConvertType() != null) {
clientHeaders.set(Rest.REST_HEADER_REQ_CONVERT, req.getReqConvertType().toString());
}
if (req.getRespConvertType() != null) {
clientHeaders.set(Rest.REST_HEADER_RESP_CONVERT, req.getRespConvertType().toString());
}
if (userid != null) {
clientHeaders.set(Rest.REST_HEADER_CURRUSERID, "" + userid);
}
clientHeaders.set("Content-Type", "x-www-form-urlencoded");
if (req.getBody() != null && req.getBody().length > 0) {
String paramstr = req.getParametersToString();
if (paramstr != null) {
@@ -150,7 +150,7 @@ public class HttpClusterRpcClient extends HttpRpcClient {
}
private CompletableFuture<HttpResult<byte[]>> forEachCollectionFuture(final WorkThread workThread, boolean finest, Serializable userid,
HttpSimpleRequest req, String requesturi, final Map<String, String> clientHeaders, byte[] clientBody, Iterator<InetSocketAddress> it) {
HttpSimpleRequest req, String requesturi, final HttpHeader clientHeaders, byte[] clientBody, Iterator<InetSocketAddress> it) {
if (!it.hasNext()) {
return CompletableFuture.completedFuture(null);
}
@@ -173,7 +173,7 @@ public class HttpClusterRpcClient extends HttpRpcClient {
//存在sendHeader后不发送body数据的问题， java.net.http.HttpRequest的bug?
.method("POST", clientBody == null ? java.net.http.HttpRequest.BodyPublishers.noBody() : java.net.http.HttpRequest.BodyPublishers.ofByteArray(clientBody));
if (clientHeaders != null) {
clientHeaders.forEach((n, v) -> builder.header(n, v));
clientHeaders.forEach(builder::header);
}
return httpClient.sendAsync(builder.build(), java.net.http.HttpResponse.BodyHandlers.ofByteArray())
.thenApply((java.net.http.HttpResponse<byte[]> resp) -> {

4
src/main/java/org/redkale/cluster/HttpRpcClient.java
View File

@@ -7,7 +7,6 @@ package org.redkale.cluster;
import java.io.Serializable;
import java.lang.reflect.Type;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.http.*;
@@ -112,8 +111,7 @@ public abstract class HttpRpcClient implements ClusterRpcClient<HttpSimpleReques
}
module = module.substring(1); //去掉/
module = module.substring(0, module.indexOf('/'));
Map<String, String> headers = request.getHeaders();
String resname = headers == null ? "" : headers.getOrDefault(Rest.REST_HEADER_RESNAME, "");
String resname = request.getHeader(Rest.REST_HEADER_RESNAME, "");
return Rest.generateHttpReqTopic(module, resname, getNodeid());
}

6
src/main/java/org/redkale/mq/HttpResultCoder.java
View File

@@ -38,14 +38,14 @@ public class HttpResultCoder implements MessageCoder<HttpResult> {
public byte ctype() {
return MessageRecord.CTYPE_HTTP_RESULT;
}
@Override
public byte[] encode(HttpResult data) {
if (data == null) {
return null;
}
byte[] contentType = MessageCoder.getBytes(data.getContentType());
byte[] headers = MessageCoder.getBytes(data.getHeaders());
byte[] headers = MessageCoder.getSeriMapBytes(data.getHeaders());
byte[] cookies = getBytes(data.getCookies());
byte[] content;
if (data.getResult() == null) {
@@ -89,7 +89,7 @@ public class HttpResultCoder implements MessageCoder<HttpResult> {
HttpResult result = new HttpResult();
result.setStatus(buffer.getInt());
result.setContentType(MessageCoder.getSmallString(buffer));
result.setHeaders(MessageCoder.getMap(buffer));
result.setHeaders(MessageCoder.getSeriMap(buffer));
result.setCookies(getCookieList(buffer));
int len = buffer.getInt();
if (len > 0) {

89
src/main/java/org/redkale/mq/HttpSimpleRequestCoder.java
View File

@@ -5,10 +5,14 @@
*/
package org.redkale.mq;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import org.redkale.convert.ConvertType;
import org.redkale.net.http.HttpHeader;
import org.redkale.net.http.HttpSimpleRequest;
import org.redkale.util.Utility;
/**
* HttpSimpleRequest的MessageCoder实现
@@ -39,62 +43,97 @@ public class HttpSimpleRequestCoder implements MessageCoder<HttpSimpleRequest> {
byte[] traceid = MessageCoder.getBytes(data.getTraceid());//short-string
byte[] requestURI = MessageCoder.getBytes(data.getRequestURI()); //long-string
byte[] path = MessageCoder.getBytes(data.getPath()); //short-string
byte[] method = MessageCoder.getBytes(data.getMethod());//short-string
byte[] remoteAddr = MessageCoder.getBytes(data.getRemoteAddr());//short-string
byte[] sessionid = MessageCoder.getBytes(data.getSessionid());//short-string
byte[] contentType = MessageCoder.getBytes(data.getContentType());//short-string
byte[] headers = MessageCoder.getBytes(data.getHeaders());
byte[] params = MessageCoder.getBytes(data.getParams());
byte[] body = MessageCoder.getBytes(data.getBody());
byte[] userid = MessageCoder.encodeUserid(data.getCurrentUserid());
byte[] headers = MessageCoder.getSeriMapBytes(data.getHeaders() == null ? null : data.getHeaders().map());
byte[] params = MessageCoder.getStringMapBytes(data.getParams());
byte[] body = MessageCoder.getBytes(data.getBody());
int count = 1 //rpc
+ 4 //reqConvertType
+ 4 //respConvertType
+ 2 + traceid.length
+ 4 + requestURI.length
+ 2 + path.length
+ 2 + method.length
+ 2 + remoteAddr.length
+ 2 + sessionid.length
+ 2 + contentType.length
+ 2 + userid.length
+ headers.length + params.length
+ headers.length
+ params.length
+ 4 + body.length;
byte[] bs = new byte[count];
ByteBuffer buffer = ByteBuffer.wrap(bs);
buffer.put((byte) (data.isRpc() ? 0b01 : 0));
buffer.putInt(data.getReqConvertType() == null ? 0 : data.getReqConvertType().getValue());
buffer.putInt(data.getRespConvertType() == null ? 0 : data.getRespConvertType().getValue());
buffer.putChar((char) traceid.length);
if (traceid.length > 0) {
if (data.getTraceid() == null) {
buffer.putShort((short) -1);
} else {
buffer.putShort((short) traceid.length);
buffer.put(traceid);
}
buffer.putInt(requestURI.length);
if (requestURI.length > 0) {
if (data.getRequestURI() == null) {
buffer.putInt(-1);
} else {
buffer.putInt(requestURI.length);
buffer.put(requestURI);
}
buffer.putChar((char) path.length);
if (path.length > 0) {
if (data.getPath() == null) {
buffer.putShort((short) -1);
} else {
buffer.putShort((short) path.length);
buffer.put(path);
}
buffer.putChar((char) remoteAddr.length);
if (remoteAddr.length > 0) {
if (data.getMethod() == null) {
buffer.putShort((short) -1);
} else {
buffer.putShort((short) method.length);
buffer.put(method);
}
if (data.getRemoteAddr() == null) {
buffer.putShort((short) -1);
} else {
buffer.putShort((short) remoteAddr.length);
buffer.put(remoteAddr);
}
buffer.putChar((char) sessionid.length);
if (sessionid.length > 0) {
if (data.getSessionid() == null) {
buffer.putShort((short) -1);
} else {
buffer.putShort((short) sessionid.length);
buffer.put(sessionid);
}
buffer.putChar((char) contentType.length);
if (contentType.length > 0) {
if (data.getContentType() == null) {
buffer.putShort((short) -1);
} else {
buffer.putShort((short) contentType.length);
buffer.put(contentType);
}
buffer.putChar((char) userid.length);
if (userid.length > 0) {
if (data.getCurrentUserid() == null) {
buffer.putShort((short) -1);
} else {
buffer.putShort((short) userid.length);
buffer.put(userid);
}
buffer.put(headers);
buffer.put(params);
buffer.putInt(body.length);
if (body.length > 0) {
if (data.getBody() == null) {
buffer.putInt(-1);
} else {
buffer.putInt(body.length);
buffer.put(body);
}
return bs;
@@ -120,14 +159,18 @@ public class HttpSimpleRequestCoder implements MessageCoder<HttpSimpleRequest> {
req.setTraceid(MessageCoder.getSmallString(buffer));
req.setRequestURI(MessageCoder.getBigString(buffer));
req.setPath(MessageCoder.getSmallString(buffer));
req.setMethod(MessageCoder.getSmallString(buffer));
req.setRemoteAddr(MessageCoder.getSmallString(buffer));
req.setSessionid(MessageCoder.getSmallString(buffer));
req.setContentType(MessageCoder.getSmallString(buffer));
req.setCurrentUserid(MessageCoder.decodeUserid(buffer));
req.setHeaders(MessageCoder.getMap(buffer));
req.setParams(MessageCoder.getMap(buffer));
Map<String, Serializable> headerMap = MessageCoder.getSeriMap(buffer);
if (Utility.isNotEmpty(headerMap)) {
req.setHeaders(HttpHeader.ofValid(headerMap));
}
req.setParams(MessageCoder.getStringMap(buffer));
int len = buffer.getInt();
if (len > 0) {
if (len >= 0) {
byte[] bs = new byte[len];
buffer.get(bs);
req.setBody(bs);

124
src/main/java/org/redkale/mq/MessageCoder.java
View File

@@ -60,10 +60,10 @@ public interface MessageCoder<T> {
return Utility.append(new byte[]{(byte) 1}, str.getBytes(StandardCharsets.UTF_8));
}
//type: 1:string, 2:int, 3:long
//type: 1:string, 2:int, 3:long, 4:BigInteger
public static Serializable decodeUserid(ByteBuffer buffer) {
int len = buffer.getChar();
if (len == 0) {
int len = buffer.getShort();
if (len == -1) {
return null;
}
byte type = buffer.get();
@@ -95,7 +95,7 @@ public interface MessageCoder<T> {
return value.getBytes(StandardCharsets.UTF_8);
}
public static byte[] getBytes(final Map<String, String> map) {
public static byte[] getStringMapBytes(final Map<String, String> map) {
if (map == null || map.isEmpty()) {
return new byte[2];
}
@@ -106,7 +106,7 @@ public interface MessageCoder<T> {
});
final byte[] bs = new byte[len.get()];
final ByteBuffer buffer = ByteBuffer.wrap(bs);
buffer.putChar((char) map.size());
buffer.putShort((short) map.size());
map.forEach((key, value) -> {
putSmallString(buffer, key);
putBigString(buffer, value);
@@ -114,8 +114,53 @@ public interface MessageCoder<T> {
return bs;
}
public static Map<String, String> getStringMap(ByteBuffer buffer) {
int len = buffer.getShort();
if (len == 0) {
return null;
}
Map<String, String> map = new HashMap<>(len);
for (int i = 0; i < len; i++) {
map.put(getSmallString(buffer), getBigString(buffer));
}
return map;
}
public static byte[] getSeriMapBytes(final Map<String, Serializable> map) {
if (map == null || map.isEmpty()) {
return new byte[2];
}
final AtomicInteger len = new AtomicInteger(2);
map.forEach((key, value) -> {
len.addAndGet(2 + (key == null ? 0 : Utility.encodeUTF8Length(key)));
len.addAndGet(2 + (value == null ? 0 : lengthSeriStringOrList(value)));
});
final byte[] bs = new byte[len.get()];
final ByteBuffer buffer = ByteBuffer.wrap(bs);
buffer.putShort((short) map.size());
map.forEach((key, value) -> {
putSmallString(buffer, key);
putSeriStringOrList(buffer, value);
});
return bs;
}
public static Map<String, Serializable> getSeriMap(ByteBuffer buffer) {
int len = buffer.getShort();
if (len <= 0) {
return null;
}
Map<String, Serializable> map = new HashMap<>(len);
for (int i = 0; i < len; i++) {
map.put(getSmallString(buffer), getSeriStringOrList(buffer));
}
return map;
}
public static void putBigString(ByteBuffer buffer, String value) {
if (value == null || value.isEmpty()) {
if (value == null) {
buffer.putInt(-1);
} else if (value.isEmpty()) {
buffer.putInt(0);
} else {
byte[] bs = value.getBytes(StandardCharsets.UTF_8);
@@ -126,8 +171,10 @@ public interface MessageCoder<T> {
public static String getBigString(ByteBuffer buffer) {
int len = buffer.getInt();
if (len == 0) {
if (len == -1) {
return null;
} else if (len == 0) {
return "";
}
byte[] bs = new byte[len];
buffer.get(bs);
@@ -136,34 +183,69 @@ public interface MessageCoder<T> {
//一般用于存放类名、字段名、map中的key
public static void putSmallString(ByteBuffer buffer, String value) {
if (value == null || value.isEmpty()) {
buffer.putChar((char) 0);
if (value == null) {
buffer.putShort((short) -1);
} else if (value.isEmpty()) {
buffer.putShort((short) 0);
} else {
byte[] bs = value.getBytes(StandardCharsets.UTF_8);
buffer.putChar((char) bs.length);
buffer.putShort((short) bs.length);
buffer.put(bs);
}
}
public static String getSmallString(ByteBuffer buffer) {
int len = buffer.getChar();
if (len == 0) {
int len = buffer.getShort();
if (len == -1) {
return null;
} else if (len == 0) {
return "";
}
byte[] bs = new byte[len];
buffer.get(bs);
return new String(bs, StandardCharsets.UTF_8);
}
public static Map<String, String> getMap(ByteBuffer buffer) {
int len = buffer.getChar();
if (len == 0) {
return null;
private static void putSeriStringOrList(ByteBuffer buffer, Serializable value) {
if (value == null) {
buffer.putShort((short) -1);
} else if (value instanceof Collection) {
buffer.putShort((short) ((Collection) value).size());
for (Object val : (Collection) value) {
putBigString(buffer, val == null ? null : val.toString());
}
} else {
buffer.putShort((short) 0);
putBigString(buffer, value == null ? null : value.toString());
}
Map<String, String> map = new HashMap<>(len);
for (int i = 0; i < len; i++) {
map.put(getSmallString(buffer), getBigString(buffer));
}
return map;
}
private static Serializable getSeriStringOrList(ByteBuffer buffer) {
int size = buffer.getShort();
if (size == -1) {
return null;
} else if (size == 0) { //单个字符串
return getBigString(buffer);
}
ArrayList list = new ArrayList();
for (int i = 0; i < size; i++) {
list.add(getBigString(buffer));
}
return list;
}
private static int lengthSeriStringOrList(Serializable value) {
if (value == null) {
return 0;
} else if (value instanceof Collection) {
int c = 0;
for (Object val : (Collection) value) {
c += 4 + (val == null ? 0 : Utility.encodeUTF8Length(val.toString()));
}
return c;
} else {
return 4 + (value == null ? 0 : Utility.encodeUTF8Length(value.toString()));
}
}
}

16
src/main/java/org/redkale/net/AsyncGroup.java
View File

@@ -32,23 +32,27 @@ public abstract class AsyncGroup {
}
public CompletableFuture<AsyncConnection> createTCPClient(final SocketAddress address) {
return createTCPClient(address, 0, 0);
return createTCPClient(address, 0, 0, 0);
}
public abstract CompletableFuture<AsyncConnection> createTCPClient(final SocketAddress address, final int readTimeoutSeconds, final int writeTimeoutSeconds);
public abstract CompletableFuture<AsyncConnection> createTCPClient(final SocketAddress address,
final int connectTimeoutSeconds, final int readTimeoutSeconds, final int writeTimeoutSeconds);
public CompletableFuture<AsyncConnection> createUDPClient(final SocketAddress address) {
return createUDPClient(address, 0, 0);
return createUDPClient(address, 0, 0, 0);
}
public abstract CompletableFuture<AsyncConnection> createUDPClient(final SocketAddress address, final int readTimeoutSeconds, final int writeTimeoutSeconds);
public abstract CompletableFuture<AsyncConnection> createUDPClient(final SocketAddress address,
final int connectTimeoutSeconds, final int readTimeoutSeconds, final int writeTimeoutSeconds);
public CompletableFuture<AsyncConnection> createClient(final boolean tcp, final SocketAddress address) {
return tcp ? createTCPClient(address) : createUDPClient(address);
}
public CompletableFuture<AsyncConnection> createClient(final boolean tcp, final SocketAddress address, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
return tcp ? createTCPClient(address, readTimeoutSeconds, writeTimeoutSeconds) : createUDPClient(address, readTimeoutSeconds, writeTimeoutSeconds);
public CompletableFuture<AsyncConnection> createClient(final boolean tcp, final SocketAddress address,
final int connectTimeoutSeconds, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
return tcp ? createTCPClient(address, connectTimeoutSeconds, readTimeoutSeconds, writeTimeoutSeconds)
: createUDPClient(address, connectTimeoutSeconds, readTimeoutSeconds, writeTimeoutSeconds);
}
public abstract ScheduledFuture scheduleTimeout(Runnable callable, long delay, TimeUnit unit);

12
src/main/java/org/redkale/net/AsyncIOGroup.java
View File

@@ -229,7 +229,8 @@ public class AsyncIOGroup extends AsyncGroup {
}
@Override
public CompletableFuture<AsyncConnection> createTCPClient(final SocketAddress address, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
public CompletableFuture<AsyncConnection> createTCPClient(final SocketAddress address,
final int connectTimeoutSeconds, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
Objects.requireNonNull(address);
AsyncNioTcpConnection conn;
try {
@@ -237,7 +238,8 @@ public class AsyncIOGroup extends AsyncGroup {
} catch (IOException e) {
return CompletableFuture.failedFuture(e);
}
final CompletableFuture<AsyncConnection> future = new CompletableFuture<>();
int seconds = connectTimeoutSeconds > 0 ? connectTimeoutSeconds : 6;
final CompletableFuture future = Utility.orTimeout(new CompletableFuture(), seconds, TimeUnit.SECONDS);
conn.connect(address, null, new CompletionHandler<Void, Void>() {
@Override
public void completed(Void result, Void attachment) {
@@ -309,14 +311,16 @@ public class AsyncIOGroup extends AsyncGroup {
}
@Override
public CompletableFuture<AsyncConnection> createUDPClient(final SocketAddress address, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
public CompletableFuture<AsyncConnection> createUDPClient(final SocketAddress address,
final int connectTimeoutSeconds, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
AsyncNioUdpConnection conn;
try {
conn = newUDPClientConnection(address);
} catch (IOException e) {
return CompletableFuture.failedFuture(e);
}
CompletableFuture future = new CompletableFuture();
int seconds = connectTimeoutSeconds > 0 ? connectTimeoutSeconds : 6;
final CompletableFuture future = Utility.orTimeout(new CompletableFuture(), seconds, TimeUnit.SECONDS);
conn.connect(address, null, new CompletionHandler<Void, Void>() {
@Override
public void completed(Void result, Void attachment) {

10
src/main/java/org/redkale/net/Transport.java
View File

@@ -274,14 +274,14 @@ public final class Transport {
try {
if (!tcp) { // UDP
SocketAddress udpaddr = rand ? nodes[0].address : addr;
return asyncGroup.createUDPClient(udpaddr, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
return asyncGroup.createUDPClient(udpaddr, 6, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
}
if (!rand) { //指定地址
TransportNode node = findTransportNode(addr);
if (node == null) {
return asyncGroup.createTCPClient(addr, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
return asyncGroup.createTCPClient(addr, 6, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
}
return pollAsync(node, addr, () -> asyncGroup.createTCPClient(addr, factory.readTimeoutSeconds, factory.writeTimeoutSeconds));
return pollAsync(node, addr, () -> asyncGroup.createTCPClient(addr, 6, factory.readTimeoutSeconds, factory.writeTimeoutSeconds));
}
//---------------------随机取地址------------------------
@@ -308,7 +308,7 @@ public final class Transport {
}
}
return pollAsync(one, one.getAddress(), () -> {
return asyncGroup.createTCPClient(one.address, factory.readTimeoutSeconds, factory.writeTimeoutSeconds)
return asyncGroup.createTCPClient(one.address, 6, factory.readTimeoutSeconds, factory.writeTimeoutSeconds)
.whenComplete((c, t) -> {
one.disabletime = t == null ? 0 : System.currentTimeMillis();
});
@@ -330,7 +330,7 @@ public final class Transport {
if (future.isDone()) {
return future;
}
asyncGroup.createTCPClient(node.address, factory.readTimeoutSeconds, factory.writeTimeoutSeconds)
asyncGroup.createTCPClient(node.address, 6, factory.readTimeoutSeconds, factory.writeTimeoutSeconds)
.whenComplete((c, t) -> {
if (c != null && !future.complete(c)) {
node.connQueue.offer(c);

87
src/main/java/org/redkale/net/client/Client.java
View File

@@ -78,6 +78,8 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
protected int maxPipelines = DEFAULT_MAX_PIPELINES; //单个连接最大并行处理数
protected int connectTimeoutSeconds;
protected int readTimeoutSeconds;
protected int writeTimeoutSeconds;
@@ -212,7 +214,7 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
conn.dispose(null);
} else {
try {
conn.writeChannel(closeReq).get(1, TimeUnit.SECONDS);
conn.writeChannel(closeReq).get(3, TimeUnit.SECONDS);
} catch (Exception e) {
//do nothing
}
@@ -225,7 +227,7 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
return connect().thenCompose(conn -> writeChannel(conn, request));
return connect(request.workThread).thenCompose(conn -> writeChannel(conn, request));
}
public final <T> CompletableFuture<T> sendAsync(R request, Function<P, T> respTransfer) {
@@ -233,7 +235,7 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
return connect().thenCompose(conn -> writeChannel(conn, request, respTransfer));
return connect(request.workThread).thenCompose(conn -> writeChannel(conn, request, respTransfer));
}
public final CompletableFuture<P> sendAsync(SocketAddress addr, R request) {
@@ -241,7 +243,7 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
return connect(addr).thenCompose(conn -> writeChannel(conn, request));
return connect(request.workThread, addr).thenCompose(conn -> writeChannel(conn, request));
}
public final <T> CompletableFuture<T> sendAsync(SocketAddress addr, R request, Function<P, T> respTransfer) {
@@ -249,7 +251,7 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
return connect(addr).thenCompose(conn -> writeChannel(conn, request, respTransfer));
return connect(request.workThread, addr).thenCompose(conn -> writeChannel(conn, request, respTransfer));
}
protected CompletableFuture<P> writeChannel(ClientConnection conn, R request) {
@@ -261,43 +263,47 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
}
public final CompletableFuture<List<P>> sendAsync(R[] requests) {
requests[0].traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
String traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
for (R request : requests) {
request.traceid = traceid;
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
}
return connect().thenCompose(conn -> writeChannel(conn, requests));
return connect(requests[0].workThread).thenCompose(conn -> writeChannel(conn, requests));
}
public final <T> CompletableFuture<List<T>> sendAsync(R[] requests, Function<P, T> respTransfer) {
requests[0].traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
String traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
for (R request : requests) {
request.traceid = traceid;
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
}
return connect().thenCompose(conn -> writeChannel(conn, requests, respTransfer));
return connect(requests[0].workThread).thenCompose(conn -> writeChannel(conn, requests, respTransfer));
}
public final CompletableFuture<List<P>> sendAsync(SocketAddress addr, R[] requests) {
requests[0].traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
String traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
for (R request : requests) {
request.traceid = traceid;
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
}
return connect(addr).thenCompose(conn -> writeChannel(conn, requests));
return connect(requests[0].workThread, addr).thenCompose(conn -> writeChannel(conn, requests));
}
public final <T> CompletableFuture<List<T>> sendAsync(SocketAddress addr, R[] requests, Function<P, T> respTransfer) {
requests[0].traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
String traceid = Traces.computeIfAbsent(requests[0].traceid, Traces.currentTraceid());
for (R request : requests) {
request.traceid = traceid;
if (request.workThread == null) {
request.workThread = WorkThread.currentWorkThread();
}
}
return connect(addr).thenCompose(conn -> writeChannel(conn, requests, respTransfer));
return connect(requests[0].workThread, addr).thenCompose(conn -> writeChannel(conn, requests, respTransfer));
}
protected CompletableFuture<List<P>> writeChannelBatch(ClientConnection conn, R... requests) {
@@ -314,17 +320,20 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
}
public final CompletableFuture<C> connect() {
return connect(true);
return connect(WorkThread.currentWorkThread(), true);
}
public final CompletableFuture<C> newConnection() {
return connect(false);
return connect(WorkThread.currentWorkThread(), false);
}
private CompletableFuture<C> connect(final boolean pool) {
protected CompletableFuture<C> connect(WorkThread workThread) {
return connect(workThread, true);
}
private CompletableFuture<C> connect(final WorkThread workThread, final boolean pool) {
final String traceid = Traces.currentTraceid();
final int size = this.connArray.length;
WorkThread workThread = WorkThread.currentWorkThread();
final int connIndex = (workThread != null && workThread.threads() == size) ? workThread.index() : (int) Math.abs(connIndexSeq.getAndIncrement()) % size;
C cc = (C) this.connArray[connIndex];
if (pool && cc != null && cc.isOpen()) {
@@ -333,7 +342,7 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
long s = System.currentTimeMillis();
final Queue<CompletableFuture<C>> waitQueue = this.connAcquireWaitings[connIndex];
if (!pool || this.connOpenStates[connIndex].compareAndSet(false, true)) {
CompletableFuture<C> future = group.createClient(tcp, this.address.randomAddress(), readTimeoutSeconds, writeTimeoutSeconds)
CompletableFuture<C> future = group.createClient(tcp, this.address.randomAddress(), connectTimeoutSeconds, readTimeoutSeconds, writeTimeoutSeconds)
.thenApply(c -> {
Traces.currentTraceid(traceid);
C rs = (C) createClientConnection(connIndex, c).setMaxPipelines(maxPipelines);
@@ -370,41 +379,54 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
if (!f.isDone()) {
if (workThread != null) {
CompletableFuture<C> fs = f;
workThread.execute(() -> {
workThread.runWork(() -> {
Traces.currentTraceid(traceid);
fs.complete(c);
if (!fs.isDone()) {
fs.complete(c);
}
});
} else {
f.complete(c);
CompletableFuture<C> fs = f;
Utility.execute(() -> {
if (!fs.isDone()) {
fs.complete(c);
}
});
}
}
}
}
return c;
}).whenComplete((r, t) -> {
}
).whenComplete((r, t) -> {
if (pool && t != null) {
this.connOpenStates[connIndex].set(false);
}
});
} else {
CompletableFuture rs = Utility.orTimeout(new CompletableFuture(), readTimeoutSeconds, TimeUnit.SECONDS);
int seconds = connectTimeoutSeconds > 0 ? connectTimeoutSeconds : 6;
CompletableFuture rs = Utility.orTimeout(new CompletableFuture(), seconds, TimeUnit.SECONDS);
waitQueue.offer(rs);
return rs;
}
}
//指定地址获取连接
public final CompletableFuture<C> connect(final SocketAddress addr) {
return connect(true, addr);
return connect(WorkThread.currentWorkThread(), true, addr);
}
//指定地址获取连接
public final CompletableFuture<C> newConnection(final SocketAddress addr) {
return connect(false, addr);
return connect(WorkThread.currentWorkThread(), false, addr);
}
protected CompletableFuture<C> connect(WorkThread workThread, final SocketAddress addr) {
return connect(workThread, true, addr);
}
//指定地址获取连接
private CompletableFuture<C> connect(final boolean pool, final SocketAddress addr) {
private CompletableFuture<C> connect(final WorkThread workThread, final boolean pool, final SocketAddress addr) {
final String traceid = Traces.currentTraceid();
if (addr == null) {
return connect();
@@ -414,11 +436,10 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
if (pool && ec != null && ec.isOpen()) {
return CompletableFuture.completedFuture(ec);
}
WorkThread workThread = WorkThread.currentWorkThread();
final Queue<CompletableFuture<C>> waitQueue = entry.connAcquireWaitings;
if (!pool || entry.connOpenState.compareAndSet(false, true)) {
long s = System.currentTimeMillis();
CompletableFuture<C> future = group.createClient(tcp, addr, readTimeoutSeconds, writeTimeoutSeconds)
CompletableFuture<C> future = group.createClient(tcp, addr, connectTimeoutSeconds, readTimeoutSeconds, writeTimeoutSeconds)
.thenApply(c -> (C) createClientConnection(-1, c).setMaxPipelines(maxPipelines));
R virtualReq = createVirtualRequestAfterConnect();
if (virtualReq != null) {
@@ -450,12 +471,13 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
if (!f.isDone()) {
if (workThread != null) {
CompletableFuture<C> fs = f;
workThread.execute(() -> {
workThread.runWork(() -> {
Traces.currentTraceid(traceid);
fs.complete(c);
});
} else {
f.complete(c);
CompletableFuture<C> fs = f;
Utility.execute(() -> fs.complete(c));
}
}
}
@@ -467,7 +489,8 @@ public abstract class Client<C extends ClientConnection<R, P>, R extends ClientR
}
});
} else {
CompletableFuture rs = Utility.orTimeout(new CompletableFuture(), 6, TimeUnit.SECONDS);
int seconds = connectTimeoutSeconds > 0 ? connectTimeoutSeconds : 6;
CompletableFuture rs = Utility.orTimeout(new CompletableFuture(), seconds, TimeUnit.SECONDS);
waitQueue.offer(rs);
return rs;
}

2
src/main/java/org/redkale/net/client/ClientCodec.java
View File

@@ -117,7 +117,7 @@ public abstract class ClientCodec<R extends ClientRequest, P extends ClientResul
}
void responseComplete(boolean halfCompleted, ClientFuture<R, Object> respFuture, P message, Throwable exc) {
R request = respFuture.request;
final R request = respFuture.request;
Traces.currentTraceid(request.getTraceid());
AsyncIOThread readThread = connection.channel.getReadIOThread();
final WorkThread workThread = request.workThread == null ? readThread : request.workThread;

4
src/main/java/org/redkale/net/http/HttpFilter.java
View File

@@ -46,6 +46,10 @@ public abstract class HttpFilter extends Filter<HttpContext, HttpRequest, HttpRe
request.setHeader(name, value);
}
protected void addHeader(HttpRequest request, String name, String value) {
request.addHeader(name, value);
}
protected void removeParameter(HttpRequest request, String name) {
request.removeParameter(name);
}

375
src/main/java/org/redkale/net/http/HttpHeader.java Normal file
View File

@@ -0,0 +1,375 @@
/*
*
*/
package org.redkale.net.http;
import java.io.Serializable;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BiConsumer;
import org.redkale.convert.TextConvert;
import org.redkale.convert.json.JsonConvert;
import org.redkale.util.RedkaleException;
/**
* Http Header Object
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
* @since 2.8.0
*/
public class HttpHeader implements RestHeaders, Serializable {
//value值只能是String、List<String>
protected LinkedHashMap<String, Serializable> map;
protected HttpHeader() {
}
public static HttpHeader create() {
return new HttpHeader();
}
public static HttpHeader of(String... items) {
HttpHeader header = new HttpHeader();
int len = items.length / 2;
for (int i = 0; i < len; i++) {
header.add(items[i * 2], items[i * 2 + 1]);
}
return header;
}
/**
* 无需校验参数合法性
*
* @param map 参数
*
* @return HttpHeader
*/
public static HttpHeader ofValid(Map<String, Serializable> map) {
HttpHeader header = new HttpHeader();
if (map != null) {
header.map = map instanceof LinkedHashMap ? (LinkedHashMap) map : new LinkedHashMap(map);
}
return header;
}
@Override
public String firstValue(String name) {
return firstValue(name, null);
}
@Override
public String firstValue(String name, String defaultValue) {
if (map == null) {
return defaultValue;
}
Serializable val = map.get(name);
if (val == null) {
return defaultValue;
}
if (val instanceof Collection) {
for (Object item : (Collection) val) {
return String.valueOf(item); //return fisrt value
}
return defaultValue;
}
return String.valueOf(val);
}
@Override
public List<String> listValue(String name) {
if (this.map == null) {
return null;
}
Serializable val = this.map.get(name);
if (val == null) {
return null;
}
if (val instanceof Collection) {
return new ArrayList<>((Collection) val);
}
List list = new ArrayList<>();
list.add(val);
return list;
}
@Override
public void forEach(BiConsumer<String, String> consumer) {
if (map != null) {
map.forEach((k, v) -> {
if (v instanceof Collection) {
for (Object item : (Collection) v) {
consumer.accept(k, item == null ? null : item.toString());
}
} else {
consumer.accept(k, v == null ? null : v.toString());
}
});
}
}
@Override
public String[] names() {
if (this.map == null) {
return new String[0];
}
Set<String> names = this.map.keySet();
return names.toArray(new String[names.size()]);
}
@Override
public boolean contains(String key) {
return this.map != null && this.map.containsKey(key);
}
public HttpHeader addAll(HttpHeader header) {
if (header.map != null) {
if (this.map == null) {
this.map = new LinkedHashMap<>(header.map);
} else {
header.forEach(this::add);
}
}
return this;
}
public HttpHeader add(Map<String, String> values) {
if (values != null) {
values.forEach(this::add);
}
return this;
}
//服务端接收，无需校验参数合法性
void addValid(String key, String value) {
if (this.map == null) {
this.map = new LinkedHashMap<>();
this.map.put(key, value);
} else {
Serializable old = this.map.get(key);
if (old == null) {
this.map.put(key, value);
} else if (old instanceof Collection) {
((Collection) old).add(value);
} else {
ArrayList list = new ArrayList();
list.add(old);
list.add(value);
this.map.put(key, list);
}
}
}
public HttpHeader add(String key, String value) {
check(key, value);
if (this.map == null) {
this.map = new LinkedHashMap<>();
this.map.put(key, value);
} else {
Serializable old = this.map.get(key);
if (old == null) {
this.map.put(key, value);
} else if (old instanceof Collection) {
((Collection) old).add(value);
} else {
ArrayList list = new ArrayList();
list.add(old);
list.add(value);
this.map.put(key, list);
}
}
return this;
}
public HttpHeader add(String key, List<String> value) {
if (value.isEmpty()) {
return this;
}
for (String val : value) {
check(key, val);
}
if (this.map == null) {
this.map = new LinkedHashMap<>();
this.map.put(key, new ArrayList(value));
} else {
Serializable old = this.map.get(key);
if (old == null) {
this.map.put(key, new ArrayList(value));
} else if (old instanceof Collection) {
((Collection) old).addAll(value);
} else {
ArrayList list = new ArrayList();
list.add(old);
list.addAll(value);
this.map.put(key, list);
}
}
return this;
}
public HttpHeader add(String key, TextConvert convert, Object value) {
return add(key, (convert == null ? JsonConvert.root() : convert).convertTo(value));
}
public HttpHeader add(String key, Object value) {
return add(key, JsonConvert.root().convertTo(value));
}
public HttpHeader add(String key, boolean value) {
return add(key, String.valueOf(value));
}
public HttpHeader add(String key, short value) {
return add(key, String.valueOf(value));
}
public HttpHeader add(String key, int value) {
return add(key, String.valueOf(value));
}
public HttpHeader add(String key, float value) {
return add(key, String.valueOf(value));
}
public HttpHeader add(String key, long value) {
return add(key, String.valueOf(value));
}
public HttpHeader add(String key, double value) {
return add(key, String.valueOf(value));
}
public HttpHeader add(String key, BigInteger value) {
return add(key, String.valueOf(value));
}
public HttpHeader setAll(HttpHeader header) {
if (header.map != null) {
if (this.map == null) {
this.map = new LinkedHashMap<>();
}
this.map.putAll(header.map);
}
return this;
}
public HttpHeader set(Map<String, String> values) {
if (values != null) {
values.forEach(this::set);
}
return this;
}
//服务端接收，无需校验参数合法性
void setValid(String key, String value) {
if (this.map == null) {
this.map = new LinkedHashMap<>();
}
this.map.put(key, value);
}
public HttpHeader set(String key, String value) {
check(key, value);
if (this.map == null) {
this.map = new LinkedHashMap<>();
}
this.map.put(key, value);
return this;
}
public HttpHeader set(String key, List<String> value) {
if (value.isEmpty()) {
return this;
}
for (String val : value) {
check(key, val);
}
if (this.map == null) {
this.map = new LinkedHashMap<>();
}
this.map.put(key, new ArrayList(value));
return this;
}
public HttpHeader set(String key, TextConvert convert, Object value) {
return set(key, (convert == null ? JsonConvert.root() : convert).convertTo(value));
}
public HttpHeader set(String key, Object value) {
return set(key, JsonConvert.root().convertTo(value));
}
public HttpHeader set(String key, boolean value) {
return set(key, String.valueOf(value));
}
public HttpHeader set(String key, short value) {
return set(key, String.valueOf(value));
}
public HttpHeader set(String key, int value) {
return set(key, String.valueOf(value));
}
public HttpHeader set(String key, float value) {
return set(key, String.valueOf(value));
}
public HttpHeader set(String key, long value) {
return set(key, String.valueOf(value));
}
public HttpHeader set(String key, double value) {
return set(key, String.valueOf(value));
}
public HttpHeader set(String key, BigInteger value) {
return set(key, String.valueOf(value));
}
public HttpHeader remove(String key) {
if (this.map != null) {
this.map.remove(key);
}
return this;
}
@Override
public Map<String, Serializable> map() {
return this.map;
}
public boolean isEmpty() {
return this.map == null || this.map.isEmpty();
}
public HttpHeader clear() {
if (this.map != null) {
this.map.clear();
}
return this;
}
protected String check(String key, String value) {
if (key.indexOf('\r') >= 0 || key.indexOf('\n') >= 0) {
throw new RedkaleException("http-header name(name = " + key + ") is illegal");
}
if (value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0) {
throw new RedkaleException("http-header value(name = " + key + ", value = " + value + ") is illegal");
}
return value;
}
@Override
public String toString() {
return String.valueOf(this.map);
}
}

173
src/main/java/org/redkale/net/http/HttpRequest.java
View File

@@ -19,6 +19,7 @@ import org.redkale.convert.*;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.Request;
import org.redkale.util.*;
import static org.redkale.util.Utility.isEmpty;
/**
* Http请求包 与javax.servlet.http.HttpServletRequest 基本类似。 <br>
@@ -121,9 +122,9 @@ public class HttpRequest extends Request<HttpContext> {
protected Convert respConvert;
protected final Map<String, String> headers = new HashMap<>();
//---------- header 相关参数 结束 ----------
protected final HttpHeader headers = HttpHeader.create();
//---------- header 相关参数 结束 ----------
@Comment("Method GET/POST/...")
protected String method;
@@ -207,7 +208,7 @@ public class HttpRequest extends Request<HttpContext> {
this.array.put(req.getBody());
}
if (req.getHeaders() != null) {
this.headers.putAll(req.getHeaders());
this.headers.setAll(req.getHeaders());
}
this.reqConvertType = req.getReqConvertType();
this.reqConvert = req.getReqConvertType() == null ? null : ConvertFactory.findConvert(req.getReqConvertType());
@@ -239,11 +240,11 @@ public class HttpRequest extends Request<HttpContext> {
HttpSimpleRequest req = new HttpSimpleRequest();
req.setBody(array.length() == 0 ? null : array.getBytes());
if (!getHeaders().isEmpty()) {
req.setHeaders(new HashMap<>(headers));
if (headers.containsKey(Rest.REST_HEADER_RPC)) { //外部request不能包含RPC的header信息
req.setHeaders(headers);
if (headers.contains(Rest.REST_HEADER_RPC)) { //外部request不能包含RPC的header信息
req.removeHeader(Rest.REST_HEADER_RPC);
}
if (headers.containsKey(Rest.REST_HEADER_CURRUSERID)) { //外部request不能包含RPC的header信息
if (headers.contains(Rest.REST_HEADER_CURRUSERID)) { //外部request不能包含RPC的header信息
req.removeHeader(Rest.REST_HEADER_CURRUSERID);
}
}
@@ -327,7 +328,7 @@ public class HttpRequest extends Request<HttpContext> {
this.headerHalfLen = httplast.headerHalfLen;
this.headerBytes = httplast.headerBytes;
this.headerParsed = httplast.headerParsed;
this.headers.putAll(httplast.headers);
this.headers.setAll(httplast.headers);
} else if (context.lazyHeaders && getmethod) { //非GET必须要读header，会有Content-Length
int rs = loadHeaderBytes(buffer);
if (rs != 0) {
@@ -789,49 +790,49 @@ public class HttpRequest extends Request<HttpContext> {
} else {
value = "";
}
headers.put(HEAD_CONNECTION, value);
headers.setValid(HEAD_CONNECTION, value);
break;
case HEAD_UPGRADE: //Upgrade
this.maybews = vlen == 9 && content[0] == 'w' && content[1] == 'e' && content[2] == 'b' && content[3] == 's'
&& content[4] == 'o' && content[5] == 'c' && content[6] == 'k' && content[7] == 'e' && content[8] == 't';
headers.put(HEAD_UPGRADE, this.maybews ? "websocket" : bytes.toString(true, charset));
headers.setValid(HEAD_UPGRADE, this.maybews ? "websocket" : bytes.toString(true, charset));
break;
case HEAD_EXPECT: //Expect
this.expect = vlen == 12 && content[0] == '1' && content[1] == '0' && content[2] == '0' && content[3] == '-'
&& content[4] == 'c' && content[5] == 'o' && content[6] == 'n' && content[7] == 't' && content[8] == 'i'
&& content[9] == 'n' && content[10] == 'u' && content[11] == 'e';
headers.put(HEAD_EXPECT, this.expect ? "100-continue" : bytes.toString(true, charset));
headers.setValid(HEAD_EXPECT, this.expect ? "100-continue" : bytes.toString(true, charset));
break;
case Rest.REST_HEADER_RPC: //rest-rpc
this.rpc = vlen == 4 && content[0] == 't' && content[1] == 'r' && content[2] == 'u' && content[3] == 'e';
headers.put(name, this.rpc ? "true"
headers.setValid(name, this.rpc ? "true"
: (vlen == 5 && content[0] == 'f' && content[1] == 'a' && content[2] == 'l' && content[3] == 's' && content[4] == 'e'
? "false" : bytes.toString(true, charset)));
break;
case Rest.REST_HEADER_TRACEID: //rest-traceid
value = bytes.toString(true, charset);
this.traceid = value;
headers.put(name, value);
headers.setValid(name, value);
break;
case Rest.REST_HEADER_CURRUSERID: //rest-curruserid
value = bytes.toString(true, charset);
this.currentUserid = value;
headers.put(name, value);
headers.setValid(name, value);
break;
case Rest.REST_HEADER_REQ_CONVERT: //rest-req-convert-type
value = bytes.toString(true, charset);
reqConvertType = ConvertType.valueOf(value);
reqConvert = ConvertFactory.findConvert(reqConvertType);
headers.put(name, value);
headers.setValid(name, value);
break;
case Rest.REST_HEADER_RESP_CONVERT: //rest-resp-convert-type
value = bytes.toString(true, charset);
respConvertType = ConvertType.valueOf(value);
respConvert = ConvertFactory.findConvert(respConvertType);
headers.put(name, value);
headers.setValid(name, value);
break;
default:
headers.put(name, bytes.toString(charset));
headers.addValid(name, bytes.toString(charset));
}
}
}
@@ -941,7 +942,7 @@ public class HttpRequest extends Request<HttpContext> {
req.reqConvert = this.reqConvert;
req.respConvert = this.respConvert;
req.respConvertType = this.respConvertType;
req.headers.putAll(this.headers);
req.headers.setAll(this.headers);
return req;
}
@@ -1065,7 +1066,12 @@ public class HttpRequest extends Request<HttpContext> {
}
protected HttpRequest setHeader(String name, String value) {
this.headers.put(name, value);
this.headers.setValid(name, value);
return this;
}
protected HttpRequest addHeader(String name, String value) {
this.headers.add(name, value);
return this;
}
@@ -1541,16 +1547,25 @@ public class HttpRequest extends Request<HttpContext> {
+ (this.array.length() > 0 ? (", \r\n bodyLength: " + this.array.length()) : "")
+ (this.boundary || this.array.isEmpty() ? "" : (", \r\n bodyContent: " + (this.respConvertType == null || this.respConvertType == ConvertType.JSON ? this.getBodyUTF8() : Arrays.toString(getBody()))))
+ ", \r\n params: " + toMapString(this.params, 4)
+ ", \r\n header: " + toMapString(this.headers, 4)
+ ", \r\n header: " + toMapString(this.headers.map, 4)
+ "\r\n}"; //this.headers.toString(4)
}
private static CharSequence toMapString(Map<String, String> map, int indent) {
private static CharSequence toMapString(Map<String, ?> map, int indent) {
final String space = " ".repeat(indent);
StringBuilder sb = new StringBuilder();
sb.append("{\r\n");
for (Map.Entry en : map.entrySet()) {
sb.append(space).append(" '").append(en.getKey()).append("': '").append(en.getValue()).append("',\r\n");
if (map != null) {
for (Map.Entry en : map.entrySet()) {
Object val = en.getValue();
if (val instanceof Collection) {
for (Object item : (Collection) val) {
sb.append(space).append(" '").append(en.getKey()).append("': '").append(item).append("',\r\n");
}
} else {
sb.append(space).append(" '").append(en.getKey()).append("': '").append(val).append("',\r\n");
}
}
}
sb.append(space).append('}');
return sb;
@@ -2146,29 +2161,12 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return AnyValue
*/
public Map<String, String> getHeaders() {
@AsmDepends
public HttpHeader getHeaders() {
parseHeader();
return headers;
}
/**
* 将请求Header转换成Map
*
* @param map Map
*
* @return Map
*/
@ConvertDisabled
public Map<String, String> getHeadersToMap(Map<String, String> map) {
parseHeader();
if (map == null) {
map = new LinkedHashMap<>();
}
final Map<String, String> map0 = map;
headers.forEach((k, v) -> map0.put(k, v));
return map0;
}
/**
* 获取所有的header名
*
@@ -2177,8 +2175,7 @@ public class HttpRequest extends Request<HttpContext> {
@ConvertDisabled
public String[] getHeaderNames() {
parseHeader();
Set<String> names = headers.keySet();
return names.toArray(new String[names.size()]);
return headers.names();
}
/**
@@ -2189,8 +2186,7 @@ public class HttpRequest extends Request<HttpContext> {
* @return header值
*/
public String getHeader(String name) {
parseHeader();
return headers.get(name);
return getHeader(name, null);
}
/**
@@ -2201,9 +2197,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public String getHeader(String name, String defaultValue) {
parseHeader();
return headers.getOrDefault(name, defaultValue);
return headers.firstValue(name, defaultValue);
}
/**
@@ -2215,9 +2212,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public <T> T getJsonHeader(java.lang.reflect.Type type, String name) {
String v = getHeader(name);
return v == null || v.isEmpty() ? null : jsonConvert.convertFrom(type, v);
return isEmpty(v) ? null : jsonConvert.convertFrom(type, v);
}
/**
@@ -2230,9 +2228,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public <T> T getJsonHeader(JsonConvert convert, java.lang.reflect.Type type, String name) {
String v = getHeader(name);
return v == null || v.isEmpty() ? null : convert.convertFrom(type, v);
return isEmpty(v) ? null : convert.convertFrom(type, v);
}
/**
@@ -2243,11 +2242,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public boolean getBooleanHeader(String name, boolean defaultValue) {
//return headers.getBoolValue(name, defaultValue);
parseHeader();
String value = headers.get(name);
return value == null || value.length() == 0 ? defaultValue : Boolean.parseBoolean(value);
String value = getHeader(name);
return isEmpty(value) ? defaultValue : Boolean.parseBoolean(value);
}
/**
@@ -2258,11 +2256,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public short getShortHeader(String name, short defaultValue) {
//return headers.getShortValue(name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2281,11 +2278,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public short getShortHeader(int radix, String name, short defaultValue) {
//return headers.getShortValue(name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2303,11 +2299,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public short getShortHeader(String name, int defaultValue) {
//return headers.getShortValue(name, (short) defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return (short) defaultValue;
}
try {
@@ -2326,11 +2321,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public short getShortHeader(int radix, String name, int defaultValue) {
//return headers.getShortValue(radix, name, (short) defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return (short) defaultValue;
}
try {
@@ -2348,11 +2342,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public int getIntHeader(String name, int defaultValue) {
//return headers.getIntValue(name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2371,11 +2364,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public int getIntHeader(int radix, String name, int defaultValue) {
//return headers.getIntValue(radix, name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2393,11 +2385,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public long getLongHeader(String name, long defaultValue) {
//return headers.getLongValue(name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2416,11 +2407,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public long getLongHeader(int radix, String name, long defaultValue) {
//return headers.getLongValue(radix, name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2438,11 +2428,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public float getFloatHeader(String name, float defaultValue) {
//return headers.getFloatValue(name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2460,11 +2449,10 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return header值
*/
@AsmDepends
public double getDoubleHeader(String name, double defaultValue) {
//return headers.getDoubleValue(name, defaultValue);
parseHeader();
String value = headers.get(name);
if (value == null || value.length() == 0) {
String value = getHeader(name);
if (isEmpty(value)) {
return defaultValue;
}
try {
@@ -2480,6 +2468,7 @@ public class HttpRequest extends Request<HttpContext> {
*
* @return AnyValue
*/
@AsmDepends
public Map<String, String> getParameters() {
parseBody();
return params;

12
src/main/java/org/redkale/net/http/HttpResult.java
View File

@@ -14,6 +14,7 @@ import java.util.concurrent.CompletableFuture;
import org.redkale.convert.*;
import org.redkale.convert.json.*;
import org.redkale.util.Creator;
import org.redkale.util.RedkaleException;
/**
*
@@ -33,7 +34,7 @@ public class HttpResult<T> {
@ConvertColumn(index = 2)
protected String contentType;
@ConvertColumn(index = 3)
@ConvertColumn(index = 3) //不使用HttpHeader因不易反序列化
protected Map<String, String> headers;
@ConvertColumn(index = 4)
@@ -63,10 +64,17 @@ public class HttpResult<T> {
}
public HttpResult<T> header(String name, Serializable value) {
if (name.indexOf('\r') >= 0 || name.indexOf('\n') >= 0) {
throw new RedkaleException("http-header name(name = " + name + ") is illegal");
}
String val = String.valueOf(value);
if (val.indexOf('\r') >= 0 || val.indexOf('\n') >= 0) {
throw new RedkaleException("http-header value(name = " + name + ", value = " + val + ") is illegal");
}
if (this.headers == null) {
this.headers = new HashMap<>();
}
this.headers.put(name, String.valueOf(value));
this.headers.put(name, val);
return this;
}

90
src/main/java/org/redkale/net/http/HttpSimpleClient.java
View File

@@ -10,7 +10,6 @@ import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import java.nio.charset.*;
import java.util.*;
import java.util.concurrent.*;
import org.redkale.convert.Convert;
import org.redkale.convert.json.JsonConvert;
@@ -47,14 +46,13 @@ public class HttpSimpleClient extends Client<HttpSimpleConnection, HttpSimpleReq
protected ExecutorService workExecutor;
protected int readTimeoutSeconds = 6;
protected int writeTimeoutSeconds = 6;
protected HttpSimpleClient(ExecutorService workExecutor, AsyncGroup asyncGroup) {
super("Redkale-http-client", asyncGroup, new ClientAddress(new InetSocketAddress("127.0.0.1", 0)));
this.workExecutor = workExecutor;
this.asyncGroup = asyncGroup;
this.connectTimeoutSeconds = 6;
this.readTimeoutSeconds = 6;
this.writeTimeoutSeconds = 6;
}
public static HttpSimpleClient create(ExecutorService workExecutor, AsyncGroup asyncGroup) {
@@ -137,23 +135,23 @@ public class HttpSimpleClient extends Client<HttpSimpleConnection, HttpSimpleReq
return sendAsync("GET", url, null, body, convert, valueType);
}
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, Map<String, String> headers) {
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, HttpHeader headers) {
return sendAsync("GET", url, headers, (byte[]) null);
}
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, Map<String, String> headers, Type valueType) {
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, HttpHeader headers, Type valueType) {
return sendAsync("GET", url, headers, null, (Convert) null, valueType);
}
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, Map<String, String> headers, Convert convert, Type valueType) {
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, HttpHeader headers, Convert convert, Type valueType) {
return sendAsync("GET", url, headers, null, convert, valueType);
}
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, Map<String, String> headers, String body) {
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, HttpHeader headers, String body) {
return sendAsync("GET", url, headers, body == null ? null : body.getBytes(StandardCharsets.UTF_8));
}
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, Map<String, String> headers, byte[] body) {
public CompletableFuture<HttpResult<byte[]>> getAsync(String url, HttpHeader headers, byte[] body) {
return sendAsync("GET", url, headers, body);
}
@@ -193,63 +191,74 @@ public class HttpSimpleClient extends Client<HttpSimpleConnection, HttpSimpleReq
return sendAsync("POST", url, null, body, convert, valueType);
}
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, Map<String, String> headers) {
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, HttpHeader headers) {
return sendAsync("POST", url, headers, (byte[]) null);
}
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, Map<String, String> headers, Type valueType) {
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, HttpHeader headers, Type valueType) {
return sendAsync("POST", url, headers, null, (Convert) null, valueType);
}
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, Map<String, String> headers, Convert convert, Type valueType) {
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, HttpHeader headers, Convert convert, Type valueType) {
return sendAsync("POST", url, headers, null, convert, valueType);
}
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, Map<String, String> headers, String body) {
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, HttpHeader headers, String body) {
return sendAsync("POST", url, headers, body == null ? null : body.getBytes(StandardCharsets.UTF_8));
}
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, Map<String, String> headers, byte[] body) {
public CompletableFuture<HttpResult<byte[]>> postAsync(String url, HttpHeader headers, byte[] body) {
return sendAsync("POST", url, headers, body);
}
public CompletableFuture<HttpResult<byte[]>> sendAsync(String method, String url, Map<String, String> headers, byte[] body) {
public CompletableFuture<HttpResult<byte[]>> sendAsync(String url, HttpSimpleRequest req) {
return sendAsync(req.getMethod(), url, req.getHeaders(), req.getBody(), (Convert) null, null);
}
public <T> CompletableFuture<HttpResult<T>> sendAsync(String url, HttpSimpleRequest req, Type valueType) {
return sendAsync(req.getMethod(), url, req.getHeaders(), req.getBody(), (Convert) null, null);
}
public CompletableFuture<HttpResult<byte[]>> sendAsync(String method, String url, HttpHeader headers, byte[] body) {
return sendAsync(method, url, headers, body, (Convert) null, null);
}
public CompletableFuture<HttpResult<byte[]>> sendAsync(String method, String url, Map<String, String> headers, byte[] body, Type valueType) {
public <T> CompletableFuture<HttpResult<T>> sendAsync(String method, String url, HttpHeader headers, byte[] body, Type valueType) {
return sendAsync(method, url, headers, body, (Convert) null, valueType);
}
public <T> CompletableFuture<HttpResult<T>> sendAsync(String method, String url, Map<String, String> headers, byte[] body, Convert convert, Type valueType) {
public <T> CompletableFuture<HttpResult<T>> sendAsync(String method, String url, HttpHeader headers, byte[] body, Convert convert, Type valueType) {
final String traceid = Traces.computeIfAbsent(Traces.currentTraceid());
final WorkThread workThread = WorkThread.currentWorkThread();
if (url.indexOf(' ') >= 0 || url.indexOf('\r') >= 0 || url.indexOf('\n') >= 0) {
throw new RedkaleException("http-url(" + url + ") is illegal");
}
final URI uri = URI.create(url);
final String host = uri.getHost();
final ByteArray array = new ByteArray();
int urlpos = url.indexOf("/", url.indexOf("//") + 3);
array.put((method.toUpperCase() + " " + (urlpos > 0 ? url.substring(urlpos) : "/") + " HTTP/1.1\r\n"
+ "Host: " + uri.getHost() + "\r\n"
+ Rest.REST_HEADER_TRACEID + ": " + traceid + "\r\n"
+ "Content-Length: " + (body == null ? 0 : body.length) + "\r\n").getBytes(StandardCharsets.UTF_8));
if (headers == null || !headers.contains("User-Agent")) {
array.put(header_bytes_useragent);
}
if (headers == null || !headers.contains("Connection")) {
array.put(header_bytes_connclose);
}
if (headers != null) {
headers.forEach((k, v) -> array.put((k + ": " + String.valueOf(v) + "\r\n").getBytes(StandardCharsets.UTF_8)));
}
array.put((byte) '\r', (byte) '\n');
if (body != null) {
array.put(body);
}
final int port = uri.getPort() > 0 ? uri.getPort() : (url.startsWith("https:") ? 443 : 80);
return createConnection(host, port).thenCompose(conn -> {
Traces.currentTraceid(traceid);
final ByteArray array = new ByteArray();
int urlpos = url.indexOf("/", url.indexOf("//") + 3);
array.put((method.toUpperCase() + " " + (urlpos > 0 ? url.substring(urlpos) : "/") + " HTTP/1.1\r\n"
+ "Host: " + uri.getHost() + "\r\n"
+ Rest.REST_HEADER_TRACEID + ": " + traceid + "\r\n"
+ "Content-Length: " + (body == null ? 0 : body.length) + "\r\n").getBytes(StandardCharsets.UTF_8));
if (headers == null || !headers.containsKey("User-Agent")) {
array.put(header_bytes_useragent);
}
if (headers == null || !headers.containsKey("Connection")) {
array.put(header_bytes_connclose);
}
if (headers != null) {
headers.forEach((k, v) -> {
array.put((k + ": " + v + "\r\n").getBytes(StandardCharsets.UTF_8));
});
}
array.put((byte) '\r', (byte) '\n');
if (body != null) {
array.put(body);
}
final CompletableFuture<HttpResult<T>> future = new CompletableFuture();
conn.write(array, new CompletionHandler<Integer, Void>() {
@Override
@@ -279,7 +288,8 @@ public class HttpSimpleClient extends Client<HttpSimpleConnection, HttpSimpleReq
}
protected CompletableFuture<HttpConnection> createConnection(String host, int port) {
return asyncGroup.createTCPClient(new InetSocketAddress(host, port), readTimeoutSeconds, writeTimeoutSeconds).thenApply(conn -> new HttpConnection(conn));
return asyncGroup.createTCPClient(new InetSocketAddress(host, port), connectTimeoutSeconds, readTimeoutSeconds, writeTimeoutSeconds)
.thenApply(conn -> new HttpConnection(conn));
}
//

101
src/main/java/org/redkale/net/http/HttpSimpleRequest.java
View File

@@ -16,6 +16,7 @@ import org.redkale.convert.json.JsonConvert;
import org.redkale.net.client.ClientConnection;
import org.redkale.net.client.ClientRequest;
import org.redkale.util.ByteArray;
import org.redkale.util.RedkaleException;
import org.redkale.util.Traces;
/**
@@ -79,7 +80,7 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
@ConvertColumn(index = 14)
@Comment("http header信息")
protected Map<String, String> headers;
protected HttpHeader headers;
@ConvertColumn(index = 15)
@Comment("参数信息")
@@ -102,8 +103,8 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
return req;
}
public static HttpSimpleRequest create(String requestURI, Map<String, String> headers) {
return new HttpSimpleRequest().requestURI(requestURI).method("POST").headers(headers).traceid(Traces.currentTraceid());
public static HttpSimpleRequest create(String requestURI, HttpHeader header) {
return new HttpSimpleRequest().requestURI(requestURI).method("POST").headers(header).traceid(Traces.currentTraceid());
}
@Override
@@ -140,7 +141,7 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
}
public HttpSimpleRequest formUrlencoded() {
this.headers.put("Content-Type", "x-www-form-urlencoded");
this.headers.set("Content-Type", "x-www-form-urlencoded");
return this;
}
@@ -150,16 +151,27 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
}
public HttpSimpleRequest traceid(String traceid) {
if (traceid != null) {
if (traceid.indexOf('\r') >= 0 || traceid.indexOf('\n') >= 0) {
throw new RedkaleException("http-traceid(" + traceid + ") is illegal");
}
}
this.traceid = traceid;
return this;
}
public HttpSimpleRequest requestURI(String requestURI) {
if (requestURI.indexOf(' ') >= 0 || requestURI.indexOf('\r') >= 0 || requestURI.indexOf('\n') >= 0) {
throw new RedkaleException("http-uri(" + requestURI + ") is illegal");
}
this.requestURI = requestURI;
return this;
}
public HttpSimpleRequest path(String path) {
if (path.indexOf(' ') >= 0 || path.indexOf('\r') >= 0 || path.indexOf('\n') >= 0) {
throw new RedkaleException("http-path(" + path + ") is illegal");
}
this.path = path;
return this;
}
@@ -219,8 +231,8 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
return this;
}
public HttpSimpleRequest headers(Map<String, String> headers) {
this.headers = headers;
public HttpSimpleRequest headers(HttpHeader header) {
this.headers = header;
return this;
}
@@ -234,53 +246,52 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
return this;
}
public HttpSimpleRequest header(String key, String value) {
public HttpSimpleRequest addHeader(String key, String value) {
if (this.headers == null) {
this.headers = new HashMap<>();
this.headers = HttpHeader.create();
}
this.headers.put(key, value);
this.headers.add(key, value);
return this;
}
public HttpSimpleRequest header(String key, TextConvert convert, Object value) {
if (value == null) {
return this;
}
public HttpSimpleRequest addHeader(String key, TextConvert convert, Object value) {
return addHeader(key, (convert == null ? JsonConvert.root() : convert).convertTo(value));
}
public HttpSimpleRequest addHeader(String key, Object value) {
return addHeader(key, JsonConvert.root().convertTo(value));
}
public HttpSimpleRequest addHeader(String key, int value) {
return addHeader(key, String.valueOf(value));
}
public HttpSimpleRequest addHeader(String key, long value) {
return addHeader(key, String.valueOf(value));
}
public HttpSimpleRequest setHeader(String key, String value) {
if (this.headers == null) {
this.headers = new HashMap<>();
this.headers = HttpHeader.create();
}
if (convert == null) {
convert = JsonConvert.root();
}
this.headers.put(key, convert.convertTo(value));
this.headers.set(key, value);
return this;
}
public HttpSimpleRequest header(String key, Object value) {
if (value == null) {
return this;
}
if (this.headers == null) {
this.headers = new HashMap<>();
}
this.headers.put(key, JsonConvert.root().convertTo(value));
return this;
public HttpSimpleRequest setHeader(String key, TextConvert convert, Object value) {
return setHeader(key, (convert == null ? JsonConvert.root() : convert).convertTo(value));
}
public HttpSimpleRequest header(String key, int value) {
if (this.headers == null) {
this.headers = new HashMap<>();
}
this.headers.put(key, String.valueOf(value));
return this;
public HttpSimpleRequest setHeader(String key, Object value) {
return setHeader(key, JsonConvert.root().convertTo(value));
}
public HttpSimpleRequest header(String key, long value) {
if (this.headers == null) {
this.headers = new HashMap<>();
}
this.headers.put(key, String.valueOf(value));
return this;
public HttpSimpleRequest setHeader(String key, int value) {
return setHeader(key, String.valueOf(value));
}
public HttpSimpleRequest setHeader(String key, long value) {
return setHeader(key, String.valueOf(value));
}
public HttpSimpleRequest param(String key, String value) {
@@ -352,11 +363,11 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
}
public String getHeader(String name) {
return headers == null ? null : headers.get(name);
return getHeader(name, null);
}
public String getHeader(String name, String defaultValue) {
return headers == null ? defaultValue : headers.getOrDefault(name, defaultValue);
return headers.firstValue(name, defaultValue);
}
public boolean isRpc() {
@@ -439,12 +450,12 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
this.contentType = contentType;
}
public Map<String, String> getHeaders() {
public HttpHeader getHeaders() {
return headers;
}
public void setHeaders(Map<String, String> headers) {
this.headers = headers;
public void setHeaders(HttpHeader headers) {
headers(headers);
}
public Map<String, String> getParams() {
@@ -452,7 +463,7 @@ public class HttpSimpleRequest extends ClientRequest implements java.io.Serializ
}
public void setParams(Map<String, String> params) {
this.params = params;
params(params);
}
public byte[] getBody() {

92
src/main/java/org/redkale/net/http/Rest.java
View File

@@ -70,6 +70,9 @@ public final class Rest {
private static final String REST_RETURNTYPES_FIELD_NAME = "_redkale_returntypes"; //存在泛型的结果数组
private static final java.lang.reflect.Type TYPE_MAP_STRING_SERIALIZE = new TypeToken<Map<String, Serializable>>() {
}.getType();
private static final java.lang.reflect.Type TYPE_MAP_STRING_STRING = new TypeToken<Map<String, String>>() {
}.getType();
@@ -1099,6 +1102,7 @@ public final class Rest {
final String httpResultDesc = Type.getDescriptor(HttpResult.class);
final String httpScopeDesc = Type.getDescriptor(HttpScope.class);
final String stageDesc = Type.getDescriptor(CompletionStage.class);
final String httpHeaderDesc = Type.getDescriptor(HttpHeader.class);
final String flipperDesc = Type.getDescriptor(Flipper.class);
final String httpServletName = HttpServlet.class.getName().replace('.', '/');
final String actionEntryName = HttpServlet.ActionEntry.class.getName().replace('.', '/');
@@ -1308,9 +1312,10 @@ public final class Rest {
if (userid != null) {
comment = "";
}
RestHeaders annheaders = param.getAnnotation(RestHeaders.class);
if (annheaders != null) {
boolean annheaders = param.getType() == RestHeaders.class;
if (annheaders) {
comment = "";
n = "^"; //Http头信息类型特殊处理
}
RestParams annparams = param.getAnnotation(RestParams.class);
if (annparams != null) {
@@ -1376,14 +1381,13 @@ public final class Rest {
RestUploadFile annfile = (RestUploadFile) ps[headIndex + 3];
RestURI annuri = (RestURI) ps[headIndex + 4];
RestUserid annuserid = (RestUserid) ps[headIndex + 5];
RestHeaders annheaders = (RestHeaders) ps[headIndex + 6];
boolean annheaders = (Boolean) ps[headIndex + 6];
RestParams annparams = (RestParams) ps[headIndex + 7];
if (CompletionHandler.class.isAssignableFrom(ptype)) { //HttpResponse.createAsyncHandler() or HttpResponse.createAsyncHandler(Class)
} else if (annsid != null) { //HttpRequest.getSessionid(true|false)
} else if (annaddr != null) { //HttpRequest.getRemoteAddr
} else if (annlocale != null) { //HttpRequest.getLocale
} else if (annheaders != null) { //HttpRequest.getHeaders
} else if (annparams != null) { //HttpRequest.getParameters
} else if (annbody != null) { //HttpRequest.getBodyUTF8 / HttpRequest.getBody
} else if (annfile != null) { //MultiContext.partsFirstBytes / HttpRequest.partsFirstFile / HttpRequest.partsFiles
@@ -1392,6 +1396,7 @@ public final class Rest {
} else if ("#".equals(pname)) { //从request.getRequstURI 中取参数
} else if (pname != null && pname.charAt(0) == '#') { //从request.getRequstURIPath 中去参数
} else if ("&".equals(pname) && ptype == userType) { //当前用户对象的类名
} else if ("^".equals(pname) && annheaders) { //HttpRequest.getHeaders Http头信息
} else if (ptype.isPrimitive()) {
//do nothing
} else if (ptype == String.class) {
@@ -2149,38 +2154,6 @@ public final class Rest {
required = false;
}
RestHeaders annheaders = param.getAnnotation(RestHeaders.class);
if (annheaders != null) {
if (annhead != null) {
throw new RestException("@RestHeaders and @RestHeader cannot on the same Parameter in " + method);
}
if (anncookie != null) {
throw new RestException("@RestHeaders and @RestCookie cannot on the same Parameter in " + method);
}
if (annsid != null) {
throw new RestException("@RestHeaders and @RestSessionid cannot on the same Parameter in " + method);
}
if (annaddr != null) {
throw new RestException("@RestHeaders and @RestAddress cannot on the same Parameter in " + method);
}
if (annlocale != null) {
throw new RestException("@RestHeaders and @RestLocale cannot on the same Parameter in " + method);
}
if (annbody != null) {
throw new RestException("@RestHeaders and @RestBody cannot on the same Parameter in " + method);
}
if (annfile != null) {
throw new RestException("@RestHeaders and @RestUploadFile cannot on the same Parameter in " + method);
}
if (userid != null) {
throw new RestException("@RestHeaders and @RestUserid cannot on the same Parameter in " + method);
}
if (!TYPE_MAP_STRING_STRING.equals(param.getParameterizedType())) {
throw new RestException("@RestHeaders must on Map<String, String> Parameter in " + method);
}
comment = "";
required = false;
}
RestParams annparams = param.getAnnotation(RestParams.class);
if (annparams != null) {
if (annhead != null) {
@@ -2207,15 +2180,45 @@ public final class Rest {
if (userid != null) {
throw new RestException("@RestParams and @RestUserid cannot on the same Parameter in " + method);
}
if (annheaders != null) {
throw new RestException("@RestParams and @RestHeaders cannot on the same Parameter in " + method);
}
if (!TYPE_MAP_STRING_STRING.equals(param.getParameterizedType())) {
throw new RestException("@RestParams must on Map<String, String> Parameter in " + method);
}
comment = "";
}
boolean annheaders = param.getType() == RestHeaders.class;
if (annheaders) {
if (annhead != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (anncookie != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (annsid != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (annaddr != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (annlocale != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (annbody != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (annfile != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (userid != null) {
throw new RestException("@RestHeaders cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
if (annparams != null) {
throw new RestException("@RestParams cannot on the " + RestHeaders.class.getSimpleName() + " Parameter in " + method);
}
comment = "";
required = false;
}
RestParam annpara = param.getAnnotation(RestParam.class);
if (annpara != null) {
radix = annpara.radix();
@@ -2232,6 +2235,9 @@ public final class Rest {
if (n == null && ptype == userType) {
n = "&"; //用户类型特殊处理
}
if (n == null && ptype == RestHeaders.class) {
n = "^"; //Http头信息类型特殊处理
}
if (n == null && asmParamNames != null && asmParamNames.size() > i) {
n = asmParamNames.get(i);
}
@@ -2505,10 +2511,10 @@ public final class Rest {
RestUploadFile annfile = (RestUploadFile) ps[headIndex + 3];
RestURI annuri = (RestURI) ps[headIndex + 4];
RestUserid userid = (RestUserid) ps[headIndex + 5];
RestHeaders annheaders = (RestHeaders) ps[headIndex + 6];
boolean annheaders = (Boolean) ps[headIndex + 6];
RestParams annparams = (RestParams) ps[headIndex + 7];
java.lang.reflect.Type pgentype = (java.lang.reflect.Type) ps[headIndex + 8];
if (dynsimple && (annsid != null || annaddr != null || annlocale != null || annhead != null || anncookie != null || annfile != null || annheaders != null)) {
if (dynsimple && (annsid != null || annaddr != null || annlocale != null || annhead != null || anncookie != null || annfile != null || annheaders)) {
dynsimple = false;
}
@@ -2549,9 +2555,9 @@ public final class Rest {
mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getLocale", "()Ljava/lang/String;", false);
mv.visitVarInsn(ASTORE, maxLocals);
varInsns.add(new int[]{ALOAD, maxLocals});
} else if (annheaders != null) { //HttpRequest.getHeaders
} else if (annheaders) { //HttpRequest.getHeaders
mv.visitVarInsn(ALOAD, 1);
mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getHeaders", "()Ljava/util/Map;", false);
mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getHeaders", "()" + httpHeaderDesc, false);
mv.visitVarInsn(ASTORE, maxLocals);
varInsns.add(new int[]{ALOAD, maxLocals});
} else if (annparams != null) { //HttpRequest.getParameters

30
src/main/java/org/redkale/net/http/RestHeaders.java
View File

@@ -5,24 +5,34 @@
*/
package org.redkale.net.http;
import java.lang.annotation.*;
import static java.lang.annotation.ElementType.*;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
/**
* 只能注解于RestService类的方法的参数或参数内的Map&#60;String, String&#62;字段
* 用于RestService类的方法的参数获取HttpHeader
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*
* @since 2.1.0
* @since 2.8.0
*/
@Inherited
@Documented
@Target({PARAMETER, FIELD})
@Retention(RUNTIME)
public @interface RestHeaders {
public interface RestHeaders {
public String firstValue(String name);
public String firstValue(String name, String defaultValue);
public List<String> listValue(String name);
public void forEach(BiConsumer<String, String> consumer);
public String[] names();
public boolean contains(String name);
public Map<String, Serializable> map();
}

1
src/main/java/org/redkale/net/sncp/SncpClient.java
View File

@@ -32,6 +32,7 @@ public class SncpClient extends Client<SncpClientConnection, SncpClientRequest,
super(name, group, "TCP".equalsIgnoreCase(netprotocol), address, maxConns, maxPipelines, null, null, null); //maxConns
this.clientSncpAddress = clientSncpAddress;
this.nodeid = nodeid;
this.connectTimeoutSeconds = 10;
this.readTimeoutSeconds = 10;
this.writeTimeoutSeconds = 10;
}

25
src/main/java/org/redkale/util/AnyValue.java
View File

@@ -574,7 +574,7 @@ public abstract class AnyValue {
if (this.anyEntrys == null) {
return this;
}
this.anyEntrys = Utility.remove(this.anyEntrys, (t) -> name.equals(((Entry) t).name));
this.anyEntrys = Utility.remove(this.anyEntrys, t -> name.equals(((Entry) t).name));
return this;
}
@@ -583,7 +583,7 @@ public abstract class AnyValue {
if (value == null || this.anyEntrys == null) {
return this;
}
this.anyEntrys = Utility.remove(this.anyEntrys, (t) -> name.equals(((Entry) t).name) && ((Entry) t).getValue().equals(value));
this.anyEntrys = Utility.remove(this.anyEntrys, t -> name.equals(((Entry) t).name) && ((Entry) t).getValue().equals(value));
return this;
}
@@ -592,7 +592,7 @@ public abstract class AnyValue {
if (this.stringEntrys == null) {
return this;
}
this.stringEntrys = Utility.remove(this.stringEntrys, (t) -> name.equals(((Entry) t).name));
this.stringEntrys = Utility.remove(this.stringEntrys, t -> name.equals(((Entry) t).name));
return this;
}
@@ -601,18 +601,23 @@ public abstract class AnyValue {
if (value == null || this.stringEntrys == null) {
return this;
}
this.stringEntrys = Utility.remove(this.stringEntrys, (t) -> name.equals(((Entry) t).name) && ((Entry) t).getValue().equals(value));
this.stringEntrys = Utility.remove(this.stringEntrys, t -> name.equals(((Entry) t).name) && ((Entry) t).getValue().equals(value));
return this;
}
@Override
public AnyValue getAnyValue(String name) {
return getAnyValue(name, false);
}
@Override
public AnyValue getAnyValue(String name, boolean create) {
for (Entry<DefaultAnyValue> en : this.anyEntrys) {
if (predicate.test(en.name, name)) {
return en.value;
}
}
return null;
return create ? new DefaultAnyValue() : null;
}
@Override
@@ -1274,6 +1279,16 @@ public abstract class AnyValue {
*/
public abstract AnyValue getAnyValue(String name);
/**
* 根据字段名获取AnyValue类型的字段值
*
* @param name 字段名
* @param create 没有是否创建一个新的对象返回
*
* @return AnyValue
*/
public abstract AnyValue getAnyValue(String name, boolean create);
/**
* 根据字段名获取String类型的字段值
*

4
src/main/java/org/redkale/util/ResourceListener.java
View File

@@ -5,9 +5,9 @@
*/
package org.redkale.util;
import java.lang.annotation.*;
import static java.lang.annotation.ElementType.METHOD;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
import java.lang.annotation.*;
/**
* &#64;Resource资源被更新时的监听事件, 本注解只能标记在方法参数为ResourceEvent[]上 <br>
@@ -26,7 +26,7 @@ import java.lang.annotation.*;
* &#64;ResourceListener
* private void changeResource(ResourceEvent[] events) {
* for(ResourceEvent event : events) {
* System.out.println("@Resource = " + event.name() + " 资源变更: newVal = " + event.newValue() + ", oldVal = " + event.oldValue());
* System.out .println("@Resource = " + event.name() + " 资源变更: newVal = " + event.newValue() + ", oldVal = " + event.oldValue());
* }
* }
*

255
src/main/java/org/redkale/util/Utility.java
View File

@@ -144,93 +144,118 @@ public final class Utility {
try {
final ClassLoader loader = Thread.currentThread().getContextClassLoader();
{ //virtualThreadLocalFunction
Class<Function<Supplier, ThreadLocal>> virtualClazz1 = null;
Class<Function<Supplier, ThreadLocal>> virtualClazz = null;
try {
virtualClazz1 = (Class) loader.loadClass("org.redkale.util.AnonymousThreadLocal");
virtualClazz = (Class) loader.loadClass("org.redkale.util.AnonymousThreadLocal");
} catch (Throwable t) {
//do nothing
}
if (virtualClazz1 == null) {
byte[] classBytes = hexToBin(functionThreadLocalBinary);
if (virtualClazz == null) {
try {
virtualClazz1 = (Class<Function<Supplier, ThreadLocal>>) new ClassLoader(loader) {
byte[] classBytes = hexToBin(functionThreadLocalBinary);
virtualClazz = (Class<Function<Supplier, ThreadLocal>>) new ClassLoader(loader) {
public final Class<?> loadClass(String name, byte[] b) {
return defineClass(name, b, 0, b.length);
}
}.loadClass("org.redkale.util.AnonymousThreadLocal", classBytes);
RedkaleClassLoader.putDynClass(virtualClazz1.getName(), classBytes, virtualClazz1);
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz1, virtualClazz1.getName());
RedkaleClassLoader.putDynClass(virtualClazz.getName(), classBytes, virtualClazz);
} catch (Throwable t) {
//do nothing
}
}
if (virtualClazz != null) {
try {
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz, virtualClazz.getName());
Supplier supplier = () -> null;
virtualThreadLocalFunction0 = virtualClazz1.getConstructor(Supplier.class).newInstance(supplier);
virtualThreadLocalFunction0 = virtualClazz.getConstructor(Supplier.class).newInstance(supplier);
} catch (Throwable t) {
//do nothing
}
}
}
{ //virtualThreadFactoryFunction
Class<Function<String, ThreadFactory>> virtualClazz1 = null;
Class<Function<String, ThreadFactory>> virtualClazz = null;
try {
virtualClazz1 = (Class) loader.loadClass("org.redkale.util.AnonymousThreadFactory");
virtualClazz = (Class) loader.loadClass("org.redkale.util.AnonymousThreadFactory");
} catch (Throwable t) {
//do nothing
}
if (virtualClazz1 == null) {
byte[] classBytes = hexToBin(functionThreadFactoryBinary);
if (virtualClazz == null) {
try {
virtualClazz1 = (Class<Function<String, ThreadFactory>>) new ClassLoader(loader) {
byte[] classBytes = hexToBin(functionThreadFactoryBinary);
virtualClazz = (Class<Function<String, ThreadFactory>>) new ClassLoader(loader) {
public final Class<?> loadClass(String name, byte[] b) {
return defineClass(name, b, 0, b.length);
}
}.loadClass("org.redkale.util.AnonymousThreadFactory", classBytes);
RedkaleClassLoader.putDynClass(virtualClazz1.getName(), classBytes, virtualClazz1);
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz1, virtualClazz1.getName());
virtualThreadFactoryFunction0 = virtualClazz1.getConstructor(String.class).newInstance(null);
RedkaleClassLoader.putDynClass(virtualClazz.getName(), classBytes, virtualClazz);
} catch (Throwable t) {
//do nothing
}
}
if (virtualClazz != null) {
try {
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz, virtualClazz.getName());
Object[] initargs = new Object[]{null};
virtualThreadFactoryFunction0 = virtualClazz.getConstructor(String.class).newInstance(initargs);
} catch (Throwable t) {
//do nothing
}
}
}
{ //virtualPoolFunction
Class<Function<String, ExecutorService>> virtualClazz1 = null;
Class<Function<String, ExecutorService>> virtualClazz = null;
try {
virtualClazz1 = (Class) loader.loadClass("org.redkale.util.AnonymousVirtualPoolFunction");
virtualClazz = (Class) loader.loadClass("org.redkale.util.AnonymousVirtualPoolFunction");
} catch (Throwable t) {
//do nothing
}
if (virtualClazz1 == null) {
byte[] classBytes = hexToBin(functionVirtualPoolBinary);
if (virtualClazz == null) {
try {
virtualClazz1 = (Class<Function<String, ExecutorService>>) new ClassLoader(loader) {
byte[] classBytes = hexToBin(functionVirtualPoolBinary);
virtualClazz = (Class<Function<String, ExecutorService>>) new ClassLoader(loader) {
public final Class<?> loadClass(String name, byte[] b) {
return defineClass(name, b, 0, b.length);
}
}.loadClass("org.redkale.util.AnonymousVirtualPoolFunction", classBytes);
RedkaleClassLoader.putDynClass(virtualClazz1.getName(), classBytes, virtualClazz1);
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz1, virtualClazz1.getName());
virtualPoolFunction0 = virtualClazz1.getConstructor().newInstance();
RedkaleClassLoader.putDynClass(virtualClazz.getName(), classBytes, virtualClazz);
} catch (Throwable t) {
//do nothing
}
}
if (virtualClazz != null) {
try {
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz, virtualClazz.getName());
virtualPoolFunction0 = virtualClazz.getConstructor().newInstance();
} catch (Throwable t) {
//do nothing
}
}
}
{ //virtualExecutorConsumer
Class<Executor> virtualClazz1 = null;
Class<Executor> virtualClazz = null;
try {
virtualClazz1 = (Class) loader.loadClass("org.redkale.util.AnonymousVirtualExecutor");
virtualClazz = (Class) loader.loadClass("org.redkale.util.AnonymousVirtualExecutor");
} catch (Throwable t) {
//do nothing
}
if (virtualClazz1 == null) {
byte[] classBytes = hexToBin(consumerVirtualExecutorBinary);
if (virtualClazz == null) {
try {
virtualClazz1 = (Class<Executor>) new ClassLoader(loader) {
byte[] classBytes = hexToBin(consumerVirtualExecutorBinary);
virtualClazz = (Class<Executor>) new ClassLoader(loader) {
public final Class<?> loadClass(String name, byte[] b) {
return defineClass(name, b, 0, b.length);
}
}.loadClass("org.redkale.util.AnonymousVirtualExecutor", classBytes);
RedkaleClassLoader.putDynClass(virtualClazz1.getName(), classBytes, virtualClazz1);
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz1, virtualClazz1.getName());
virtualExecutorConsumer0 = virtualClazz1.getConstructor().newInstance();
RedkaleClassLoader.putDynClass(virtualClazz.getName(), classBytes, virtualClazz);
} catch (Throwable t) {
//do nothing
}
}
if (virtualClazz != null) {
try {
RedkaleClassLoader.putReflectionDeclaredConstructors(virtualClazz, virtualClazz.getName());
virtualExecutorConsumer0 = virtualClazz.getConstructor().newInstance();
} catch (Throwable t) {
//do nothing
}
@@ -5055,11 +5080,11 @@ public final class Utility {
return remoteHttpContent("POST", url, timeoutMs, null, body).toString(StandardCharsets.UTF_8);
}
public static String postHttpContent(String url, Map<String, String> headers, String body) throws IOException {
public static String postHttpContent(String url, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("POST", url, 0, headers, body).toString(StandardCharsets.UTF_8);
}
public static String postHttpContent(String url, int timeoutMs, Map<String, String> headers, String body) throws IOException {
public static String postHttpContent(String url, int timeoutMs, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("POST", url, timeoutMs, headers, body).toString(StandardCharsets.UTF_8);
}
@@ -5079,11 +5104,11 @@ public final class Utility {
return remoteHttpContent("POST", url, timeoutMs, null, body).toString(charset.name());
}
public static String postHttpContent(String url, Charset charset, Map<String, String> headers, String body) throws IOException {
public static String postHttpContent(String url, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("POST", url, 0, headers, body).toString(charset.name());
}
public static String postHttpContent(String url, int timeoutMs, Charset charset, Map<String, String> headers, String body) throws IOException {
public static String postHttpContent(String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("POST", url, timeoutMs, headers, body).toString(charset.name());
}
@@ -5095,11 +5120,11 @@ public final class Utility {
return remoteHttpContent("POST", url, timeoutMs, null, null).toByteArray();
}
public static byte[] postHttpBytesContent(String url, Map<String, String> headers, String body) throws IOException {
public static byte[] postHttpBytesContent(String url, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("POST", url, 0, headers, body).toByteArray();
}
public static byte[] postHttpBytesContent(String url, int timeoutMs, Map<String, String> headers, String body) throws IOException {
public static byte[] postHttpBytesContent(String url, int timeoutMs, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("POST", url, timeoutMs, headers, body).toByteArray();
}
@@ -5111,11 +5136,11 @@ public final class Utility {
return remoteHttpContent("GET", url, timeoutMs, null, null).toString(StandardCharsets.UTF_8);
}
public static String getHttpContent(String url, Map<String, String> headers, String body) throws IOException {
public static String getHttpContent(String url, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("GET", url, 0, headers, body).toString(StandardCharsets.UTF_8);
}
public static String getHttpContent(String url, int timeoutMs, Map<String, String> headers, String body) throws IOException {
public static String getHttpContent(String url, int timeoutMs, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("GET", url, timeoutMs, headers, body).toString(StandardCharsets.UTF_8);
}
@@ -5127,11 +5152,11 @@ public final class Utility {
return remoteHttpContent("GET", url, timeoutMs, null, null).toString(charset.name());
}
public static String getHttpContent(String url, Charset charset, Map<String, String> headers, String body) throws IOException {
public static String getHttpContent(String url, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("GET", url, 0, headers, body).toString(charset.name());
}
public static String getHttpContent(String url, int timeoutMs, Charset charset, Map<String, String> headers, String body) throws IOException {
public static String getHttpContent(String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("GET", url, timeoutMs, headers, body).toString(charset.name());
}
@@ -5143,11 +5168,11 @@ public final class Utility {
return remoteHttpContent("GET", url, timeoutMs, null, null).toByteArray();
}
public static byte[] getHttpBytesContent(String url, Map<String, String> headers, String body) throws IOException {
public static byte[] getHttpBytesContent(String url, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("GET", url, 0, headers, body).toByteArray();
}
public static byte[] getHttpBytesContent(String url, int timeoutMs, Map<String, String> headers, String body) throws IOException {
public static byte[] getHttpBytesContent(String url, int timeoutMs, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent("GET", url, timeoutMs, headers, body).toByteArray();
}
@@ -5159,43 +5184,43 @@ public final class Utility {
return remoteHttpContentAsync(client, method, url, timeoutMs, null, null).thenApply(out -> out.toString(charset == null ? StandardCharsets.UTF_8 : charset)).join();
}
public static String remoteHttpContent(HttpClient client, String method, String url, Charset charset, Map<String, String> headers) throws IOException {
public static String remoteHttpContent(HttpClient client, String method, String url, Charset charset, Map<String, Serializable> headers) throws IOException {
return remoteHttpContentAsync(client, method, url, 0, headers, null).thenApply(out -> out.toString(charset == null ? StandardCharsets.UTF_8 : charset)).join();
}
public static String remoteHttpContent(HttpClient client, String method, String url, Charset charset, Map<String, String> headers, String body) throws IOException {
public static String remoteHttpContent(HttpClient client, String method, String url, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContentAsync(client, method, url, 0, headers, body).thenApply(out -> out.toString(charset == null ? StandardCharsets.UTF_8 : charset)).join();
}
public static String remoteHttpContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, String> headers) throws IOException {
public static String remoteHttpContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, Serializable> headers) throws IOException {
return remoteHttpContentAsync(client, method, url, timeoutMs, headers, null).thenApply(out -> out.toString(charset == null ? StandardCharsets.UTF_8 : charset)).join();
}
public static String remoteHttpContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, String> headers, String body) throws IOException {
public static String remoteHttpContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContentAsync(client, method, url, timeoutMs, headers, body).thenApply(out -> out.toString(charset == null ? StandardCharsets.UTF_8 : charset)).join();
}
public static byte[] remoteHttpBytesContent(HttpClient client, String method, String url, Charset charset, Map<String, String> headers, String body) throws IOException {
public static byte[] remoteHttpBytesContent(HttpClient client, String method, String url, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContentAsync(client, method, url, 0, headers, body).thenApply(out -> out.toByteArray()).join();
}
public static byte[] remoteHttpBytesContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, String> headers) throws IOException {
public static byte[] remoteHttpBytesContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, Serializable> headers) throws IOException {
return remoteHttpContentAsync(client, method, url, timeoutMs, headers, null).thenApply(out -> out.toByteArray()).join();
}
public static byte[] remoteHttpBytesContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, String> headers, String body) throws IOException {
public static byte[] remoteHttpBytesContent(HttpClient client, String method, String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContentAsync(client, method, url, timeoutMs, headers, body).thenApply(out -> out.toByteArray()).join();
}
public static ByteArrayOutputStream remoteHttpContent(String method, String url, Map<String, String> headers, String body) throws IOException {
public static ByteArrayOutputStream remoteHttpContent(String method, String url, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContent(method, url, 0, headers, body);
}
public static ByteArrayOutputStream remoteHttpContent(String method, String url, int timeoutMs, Map<String, String> headers, String body) throws IOException {
public static ByteArrayOutputStream remoteHttpContent(String method, String url, int timeoutMs, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContentAsync(method, url, timeoutMs, headers, body).join();
}
public static ByteArrayOutputStream remoteHttpContent(HttpClient client, String method, String url, int timeoutMs, Map<String, String> headers, String body) throws IOException {
public static ByteArrayOutputStream remoteHttpContent(HttpClient client, String method, String url, int timeoutMs, Map<String, Serializable> headers, String body) throws IOException {
return remoteHttpContentAsync(client, method, url, timeoutMs, headers, body).join();
}
@@ -5203,7 +5228,7 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, 0, null, null).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, null, null, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
@@ -5211,7 +5236,7 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, timeoutMs, null, null).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, null, null, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
@@ -5219,7 +5244,7 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, 0, null, body).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, null, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
@@ -5227,23 +5252,23 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, timeoutMs, null, body).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, null, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, Map<String, String> headers, String body) {
public static CompletableFuture<String> postHttpContentAsync(String url, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("POST", url, 0, headers, body).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, headers, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Map<String, String> headers, String body) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("POST", url, timeoutMs, headers, body).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, headers, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
@@ -5251,7 +5276,7 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, 0, null, null).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, null, null, respHeaders).thenApply(out -> out.toString(charset));
}
@@ -5259,7 +5284,7 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, timeoutMs, null, null).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, null, null, respHeaders).thenApply(out -> out.toString(charset));
}
@@ -5267,7 +5292,7 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, 0, null, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, null, body, respHeaders).thenApply(out -> out.toString(charset));
}
@@ -5275,23 +5300,23 @@ public final class Utility {
return remoteHttpContentAsync(client, "POST", url, 0, null, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Charset charset, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Charset charset, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(client, "POST", url, 0, null, body, respHeaders).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Map<String, String> headers, String body) {
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync(client, "POST", url, 0, headers, body).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(client, "POST", url, 0, headers, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Charset charset, Map<String, String> headers, String body) {
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Charset charset, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync(client, "POST", url, 0, headers, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Charset charset, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(HttpClient client, String url, Charset charset, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(client, "POST", url, 0, headers, body, respHeaders).thenApply(out -> out.toString(charset));
}
@@ -5299,23 +5324,23 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, timeoutMs, null, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, null, body, respHeaders).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, Map<String, String> headers, String body) {
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("POST", url, 0, headers, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, Charset charset, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, headers, body, respHeaders).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, String> headers, String body) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("POST", url, timeoutMs, headers, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> postHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, headers, body, respHeaders).thenApply(out -> out.toString(charset));
}
@@ -5323,7 +5348,7 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, 0, null, null).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, null, null, respHeaders).thenApply(out -> out.toByteArray());
}
@@ -5331,23 +5356,23 @@ public final class Utility {
return remoteHttpContentAsync("POST", url, timeoutMs, null, null).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, int timeoutMs, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, int timeoutMs, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, null, null, respHeaders).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, Map<String, String> headers, String body) {
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("POST", url, 0, headers, body).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, 0, headers, body, respHeaders).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, int timeoutMs, Map<String, String> headers, String body) {
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("POST", url, timeoutMs, headers, body).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, int timeoutMs, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> postHttpBytesContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("POST", url, timeoutMs, headers, body, respHeaders).thenApply(out -> out.toByteArray());
}
@@ -5355,7 +5380,7 @@ public final class Utility {
return remoteHttpContentAsync("GET", url, 0, null, null).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> getHttpContentAsync(String url, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, 0, null, null, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
@@ -5363,23 +5388,23 @@ public final class Utility {
return remoteHttpContentAsync("GET", url, timeoutMs, null, null).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, timeoutMs, null, null, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> getHttpContentAsync(String url, Map<String, String> headers, String body) {
public static CompletableFuture<String> getHttpContentAsync(String url, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("GET", url, 0, headers, body).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> getHttpContentAsync(String url, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, 0, headers, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Map<String, String> headers, String body) {
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("GET", url, timeoutMs, headers, body).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, timeoutMs, headers, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
@@ -5387,7 +5412,7 @@ public final class Utility {
return remoteHttpContentAsync("GET", url, 0, null, null).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(String url, Charset charset, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, Charset charset, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, 0, null, null, respHeaders).thenApply(out -> out.toString(charset));
}
@@ -5395,35 +5420,35 @@ public final class Utility {
return remoteHttpContentAsync("GET", url, timeoutMs, null, null).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, timeoutMs, null, null, respHeaders).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(String url, Charset charset, Map<String, String> headers, String body) {
public static CompletableFuture<String> getHttpContentAsync(String url, Charset charset, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("GET", url, 0, headers, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(String url, Charset charset, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, Charset charset, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, 0, headers, body, respHeaders).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, String> headers, String body) {
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("GET", url, timeoutMs, headers, body).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(String url, int timeoutMs, Charset charset, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, timeoutMs, headers, body, respHeaders).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(java.net.http.HttpClient client, String url, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(java.net.http.HttpClient client, String url, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(client, "GET", url, 0, null, body, respHeaders).thenApply(out -> out.toString(StandardCharsets.UTF_8));
}
public static CompletableFuture<String> getHttpContentAsync(java.net.http.HttpClient client, String url, Charset charset, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(java.net.http.HttpClient client, String url, Charset charset, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(client, "GET", url, 0, null, body, respHeaders).thenApply(out -> out.toString(charset));
}
public static CompletableFuture<String> getHttpContentAsync(java.net.http.HttpClient client, String url, Charset charset, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<String> getHttpContentAsync(java.net.http.HttpClient client, String url, Charset charset, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(client, "GET", url, 0, headers, body, respHeaders).thenApply(out -> out.toString(charset));
}
@@ -5431,7 +5456,7 @@ public final class Utility {
return remoteHttpContentAsync("GET", url, 0, null, null).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, 0, null, null, respHeaders).thenApply(out -> out.toByteArray());
}
@@ -5439,56 +5464,64 @@ public final class Utility {
return remoteHttpContentAsync("GET", url, timeoutMs, null, null).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, int timeoutMs, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, int timeoutMs, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, timeoutMs, null, null, respHeaders).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, Map<String, String> headers, String body) {
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("GET", url, 0, headers, body).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, 0, headers, body, respHeaders).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, int timeoutMs, Map<String, String> headers, String body) {
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync("GET", url, timeoutMs, headers, body).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, int timeoutMs, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> getHttpBytesContentAsync(String url, int timeoutMs, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync("GET", url, timeoutMs, headers, body, respHeaders).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<byte[]> getHttpBytesContentAsync(java.net.http.HttpClient client, String url, int timeoutMs, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<byte[]> getHttpBytesContentAsync(java.net.http.HttpClient client, String url, int timeoutMs, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(client, "GET", url, timeoutMs, headers, body, respHeaders).thenApply(out -> out.toByteArray());
}
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, Map<String, String> headers, String body) {
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync(method, url, 0, headers, body);
}
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(method, url, 0, headers, body, respHeaders);
}
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, int timeoutMs, Map<String, String> headers, String body) {
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, int timeoutMs, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync(httpClient, method, url, timeoutMs, headers, body);
}
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, int timeoutMs, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(String method, String url, int timeoutMs, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
return remoteHttpContentAsync(httpClient, method, url, timeoutMs, headers, body, respHeaders);
}
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(java.net.http.HttpClient client, String method, String url, int timeoutMs, Map<String, String> headers, String body) {
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(java.net.http.HttpClient client, String method, String url, int timeoutMs, Map<String, Serializable> headers, String body) {
return remoteHttpContentAsync(client, method, url, timeoutMs, headers, body, null);
}
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(java.net.http.HttpClient client, String method, String url, int timeoutMs, Map<String, String> headers, String body, Map<String, String> respHeaders) {
public static CompletableFuture<ByteArrayOutputStream> remoteHttpContentAsync(java.net.http.HttpClient client, String method, String url, int timeoutMs, Map<String, Serializable> headers, String body, Map<String, Serializable> respHeaders) {
java.net.http.HttpRequest.Builder builder = java.net.http.HttpRequest.newBuilder().uri(URI.create(url))
.timeout(Duration.ofMillis(timeoutMs > 0 ? timeoutMs : 6000))
.method(method, body == null ? java.net.http.HttpRequest.BodyPublishers.noBody() : java.net.http.HttpRequest.BodyPublishers.ofString(body));
if (headers != null) {
headers.forEach((n, v) -> builder.header(n, v));
headers.forEach((n, v) -> {
if (v instanceof Collection) {
for (Object val : (Collection) v) {
builder.header(n, val.toString());
}
} else {
builder.header(n, v.toString());
}
});
}
java.net.http.HttpClient c = client == null ? httpClient : client;
if (c == null) {
@@ -5541,7 +5574,7 @@ public final class Utility {
});
}
//
// public static ByteArrayOutputStream remoteHttpContent(SSLContext ctx, String method, String url, int timeoutMs, Map<String, String> headers, String body) throws IOException {
// public static ByteArrayOutputStream remoteHttpContent(SSLContext ctx, String method, String url, int timeoutMs, Map<String, Serializable> headers, String body) throws IOException {
// HttpURLConnection conn = (HttpURLConnection) new URL(url).openConnection();
// boolean opening = true;
// try {

73
src/test/java/org/redkale/test/http/RequestCoderTest.java Normal file
View File

@@ -0,0 +1,73 @@
/*
*
*/
package org.redkale.test.http;
import java.lang.reflect.Field;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.redkale.mq.HttpSimpleRequestCoder;
import org.redkale.net.client.ClientRequest;
import org.redkale.net.http.HttpSimpleRequest;
/**
*
* @author zhangjx
*/
public class RequestCoderTest {
private boolean main;
public static void main(String[] args) throws Throwable {
RequestCoderTest test = new RequestCoderTest();
test.main = true;
test.run1();
test.run2();
test.run3();
}
@Test
public void run1() throws Exception {
HttpSimpleRequest req1 = HttpSimpleRequest.create("/aaa");
System.out.println("simpleRequest1: " + req1);
byte[] bytes = HttpSimpleRequestCoder.getInstance().encode(req1);
HttpSimpleRequest req2 = HttpSimpleRequestCoder.getInstance().decode(bytes);
Field timeFiedl = ClientRequest.class.getDeclaredField("createTime");
timeFiedl.setAccessible(true);
timeFiedl.set(req2, req1.getCreateTime());
System.out.println("simpleRequest2: " + req2);
Assertions.assertEquals(req1.toString(), req2.toString());
}
@Test
public void run2() throws Exception {
HttpSimpleRequest req1 = HttpSimpleRequest.create("/aaa");
req1.addHeader("X-aaa", "aaa");
req1.param("bean", "{}");
System.out.println("simpleRequest1: " + req1);
byte[] bytes = HttpSimpleRequestCoder.getInstance().encode(req1);
HttpSimpleRequest req2 = HttpSimpleRequestCoder.getInstance().decode(bytes);
Field timeFiedl = ClientRequest.class.getDeclaredField("createTime");
timeFiedl.setAccessible(true);
timeFiedl.set(req2, req1.getCreateTime());
System.out.println("simpleRequest2: " + req2);
Assertions.assertEquals(req1.toString(), req2.toString());
}
@Test
public void run3() throws Exception {
HttpSimpleRequest req1 = HttpSimpleRequest.create("/aaa");
req1.addHeader("X-aaa", "aaa");
req1.addHeader("X-bbb", "bbb1");
req1.addHeader("X-bbb", "bbb2");
req1.param("bean", "{}");
System.out.println("simpleRequest1: " + req1);
byte[] bytes = HttpSimpleRequestCoder.getInstance().encode(req1);
HttpSimpleRequest req2 = HttpSimpleRequestCoder.getInstance().decode(bytes);
Field timeFiedl = ClientRequest.class.getDeclaredField("createTime");
timeFiedl.setAccessible(true);
timeFiedl.set(req2, req1.getCreateTime());
System.out.println("simpleRequest2: " + req2);
Assertions.assertEquals(req1.toString(), req2.toString());
}
}

5
src/test/java/org/redkale/test/rest/_DynHelloRestServlet1.java
View File

@@ -1,6 +1,7 @@
package org.redkale.test.rest;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import org.redkale.annotation.Resource;
import org.redkale.net.http.*;
@@ -29,11 +30,11 @@ public class _DynHelloRestServlet1 extends SimpleRestServlet {
DefaultAnyValue conf = DefaultAnyValue.create("port", "" + port);
server.init(conf);
server.start();
Thread.sleep(100);
Utility.sleep(100);
HelloEntity entity = new HelloEntity();
entity.setHelloname("my name");
Map<String, String> headers = new HashMap<>();
Map<String, Serializable> headers = new HashMap<>();
headers.put("hello-res", "my res");
//headers.put(Rest.REST_HEADER_RESNAME, "my-res");
String url = "http://127.0.0.1:" + port + "/pipes/hello/update?entity={}&bean2={}";