tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-18 06:27:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

ref[net]: 重构net，简化了大量代码

Browse Source
This commit is contained in:
jaysunxiao
2021-10-15 12:57:57 +08:00
parent 0d8d1acfa1
commit e9e2158b33
31 changed files with 393 additions and 468 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/src/main/java/com/zfoo/net/config/model/HostConfig.java
-81
View File
@@ -1,81 +0,0 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.config.model;
import java.util.Map;
import java.util.Objects;
/**
* @author jaysunxiao
* @version 3.0
*/
public class HostConfig {
private String center;
private String user;
private String password;
private Map<String, String> address;
public void setCenter(String center) {
this.center = center;
}
public String getCenter() {
return center;
}
public String getUser() {
return user;
}
public void setUser(String user) {
this.user = user;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public Map<String, String> getAddress() {
return address;
}
public void setAddress(Map<String, String> address) {
this.address = address;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
HostConfig that = (HostConfig) o;
return Objects.equals(center, that.center) &&
Objects.equals(user, that.user) &&
Objects.equals(password, that.password) &&
Objects.equals(address, that.address);
}
@Override
public int hashCode() {
return Objects.hash(center, user, password, address);
}
}
net/src/main/java/com/zfoo/net/config/model/NetConfig.java
+1 -11
View File
@@ -39,7 +39,6 @@ public class NetConfig {
private RegistryConfig registry;
private MonitorConfig monitor;
private HostConfig host;
private ProviderConfig provider;
private ConsumerConfig consumer;
@@ -121,14 +120,6 @@ public class NetConfig {
this.monitor = monitor;
}
public HostConfig getHost() {
return host;
}
public void setHost(HostConfig host) {
this.host = host;
}
public ProviderConfig getProvider() {
return provider;
}
@@ -159,13 +150,12 @@ public class NetConfig {
Objects.equals(protocolLocation, netConfig.protocolLocation) &&
Objects.equals(registry, netConfig.registry) &&
Objects.equals(monitor, netConfig.monitor) &&
Objects.equals(host, netConfig.host) &&
Objects.equals(provider, netConfig.provider) &&
Objects.equals(consumer, netConfig.consumer);
}
@Override
public int hashCode() {
return Objects.hash(id, protocolLocation, generateJsProtocol, registry, monitor, host, provider, consumer);
return Objects.hash(id, protocolLocation, generateJsProtocol, registry, monitor, provider, consumer);
}
}
net/src/main/java/com/zfoo/net/config/model/ProviderConfig.java
+10 -10
View File
@@ -32,9 +32,9 @@ public class ProviderConfig {
/**
* 对应于ITaskDispatch
*/
private String dispatch;
private String taskDispatch;
private String dispatchThread;
private String thread;
private String address;
@@ -56,20 +56,20 @@ public class ProviderConfig {
return HostAndPort.valueOf(address);
}
public String getDispatch() {
return dispatch;
public String getTaskDispatch() {
return taskDispatch;
}
public void setDispatch(String dispatch) {
this.dispatch = dispatch;
public void setTaskDispatch(String taskDispatch) {
this.taskDispatch = taskDispatch;
}
public String getDispatchThread() {
return dispatchThread;
public String getThread() {
return thread;
}
public void setDispatchThread(String dispatchThread) {
this.dispatchThread = dispatchThread;
public void setThread(String thread) {
this.thread = thread;
}
public String getAddress() {
net/src/main/java/com/zfoo/net/packet/service/PacketService.java
+10 -3
View File
@@ -24,6 +24,7 @@ import com.zfoo.protocol.exception.ExceptionUtils;
import com.zfoo.protocol.generate.GenerateOperation;
import com.zfoo.protocol.generate.GenerateProtocolFile;
import com.zfoo.protocol.registration.IProtocolRegistration;
import com.zfoo.protocol.serializer.CodeLanguage;
import com.zfoo.protocol.util.DomUtils;
import com.zfoo.protocol.xml.XmlProtocols;
import io.netty.buffer.ByteBuf;
@@ -89,9 +90,15 @@ public class PacketService implements IPacketService {
var generateOperation = new GenerateOperation();
generateOperation.setFoldProtocol(foldProtocol);
generateOperation.setProtocolParam(protocolParam);
generateOperation.setGenerateJsProtocol(generateJsProtocol);
generateOperation.setGenerateCsharpProtocol(generateCsharpProtocol);
generateOperation.setGenerateLuaProtocol(generateLuaProtocol);
if (generateJsProtocol) {
generateOperation.getGenerateLanguages().add(CodeLanguage.JavaScript);
}
if (generateCsharpProtocol) {
generateOperation.getGenerateLanguages().add(CodeLanguage.CSharp);
}
if (generateLuaProtocol) {
generateOperation.getGenerateLanguages().add(CodeLanguage.Lua);
}
// 设置生成协议的过滤器
GenerateProtocolFile.generateProtocolFilter = netGenerateProtocolFilter;
net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java
+2 -20
View File
@@ -107,12 +107,6 @@ public class NetDefinitionParser implements BeanDefinitionParser {
builder.addPropertyReference("monitor", MonitorConfig.class.getCanonicalName());
}
var hostElement = DomUtils.getFirstChildElementByTagName(element, "host");
if (hostElement != null) {
builder.addPropertyReference("host", HostConfig.class.getCanonicalName());
parseHostConfig(hostElement, parserContext);
}
var providerElement = DomUtils.getFirstChildElementByTagName(element, "provider");
if (providerElement != null) {
builder.addPropertyReference("provider", ProviderConfig.class.getCanonicalName());
@@ -152,24 +146,12 @@ public class NetDefinitionParser implements BeanDefinitionParser {
parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition());
}
private void parseHostConfig(Element element, ParserContext parserContext) {
var clazz = HostConfig.class;
var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz);
resolvePlaceholder("center", "center", builder, element, parserContext);
resolvePlaceholder("user", "user", builder, element, parserContext);
resolvePlaceholder("password", "password", builder, element, parserContext);
var addressMap = parseAddress(element, parserContext);
builder.addPropertyValue("address", addressMap);
parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition());
}
private void parseProviderConfig(Element element, ParserContext parserContext) {
var clazz = ProviderConfig.class;
var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz);
resolvePlaceholder("dispatch", "dispatch", builder, element, parserContext);
resolvePlaceholder("dispatch-thread", "dispatchThread", builder, element, parserContext);
resolvePlaceholder("task-dispatch", "taskDispatch", builder, element, parserContext);
resolvePlaceholder("thread", "thread", builder, element, parserContext);
resolvePlaceholder("address", "address", builder, element, parserContext);
var providerModules = parseModules("provider", element, parserContext);
net/src/main/java/com/zfoo/net/task/TaskBus.java
+8 -9
View File
@@ -14,8 +14,8 @@
package com.zfoo.net.task;
import com.zfoo.net.NetContext;
import com.zfoo.net.task.dispatcher.AbstractTaskDispatcher;
import com.zfoo.net.task.dispatcher.ITaskDispatcher;
import com.zfoo.net.task.dispatcher.AbstractTaskDispatch;
import com.zfoo.net.task.dispatcher.ITaskDispatch;
import com.zfoo.net.task.model.PacketReceiverTask;
import com.zfoo.protocol.util.StringUtils;
import org.slf4j.Logger;
@@ -35,7 +35,7 @@ public final class TaskBus {
// 线程池的大小
public static final int EXECUTOR_SIZE;
private static final ITaskDispatcher taskRoute;
private static final ITaskDispatch taskDispatch;
/**
@@ -47,12 +47,11 @@ public final class TaskBus {
var localConfig = NetContext.getConfigManager().getLocalConfig();
var providerConfig = localConfig.getProvider();
var dispatch = providerConfig == null ? "consistent-hash" : providerConfig.getDispatch();
var dispatchThread = (providerConfig == null || StringUtils.isBlank(providerConfig.getDispatchThread()))
? "default" : providerConfig.getDispatchThread();
taskDispatch = AbstractTaskDispatch.valueOf(providerConfig == null ? "consistent-hash" : providerConfig.getTaskDispatch());
EXECUTOR_SIZE = "default".equals(dispatchThread) ? (Runtime.getRuntime().availableProcessors() + 1) : Integer.parseInt(dispatchThread);
taskRoute = AbstractTaskDispatcher.valueOf(dispatch);
EXECUTOR_SIZE = (providerConfig == null || StringUtils.isBlank(providerConfig.getThread()))
? (Runtime.getRuntime().availableProcessors() + 1)
: Integer.parseInt(providerConfig.getThread());
executors = new ExecutorService[EXECUTOR_SIZE];
for (int i = 0; i < executors.length; i++) {
@@ -63,7 +62,7 @@ public final class TaskBus {
public static void submit(PacketReceiverTask task) {
taskRoute.getExecutor(task).execute(task);
taskDispatch.getExecutor(task).execute(task);
}
public static ExecutorService executor(int executorConsistentHash) {
net/src/main/java/com/zfoo/net/task/dispatcher/AbstractTaskDispatcher.java → net/src/main/java/com/zfoo/net/task/dispatcher/AbstractTaskDispatch.java
+5 -5
View File
@@ -19,16 +19,16 @@ import com.zfoo.protocol.util.StringUtils;
* @author jaysunxiao
* @version 3.0
*/
public abstract class AbstractTaskDispatcher implements ITaskDispatcher {
public abstract class AbstractTaskDispatch implements ITaskDispatch {
public static ITaskDispatcher valueOf(String taskDispatchName) {
public static ITaskDispatch valueOf(String taskDispatchName) {
switch (taskDispatchName) {
case "random":
return new RandomTaskDispatcher();
return new RandomTaskDispatch();
case "sessionId":
return new SessionIdTaskDispatcher();
return new SessionIdTaskDispatch();
case "consistent-hash":
return new ConsistentHashTaskDispatcher();
return new ConsistentHashTaskDispatch();
default:
throw new RuntimeException(StringUtils.format("没有找到对应的taskDispatch[{}]", taskDispatchName));
}
net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatcher.java → net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java
+4 -4
View File
@@ -22,11 +22,11 @@ import java.util.concurrent.ExecutorService;
* @author jaysunxiao
* @version 3.0
*/
public class ConsistentHashTaskDispatcher extends AbstractTaskDispatcher {
public class ConsistentHashTaskDispatch extends AbstractTaskDispatch {
private static ConsistentHashTaskDispatcher INSTANCE = new ConsistentHashTaskDispatcher();
private static ConsistentHashTaskDispatch INSTANCE = new ConsistentHashTaskDispatch();
public static ConsistentHashTaskDispatcher getINSTANCE() {
public static ConsistentHashTaskDispatch getINSTANCE() {
return INSTANCE;
}
@@ -35,7 +35,7 @@ public class ConsistentHashTaskDispatcher extends AbstractTaskDispatcher {
var packetAttachment = packetReceiverTask.getPacketAttachment();
if (packetAttachment == null) {
return SessionIdTaskDispatcher.getInstance().getExecutor(packetReceiverTask);
return SessionIdTaskDispatch.getInstance().getExecutor(packetReceiverTask);
}
return TaskBus.executor(packetAttachment.executorConsistentHash());
net/src/main/java/com/zfoo/net/task/dispatcher/ITaskDispatcher.java → net/src/main/java/com/zfoo/net/task/dispatcher/ITaskDispatch.java
+1 -1
View File
@@ -21,7 +21,7 @@ import java.util.concurrent.ExecutorService;
* @author jaysunxiao
* @version 3.0
*/
public interface ITaskDispatcher {
public interface ITaskDispatch {
ExecutorService getExecutor(PacketReceiverTask packetReceiverTask);
net/src/main/java/com/zfoo/net/task/dispatcher/RandomTaskDispatcher.java → net/src/main/java/com/zfoo/net/task/dispatcher/RandomTaskDispatch.java
+3 -3
View File
@@ -22,11 +22,11 @@ import java.util.concurrent.ExecutorService;
* @author jaysunxiao
* @version 3.0
*/
public class RandomTaskDispatcher extends AbstractTaskDispatcher {
public class RandomTaskDispatch extends AbstractTaskDispatch {
private static final RandomTaskDispatcher INSTANCE = new RandomTaskDispatcher();
private static final RandomTaskDispatch INSTANCE = new RandomTaskDispatch();
public static RandomTaskDispatcher getInstance() {
public static RandomTaskDispatch getInstance() {
return INSTANCE;
}
net/src/main/java/com/zfoo/net/task/dispatcher/SessionIdTaskDispatcher.java → net/src/main/java/com/zfoo/net/task/dispatcher/SessionIdTaskDispatch.java
+3 -3
View File
@@ -25,11 +25,11 @@ import java.util.concurrent.ExecutorService;
* @author jaysunxiao
* @version 3.0
*/
public class SessionIdTaskDispatcher extends AbstractTaskDispatcher {
public class SessionIdTaskDispatch extends AbstractTaskDispatch {
private static final SessionIdTaskDispatcher INSTANCE = new SessionIdTaskDispatcher();
private static final SessionIdTaskDispatch INSTANCE = new SessionIdTaskDispatch();
public static SessionIdTaskDispatcher getInstance() {
public static SessionIdTaskDispatch getInstance() {
return INSTANCE;
}
net/src/main/resources/net-1.0.xsd
+2 -5
View File
@@ -32,8 +32,8 @@
<xsd:sequence>
<xsd:element name="module" maxOccurs="unbounded" type="moduleAttributeType" minOccurs="0"/>
</xsd:sequence>
<xsd:attribute name="dispatch" type="xsd:string" use="required"/>
<xsd:attribute name="dispatch-thread" type="xsd:string" use="optional"/>
<xsd:attribute name="task-dispatch" type="xsd:string" use="required"/>
<xsd:attribute name="thread" type="xsd:string" use="optional"/>
<xsd:attribute name="address" type="xsd:string" use="optional"/>
</xsd:complexType>
@@ -57,9 +57,6 @@
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="monitor" type="addressType"/>
</xsd:choice>
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="host" type="addressType"/>
</xsd:choice>
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="provider" type="providerType"/>
</xsd:choice>
net/src/test/java/com/zfoo/net/core/tcp/client/TcpClientPacketController.java → net/src/test/java/com/zfoo/net/core/tcp/client/TcpClientController.java
+2 -2
View File
@@ -26,10 +26,10 @@ import org.springframework.stereotype.Component;
* @version 3.0
*/
@Component
public class TcpClientPacketController {
public class TcpClientController {
private static final Logger logger = LoggerFactory.getLogger(TcpClientPacketController.class);
private static final Logger logger = LoggerFactory.getLogger(TcpClientController.class);
@PacketReceiver
public void atTcpHelloResponse(Session session, TcpHelloResponse response) {
net/src/test/java/com/zfoo/net/core/tcp/client/TcpClientTest.java
+4 -79
View File
@@ -15,11 +15,7 @@ package com.zfoo.net.core.tcp.client;
import com.zfoo.net.NetContext;
import com.zfoo.net.core.tcp.TcpClient;
import com.zfoo.net.packet.tcp.*;
import com.zfoo.net.session.SessionUtils;
import com.zfoo.net.task.model.SafeRunnable;
import com.zfoo.protocol.exception.ExceptionUtils;
import com.zfoo.protocol.util.JsonUtils;
import com.zfoo.net.packet.tcp.TcpHelloRequest;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
@@ -28,9 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author jaysunxiao
* @version 3.0
@@ -41,10 +34,10 @@ public class TcpClientTest {
private static final Logger logger = LoggerFactory.getLogger(TcpClientTest.class);
@Test
public void startClient0() {
public void startClient() {
var context = new ClassPathXmlApplicationContext("config.xml");
SessionUtils.printSessionInfo();
var client = new TcpClient(HostAndPort.valueOf(NetContext.getConfigManager().getLocalConfig().getHost().getAddress().get("server0")));
var client = new TcpClient(HostAndPort.valueOf("127.0.0.1:9000"));
var session = client.start();
var request = new TcpHelloRequest();
@@ -59,72 +52,4 @@ public class TcpClientTest {
ThreadUtils.sleep(Long.MAX_VALUE);
}
@Test
public void syncClientTest() {
var context = new ClassPathXmlApplicationContext("config.xml");
SessionUtils.printSessionInfo();
var client = new TcpClient(HostAndPort.valueOf(NetContext.getConfigManager().getLocalConfig().getHost().getAddress().get("server1")));
var session = client.start();
var executorSize = Runtime.getRuntime().availableProcessors() * 2;
var executor = Executors.newFixedThreadPool(executorSize);
var atomicInteger = new AtomicInteger(0);
for (int i = 0; i < executorSize; i++) {
var thread = new Thread(() -> {
try {
for (int j = 0; j < 10000; j++) {
var ask = new SyncMessAsk();
ask.setMessage("Hello, this is sync client!");
var answer = NetContext.getRouter().syncAsk(session, ask, SyncMessAnswer.class, null).packet();
logger.info("同步请求[{}]收到结果[{}]", atomicInteger.incrementAndGet(), JsonUtils.object2String(answer));
}
} catch (Exception e) {
logger.error(ExceptionUtils.getMessage(e));
}
});
executor.execute(thread);
}
ThreadUtils.sleep(Long.MAX_VALUE);
}
@Test
public void asyncClientTest() {
var context = new ClassPathXmlApplicationContext("config.xml");
var client1 = new TcpClient(HostAndPort.valueOf(NetContext.getConfigManager().getLocalConfig().getHost().getAddress().get("server1")));
var session1 = client1.start();
var executorSize = Runtime.getRuntime().availableProcessors() * 2;
var executor = Executors.newFixedThreadPool(executorSize);
var atomicInteger = new AtomicInteger(0);
for (int i = 0; i < executorSize; i++) {
var thread = new Thread(() -> {
for (int j = 0; j < 1000; j++) {
var ask = new AsyncMess0Ask();
ask.setMessage("Hello, client0 -> server0!");
NetContext.getRouter().asyncAsk(null, ask, AsyncMess0Answer.class, null)
.notComplete(new SafeRunnable() {
@Override
public void doRun() {
logger.info("异步请求没有完成");
}
})
.whenComplete(answer -> {
logger.info("异步请求[{}]收到结果[{}]", atomicInteger.incrementAndGet(), JsonUtils.object2String(answer));
}
);
}
});
executor.execute(thread);
}
SessionUtils.printSessionInfo();
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/core/tcp/server/TcpServerController.java
+44
View File
@@ -0,0 +1,44 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.core.tcp.server;
import com.zfoo.net.NetContext;
import com.zfoo.net.packet.tcp.TcpHelloRequest;
import com.zfoo.net.packet.tcp.TcpHelloResponse;
import com.zfoo.net.router.receiver.PacketReceiver;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.util.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* @author jaysunxiao
* @version 3.0
*/
@Component
public class TcpServerController {
private static final Logger logger = LoggerFactory.getLogger(TcpServerController.class);
@PacketReceiver
public void atTcpHelloRequest(Session session, TcpHelloRequest request) {
logger.info("receive [packet:{}] from client", JsonUtils.object2String(request));
var response = new TcpHelloResponse();
response.setMessage("Hello, this is the tcp server!");
NetContext.getRouter().send(session, response);
}
}
net/src/test/java/com/zfoo/net/core/tcp/server/TcpServerPacketController.java
-99
View File
@@ -1,99 +0,0 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.core.tcp.server;
import com.zfoo.net.NetContext;
import com.zfoo.net.packet.tcp.*;
import com.zfoo.net.router.receiver.PacketReceiver;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.util.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* @author jaysunxiao
* @version 3.0
*/
@Component
public class TcpServerPacketController {
private static final Logger logger = LoggerFactory.getLogger(TcpServerPacketController.class);
@PacketReceiver
public void atTcpHelloRequest(Session session, TcpHelloRequest request) {
logger.info("receive [packet:{}] from client", JsonUtils.object2String(request));
var response = new TcpHelloResponse();
response.setMessage("Hello, this is the tcp server!");
NetContext.getRouter().send(session, response);
}
@PacketReceiver
public void atSyncMessAsk(Session session, SyncMessAsk ask) {
logger.info("receive [packet:{}] from client", JsonUtils.object2String(ask));
// 测试超时
// ThreadUtils.sleep(Integer.MAX_VALUE);
// 测试正常返回
var answer = new SyncMessAnswer();
answer.setMessage("Hello, this is sync server answer!");
// 测试返回不是预期的消息
// SM_Int sm = new SM_Int();
// 测试错误返回
// var sm = ErrorResponse.valueOf(1, 1, "this is error response");
NetContext.getRouter().send(session, answer);
}
// client->server1->server0->client
@PacketReceiver
public void atAsyncMess0Ask(Session session, AsyncMess0Ask ask0) {
var ask1 = new AsyncMess1Ask();
ask1.setMessage("Hello, server0 -> server1");
var client0 = NetContext.getSessionManager().getClientSession(0L);
NetContext.getRouter().asyncAsk(client0, ask1, AsyncMess1Answer.class, null)
.whenComplete(sm_asyncMess0 -> {
var answer = new AsyncMess0Answer();
answer.setMessage("Hello, server1 -> client!");
NetContext.getRouter().send(session, answer);
});
}
@PacketReceiver
public void atAsyncMess1Ask(Session session, AsyncMess1Ask ask) {
// 测试超时
// ThreadUtils.sleep(Integer.MAX_VALUE);
// 测试正常返回
var answer = new AsyncMess1Answer();
answer.setMessage("Hello, server0 -> server1!");
// 测试返回不是预期的消息
// SM_Int sm = new SM_Int();
// 测试错误返回
// var sm = ErrorResponse.valueOf(1, 1, "this is error response");
NetContext.getRouter().send(session, answer);
}
}
net/src/test/java/com/zfoo/net/core/tcp/server/TcpServerTest.java
+3 -38
View File
@@ -13,18 +13,13 @@
package com.zfoo.net.core.tcp.server;
import com.zfoo.net.NetContext;
import com.zfoo.net.core.tcp.TcpClient;
import com.zfoo.net.core.tcp.TcpServer;
import com.zfoo.net.session.SessionUtils;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import java.util.concurrent.Executors;
/**
* @author jaysunxiao
* @version 3.0
@@ -38,42 +33,12 @@ public class TcpServerTest {
* startClientTest连接服务器成功过后，会不断的发消息给服务器
*/
@Test
public void startServer0() {
var context = new ClassPathXmlApplicationContext("config.xml");
SessionUtils.printSessionInfo();
var server0 = new TcpServer(HostAndPort.valueOf(NetContext.getConfigManager().getLocalConfig().getHost().getAddress().get("server0")));
server0.start();
ThreadUtils.sleep(Long.MAX_VALUE);
}
@Test
public void startServer1() {
public void startServer() {
var context = new ClassPathXmlApplicationContext("config.xml");
SessionUtils.printSessionInfo();
var server1 = new TcpServer(HostAndPort.valueOf(NetContext.getConfigManager().getLocalConfig().getHost().getAddress().get("server1")));
server1.start();
// 连接server0
var executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
executor.execute(() -> {
while (true) {
try {
var client0 = new TcpClient(HostAndPort.valueOf(NetContext.getConfigManager().getLocalConfig().getHost().getAddress().get("client0")));
client0.start();
break;
} catch (Exception e) {
System.out.println("连接失败，开始重新连接");
ThreadUtils.sleep(3000);
e.printStackTrace();
}
}
});
var server = new TcpServer(HostAndPort.valueOf("127.0.0.1:9000"));
server.start();
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/core/tcpSync/client/TcpClientTest.java
+56
View File
@@ -0,0 +1,56 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.core.tcpSync.client;
import com.zfoo.net.NetContext;
import com.zfoo.net.core.tcp.TcpClient;
import com.zfoo.net.packet.tcp.SyncMessAnswer;
import com.zfoo.net.packet.tcp.SyncMessAsk;
import com.zfoo.protocol.util.JsonUtils;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author jaysunxiao
* @version 3.0
*/
@Ignore
public class TcpClientTest {
private static final Logger logger = LoggerFactory.getLogger(TcpClientTest.class);
@Test
public void startClient() throws Exception {
var context = new ClassPathXmlApplicationContext("config.xml");
var client = new TcpClient(HostAndPort.valueOf("127.0.0.1:9000"));
var session = client.start();
for (int i = 0; i < 1000; i++) {
var ask = new SyncMessAsk();
ask.setMessage("Hello, this is sync client!");
var answer = NetContext.getRouter().syncAsk(session, ask, SyncMessAnswer.class, null).packet();
logger.info("同步请求收到结果[{}]", JsonUtils.object2String(answer));
ThreadUtils.sleep(1000);
}
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/core/tcpSync/server/TcpServerTest.java
+45
View File
@@ -0,0 +1,45 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.core.tcpSync.server;
import com.zfoo.net.core.tcp.TcpServer;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author jaysunxiao
* @version 3.0
*/
@Ignore
public class TcpServerTest {
/**
* 单机服务器教程，启动成功过后在com.zfoo.net.core.tcp.client.TcpClientTest中运行startClientTest
* <p>
* startClientTest连接服务器成功过后，会不断的发消息给服务器
*/
@Test
public void startServer() {
var context = new ClassPathXmlApplicationContext("config.xml");
var server = new TcpServer(HostAndPort.valueOf("127.0.0.1:9000"));
server.start();
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/core/tcpSync/server/TcpSyncController.java
+44
View File
@@ -0,0 +1,44 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.core.tcpSync.server;
import com.zfoo.net.NetContext;
import com.zfoo.net.packet.tcp.SyncMessAnswer;
import com.zfoo.net.packet.tcp.SyncMessAsk;
import com.zfoo.net.router.receiver.PacketReceiver;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.util.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* @author jaysunxiao
* @version 3.0
*/
@Component
public class TcpSyncController {
private static final Logger logger = LoggerFactory.getLogger(TcpSyncController.class);
@PacketReceiver
public void atSyncMessAsk(Session session, SyncMessAsk ask) {
logger.info("receive [packet:{}] from client", JsonUtils.object2String(ask));
var answer = new SyncMessAnswer();
answer.setMessage("Hello, this is the tcp server!");
NetContext.getRouter().send(session, answer);
}
}
net/src/test/java/com/zfoo/net/core/tpcAsync/client/TcpClientTest.java
+61
View File
@@ -0,0 +1,61 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.core.tpcAsync.client;
import com.zfoo.net.NetContext;
import com.zfoo.net.core.tcp.TcpClient;
import com.zfoo.net.packet.tcp.AsyncMessAnswer;
import com.zfoo.net.packet.tcp.AsyncMessAsk;
import com.zfoo.protocol.util.JsonUtils;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author jaysunxiao
* @version 3.0
*/
@Ignore
public class TcpClientTest {
private static final Logger logger = LoggerFactory.getLogger(TcpClientTest.class);
@Test
public void startClient() throws Exception {
var context = new ClassPathXmlApplicationContext("config.xml");
var client = new TcpClient(HostAndPort.valueOf("127.0.0.1:9000"));
var session = client.start();
for (int i = 0; i < 1000; i++) {
var ask = new AsyncMessAsk();
ask.setMessage("Hello, this is async client!");
NetContext.getRouter().asyncAsk(session, ask, AsyncMessAnswer.class, null)
.whenComplete(answer -> {
logger.info("同步请求收到结果[{}]", JsonUtils.object2String(answer));
}
);
ThreadUtils.sleep(1000);
}
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/core/tpcAsync/server/TcpAsyncController.java
+44
View File
@@ -0,0 +1,44 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.core.tpcAsync.server;
import com.zfoo.net.NetContext;
import com.zfoo.net.packet.tcp.AsyncMessAnswer;
import com.zfoo.net.packet.tcp.AsyncMessAsk;
import com.zfoo.net.router.receiver.PacketReceiver;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.util.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* @author jaysunxiao
* @version 3.0
*/
@Component
public class TcpAsyncController {
private static final Logger logger = LoggerFactory.getLogger(TcpAsyncController.class);
@PacketReceiver
public void atAsyncMessAsk(Session session, AsyncMessAsk ask) {
logger.info("receive [packet:{}] from client", JsonUtils.object2String(ask));
var answer = new AsyncMessAnswer();
answer.setMessage("Hello, this is the tcp server!");
NetContext.getRouter().send(session, answer);
}
}
net/src/test/java/com/zfoo/net/packet/tcp/AsyncMess1Answer.java → net/src/test/java/com/zfoo/net/core/tpcAsync/server/TcpServerTest.java
+16 -17
View File
@@ -1,5 +1,6 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
* in compliance with the License. You may obtain a copy of the License at
*
@@ -10,32 +11,30 @@
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.packet.tcp;
package com.zfoo.net.core.tpcAsync.server;
import com.zfoo.protocol.IPacket;
import com.zfoo.net.core.tcp.TcpServer;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author jaysunxiao
* @version 3.0
*/
public class AsyncMess1Answer implements IPacket {
@Ignore
public class TcpServerTest {
public static final transient short PROTOCOL_ID = 1155;
@Test
public void startServer() {
var context = new ClassPathXmlApplicationContext("config.xml");
private String message;
var server = new TcpServer(HostAndPort.valueOf("127.0.0.1:9000"));
server.start();
@Override
public short protocolId() {
return PROTOCOL_ID;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/packet/tcp/AsyncMess0Answer.java → net/src/test/java/com/zfoo/net/packet/tcp/AsyncMessAnswer.java
+1 -1
View File
@@ -18,7 +18,7 @@ import com.zfoo.protocol.IPacket;
* @author jaysunxiao
* @version 3.0
*/
public class AsyncMess0Answer implements IPacket {
public class AsyncMessAnswer implements IPacket {
public static final transient short PROTOCOL_ID = 1153;
net/src/test/java/com/zfoo/net/packet/tcp/AsyncMess0Ask.java → net/src/test/java/com/zfoo/net/packet/tcp/AsyncMessAsk.java
+1 -1
View File
@@ -18,7 +18,7 @@ import com.zfoo.protocol.IPacket;
* @author jaysunxiao
* @version 3.0
*/
public class AsyncMess0Ask implements IPacket {
public class AsyncMessAsk implements IPacket {
public static final transient short PROTOCOL_ID = 1152;
net/src/test/resources/config.xml
-8
View File
@@ -19,14 +19,6 @@
<context:component-scan base-package="com.zfoo"/>
<net:config id="applicationNameTest" protocol-location="protocol.xml">
<net:host center="direct connect" user="jaysunxiao" password="123456">
<net:address name="server0" url="127.0.0.1:9000"/>
<net:address name="server1" url="127.0.0.1:9001"/>
<net:address name="client0" url="127.0.0.1:9000"/>
<net:address name="client1" url="127.0.0.1:9001"/>
</net:host>
</net:config>
</beans>
net/src/test/resources/protocol.xml
+2 -4
View File
@@ -51,10 +51,8 @@
<protocol id="1150" location="com.zfoo.net.packet.tcp.SyncMessAsk" enhance="false"/>
<protocol id="1151" location="com.zfoo.net.packet.tcp.SyncMessAnswer" enhance="false"/>
<protocol id="1152" location="com.zfoo.net.packet.tcp.AsyncMess0Ask" enhance="false"/>
<protocol id="1153" location="com.zfoo.net.packet.tcp.AsyncMess0Answer" enhance="false"/>
<protocol id="1154" location="com.zfoo.net.packet.tcp.AsyncMess1Ask" enhance="false"/>
<protocol id="1155" location="com.zfoo.net.packet.tcp.AsyncMess1Answer" enhance="false"/>
<protocol id="1152" location="com.zfoo.net.packet.tcp.AsyncMessAsk" enhance="false"/>
<protocol id="1153" location="com.zfoo.net.packet.tcp.AsyncMessAnswer" enhance="false"/>
<protocol id="1165" location="com.zfoo.net.packet.csharp.CM_CSharpRequest" enhance="false"/>
<protocol id="1166" location="com.zfoo.net.packet.csharp.CSharpObjectA" enhance="false"/>
net/src/test/resources/provider/provider_config.xml
+1 -1
View File
@@ -25,7 +25,7 @@
<net:address name="${registry.address.name}" url="${registry.address.url}"/>
</net:registry>
<net:provider dispatch="consistent-hash">
<net:provider task-dispatch="consistent-hash">
<net:module name="providerTest"/>
</net:provider>
protocol/src/main/java/com/zfoo/protocol/generate/GenerateOperation.java
+9 -34
View File
@@ -13,6 +13,11 @@
package com.zfoo.protocol.generate;
import com.zfoo.protocol.serializer.CodeLanguage;
import java.util.HashSet;
import java.util.Set;
/**
* 创建协议文件的操作类
*
@@ -37,19 +42,9 @@ public class GenerateOperation {
private String protocolParam;
/**
* 生成javascript协议文件
* 需要生成的协议文件
*/
private boolean generateJsProtocol;
/**
* 生成C#协议文件
*/
private boolean generateCsharpProtocol;
/**
* 生成Lua协议文件
*/
private boolean generateLuaProtocol;
private Set<CodeLanguage> generateLanguages = new HashSet<>();
public boolean isFoldProtocol() {
return foldProtocol;
@@ -67,28 +62,8 @@ public class GenerateOperation {
this.protocolParam = protocolParam;
}
public boolean isGenerateJsProtocol() {
return generateJsProtocol;
}
public void setGenerateJsProtocol(boolean generateJsProtocol) {
this.generateJsProtocol = generateJsProtocol;
}
public boolean isGenerateCsharpProtocol() {
return generateCsharpProtocol;
}
public void setGenerateCsharpProtocol(boolean generateCsharpProtocol) {
this.generateCsharpProtocol = generateCsharpProtocol;
}
public boolean isGenerateLuaProtocol() {
return generateLuaProtocol;
}
public void setGenerateLuaProtocol(boolean generateLuaProtocol) {
this.generateLuaProtocol = generateLuaProtocol;
public Set<CodeLanguage> getGenerateLanguages() {
return generateLanguages;
}
}
protocol/src/main/java/com/zfoo/protocol/generate/GenerateProtocolFile.java
+6 -12
View File
@@ -16,10 +16,10 @@ import com.zfoo.protocol.ProtocolManager;
import com.zfoo.protocol.registration.IProtocolRegistration;
import com.zfoo.protocol.registration.ProtocolAnalysis;
import com.zfoo.protocol.registration.ProtocolRegistration;
import com.zfoo.protocol.serializer.CodeLanguage;
import com.zfoo.protocol.serializer.cs.GenerateCsUtils;
import com.zfoo.protocol.serializer.js.GenerateJsUtils;
import com.zfoo.protocol.serializer.lua.GenerateLuaUtils;
import com.zfoo.protocol.util.ReflectionUtils;
import java.io.IOException;
import java.util.Arrays;
@@ -58,14 +58,7 @@ public abstract class GenerateProtocolFile {
var protocols = ProtocolManager.protocols;
// 如果没有需要生成的协议则直接返回
var generateProtocolFlag = Arrays.stream(generateOperation.getClass().getDeclaredFields())
.filter(it -> it.getName().startsWith("generate"))
.peek(it -> ReflectionUtils.makeAccessible(it))
.map(it -> ReflectionUtils.getField(it, generateOperation))
.filter(it -> it instanceof Boolean)
.anyMatch(it -> ((Boolean) it).booleanValue() == true);
if (!generateProtocolFlag) {
if (generateOperation.getGenerateLanguages().isEmpty()) {
return;
}
@@ -102,21 +95,22 @@ public abstract class GenerateProtocolFile {
}
// 生成C#协议
if (generateOperation.isGenerateCsharpProtocol()) {
var generateLanguages = generateOperation.getGenerateLanguages();
if (generateLanguages.contains(CodeLanguage.CSharp)) {
GenerateCsUtils.init();
GenerateCsUtils.createProtocolManager();
allSortedGenerateProtocols.forEach(it -> GenerateCsUtils.createCsProtocolFile((ProtocolRegistration) it));
}
// 生成Javascript协议
if (generateOperation.isGenerateJsProtocol()) {
if (generateLanguages.contains(CodeLanguage.JavaScript)) {
GenerateJsUtils.init();
allSortedGenerateProtocols.forEach(it -> GenerateJsUtils.createJsProtocolFile((ProtocolRegistration) it));
GenerateJsUtils.createProtocolManager(allSortedGenerateProtocols);
}
// 生成Lua协议
if (generateOperation.isGenerateLuaProtocol()) {
if (generateLanguages.contains(CodeLanguage.Lua)) {
GenerateLuaUtils.init();
GenerateLuaUtils.createProtocolManager(allSortedGenerateProtocols);
allSortedGenerateProtocols.forEach(it -> GenerateLuaUtils.createLuaProtocolFile((ProtocolRegistration) it));
net/src/test/java/com/zfoo/net/packet/tcp/AsyncMess1Ask.java → protocol/src/main/java/com/zfoo/protocol/serializer/CodeLanguage.java
+5 -17
View File
@@ -10,30 +10,18 @@
* See the License for the specific language governing permissions and limitations under the License.
*/
package com.zfoo.net.packet.tcp;
import com.zfoo.protocol.IPacket;
package com.zfoo.protocol.serializer;
/**
* @author jaysunxiao
* @version 3.0
*/
public class AsyncMess1Ask implements IPacket {
public enum CodeLanguage {
public static final transient short PROTOCOL_ID = 1154;
JavaScript,
private String message;
Lua,
@Override
public short protocolId() {
return PROTOCOL_ID;
}
CSharp;
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}