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

sncp优化

Browse Source
This commit is contained in:
redkale
2023-02-07 00:50:53 +08:00
parent 5df83b2649
commit 797d04fea7
16 changed files with 393 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/redkale/boot/NodeServer.java
View File

@@ -490,7 +490,7 @@ public abstract class NodeServer {
}
final ResourceTypeLoader resourceLoader = (ResourceFactory rf, String srcResourceName, final Object srcObj, final String resourceName, Field field, final Object attachment) -> {
try {
if (SncpOldClient.parseMethodActions(serviceImplClass).isEmpty()
if (Sncp.loadMethodActions(serviceImplClass).isEmpty()
&& (serviceImplClass.getAnnotation(Priority.class) == null && serviceImplClass.getAnnotation(javax.annotation.Priority.class) == null)) { //class没有可用的方法且没有标记启动优先级的， 通常为BaseService
if (!serviceImplClass.getName().startsWith("org.redkale.") && !serviceImplClass.getSimpleName().contains("Base")) {
logger.log(Level.FINE, serviceImplClass + " cannot load because not found less one public non-final method");

2
src/main/java/org/redkale/net/Server.java
View File

@@ -309,7 +309,7 @@ public abstract class Server<K extends Serializable, C extends Context, R extend
SocketAddress localAddress = serverChannel.getLocalAddress();
if (localAddress instanceof InetSocketAddress && !Objects.equals(localAddress, this.address)) {
this.address = (InetSocketAddress) localAddress;
this.context.updateServerAddress(this.address);
//this.context.updateServerAddress(this.address);
}
serverChannel.accept(application, this);
postStart();

88
src/main/java/org/redkale/net/sncp/Sncp.java
View File

@@ -19,10 +19,10 @@ import static org.redkale.asm.ClassWriter.COMPUTE_FRAMES;
import org.redkale.asm.*;
import static org.redkale.asm.Opcodes.*;
import org.redkale.asm.Type;
import org.redkale.mq.MessageAgent;
import org.redkale.mq.*;
import org.redkale.net.TransportFactory;
import org.redkale.net.http.WebSocketNode;
import org.redkale.net.sncp.SncpOldClient.SncpAction;
import org.redkale.net.sncp.SncpServiceInfo.SncpServiceAction;
import org.redkale.service.*;
import org.redkale.util.*;
@@ -77,6 +77,83 @@ public abstract class Sncp {
return Arrays.copyOf(PONG_BYTES, PONG_BYTES.length);
}
//key: actionid
public static LinkedHashMap<Uint128, Method> loadMethodActions(final Class resourceServiceType) {
final List<Method> list = new ArrayList<>();
final List<Method> multis = new ArrayList<>();
final Map<Uint128, Method> actionids = new LinkedHashMap<>();
for (final java.lang.reflect.Method method : resourceServiceType.getMethods()) {
if (method.isSynthetic()) {
continue;
}
if (Modifier.isStatic(method.getModifiers())) {
continue;
}
if (Modifier.isFinal(method.getModifiers())) {
continue;
}
if (method.getAnnotation(Local.class) != null) {
continue;
}
if (method.getName().equals("getClass") || method.getName().equals("toString")) {
continue;
}
if (method.getName().equals("equals") || method.getName().equals("hashCode")) {
continue;
}
if (method.getName().equals("notify") || method.getName().equals("notifyAll") || method.getName().equals("wait")) {
continue;
}
if (method.getParameterCount() == 1 && method.getParameterTypes()[0] == AnyValue.class) {
if (method.getName().equals("init") || method.getName().equals("stop") || method.getName().equals("destroy")) {
continue;
}
}
Uint128 actionid = Sncp.actionid(method);
Method old = actionids.get(actionid);
if (old != null) {
if (old.getDeclaringClass().equals(method.getDeclaringClass())) {
throw new SncpException(resourceServiceType.getName() + " have one more same action(Method=" + method + ", " + old + ", actionid=" + actionid + ")");
}
continue;
}
actionids.put(actionid, method);
if (method.getAnnotation(Sncp.SncpDyn.class) != null) {
multis.add(method);
} else {
list.add(method);
}
}
multis.sort((m1, m2) -> m1.getAnnotation(Sncp.SncpDyn.class).index() - m2.getAnnotation(Sncp.SncpDyn.class).index());
list.sort((Method o1, Method o2) -> {
if (!o1.getName().equals(o2.getName())) {
return o1.getName().compareTo(o2.getName());
}
if (o1.getParameterCount() != o2.getParameterCount()) {
return o1.getParameterCount() - o2.getParameterCount();
}
return 0;
});
//带SncpDyn必须排在前面
multis.addAll(list);
final LinkedHashMap<Uint128, Method> rs = new LinkedHashMap<>();
for (Method method : multis) {
for (Map.Entry<Uint128, Method> en : actionids.entrySet()) {
if (en.getValue() == method) {
rs.put(en.getKey(), en.getValue());
break;
}
}
}
return rs;
}
public static <T extends Service> SncpServiceInfo createSncpServiceInfo(String resourceName,
Class<T> resourceServiceType, T service, MessageAgent messageAgent, SncpMessageClient messageClient) {
return new SncpServiceInfo(resourceName, resourceServiceType, service, messageAgent, messageClient);
}
public static Uint128 actionid(final RpcAction action) {
return hash(action.name());
}
@@ -790,7 +867,12 @@ public abstract class Sncp {
}
int i = -1;
Uint128 serviceid = serviceid(name, serviceTypeOrImplClass);
for (final SncpAction entry : SncpOldClient.getSncpActions(realed ? createLocalServiceClass(loader, name, serviceTypeOrImplClass) : serviceTypeOrImplClass, serviceid)) {
final List<SncpServiceAction> serviceActions = new ArrayList<>();
Class serviceImpClass = realed ? createLocalServiceClass(loader, name, serviceTypeOrImplClass) : serviceTypeOrImplClass;
for (Map.Entry<Uint128, Method> en : loadMethodActions(serviceImpClass).entrySet()) {
serviceActions.add(new SncpServiceAction(serviceImpClass, en.getValue(), serviceid, en.getKey()));
}
for (final SncpServiceAction entry : serviceActions) {
final int index = ++i;
final java.lang.reflect.Method method = entry.method;
{

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

@@ -4,21 +4,31 @@
package org.redkale.net.sncp;
import java.net.InetSocketAddress;
import org.redkale.annotation.Resource;
import org.redkale.convert.bson.BsonConvert;
import org.redkale.net.*;
import org.redkale.net.client.*;
/**
* SNCP版Client
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*
* @since 2.8.0
*/
public class SncpClient extends Client<SncpClientConnection, SncpClientRequest, SncpClientResult> {
private InetSocketAddress sncpAddress;
private final InetSocketAddress clientSncpAddress;
@SuppressWarnings("OverridableMethodCallInConstructor")
public SncpClient(String name, AsyncGroup group, InetSocketAddress sncpAddress, ClientAddress address, int maxConns, int maxPipelines) {
super(name, group, true, address, maxConns, maxPipelines, null, null, null); //maxConns
this.sncpAddress = sncpAddress;
@Resource
protected BsonConvert bsonConvert;
public SncpClient(String name, AsyncGroup group, InetSocketAddress clientSncpAddress, ClientAddress address, String netprotocol, int maxConns, int maxPipelines) {
super(name, group, "TCP".equalsIgnoreCase(netprotocol), address, maxConns, maxPipelines, null, null, null); //maxConns
this.clientSncpAddress = clientSncpAddress;
}
@Override
@@ -26,8 +36,16 @@ public class SncpClient extends Client<SncpClientConnection, SncpClientRequest,
return new SncpClientConnection(this, index, channel);
}
public InetSocketAddress getSncpAddress() {
return sncpAddress;
public InetSocketAddress getClientSncpAddress() {
return clientSncpAddress;
}
protected SncpClientConnection connect(SncpServiceInfo info) {
return null;
}
//只给远程模式调用的
public <T> T remote(final SncpServiceInfo info, final int index, final Object... params) {
return null;
}
}

6
src/main/java/org/redkale/net/sncp/SncpClientCodec.java
View File

@@ -10,8 +10,14 @@ import static org.redkale.net.sncp.SncpHeader.HEADER_SIZE;
import org.redkale.util.*;
/**
* SncpClient编解码器
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*
* @since 2.8.0
*/
public class SncpClientCodec extends ClientCodec<SncpClientRequest, SncpClientResult> {

6
src/main/java/org/redkale/net/sncp/SncpClientConnection.java
View File

@@ -8,8 +8,14 @@ import org.redkale.net.client.*;
import org.redkale.util.ObjectPool;
/**
* client版连接
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*
* @since 2.8.0
*/
public class SncpClientConnection extends ClientConnection<SncpClientRequest, SncpClientResult> {

6
src/main/java/org/redkale/net/sncp/SncpClientRequest.java
View File

@@ -8,8 +8,14 @@ import org.redkale.net.client.*;
import org.redkale.util.ByteArray;
/**
* client版请求
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*
* @since 2.8.0
*/
public class SncpClientRequest extends ClientRequest {

6
src/main/java/org/redkale/net/sncp/SncpClientResult.java
View File

@@ -9,8 +9,14 @@ import java.util.Objects;
import org.redkale.util.ByteArray;
/**
* client版响应结果
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*
* @since 2.8.0
*/
public class SncpClientResult {

2
src/main/java/org/redkale/net/sncp/SncpDynServlet.java
View File

@@ -49,7 +49,7 @@ public final class SncpDynServlet extends SncpServlet {
this.maxNameLength = maxNameLength;
this.serviceid = Sncp.serviceid(serviceResourceName, serviceResourceType);
RedkaleClassLoader.putReflectionPublicMethods(service.getClass().getName());
for (Map.Entry<Uint128, Method> en : SncpOldClient.parseMethodActions(service.getClass()).entrySet()) {
for (Map.Entry<Uint128, Method> en : Sncp.loadMethodActions(service.getClass()).entrySet()) {
SncpServletAction action;
try {
action = SncpServletAction.create(service, serviceid, en.getKey(), en.getValue());

28
src/main/java/org/redkale/net/sncp/SncpHeader.java
View File

@@ -22,6 +22,7 @@ public class SncpHeader {
private Uint128 serviceid;
//【预留字段】service接口版本
private int serviceVersion;
private Uint128 actionid;
@@ -94,15 +95,24 @@ public class SncpHeader {
}
public ByteArray writeTo(ByteArray array, byte[] newAddrBytes, int newAddrPort, long newSeqid, int bodyLength, int retcode) {
array.putLong(newSeqid); //8
array.putChar((char) HEADER_SIZE); //2
array.putUint128(serviceid); //16
array.putInt(serviceVersion); //4
array.putUint128(actionid); //16
array.put(newAddrBytes); //4
array.putChar((char) newAddrPort); //2
array.putInt(bodyLength); //4
array.putInt(retcode); //4
int offset = 0;
array.putLong(offset, newSeqid); //8
offset += 8;
array.putChar(offset, (char) HEADER_SIZE); //2
offset += 2;
array.putUint128(offset, serviceid); //16
offset += 16;
array.putInt(offset, serviceVersion); //4
offset += 4;
array.putUint128(offset, actionid); //16
offset += 16;
array.put(offset, newAddrBytes); //4
offset += 4;
array.putChar(offset, (char) newAddrPort); //2
offset += 2;
array.putInt(offset, bodyLength); //4
offset += 4;
array.putInt(offset, retcode); //4
return array;
}

205
src/main/java/org/redkale/net/sncp/SncpOldClient.java
View File

@@ -5,7 +5,6 @@
*/
package org.redkale.net.sncp;
import java.lang.annotation.Annotation;
import java.lang.reflect.*;
import java.net.*;
import java.nio.ByteBuffer;
@@ -21,6 +20,7 @@ import org.redkale.net.*;
import org.redkale.net.sncp.Sncp.SncpDyn;
import static org.redkale.net.sncp.SncpHeader.HEADER_SIZE;
import static org.redkale.net.sncp.SncpRequest.*;
import org.redkale.net.sncp.SncpServiceInfo.SncpServiceAction;
import org.redkale.service.*;
import org.redkale.source.*;
import org.redkale.util.*;
@@ -54,7 +54,7 @@ public final class SncpOldClient {
protected final int serviceVersion;
protected final SncpAction[] actions;
protected final SncpServiceAction[] actions;
protected final MessageAgent messageAgent;
@@ -83,12 +83,12 @@ public final class SncpOldClient {
this.name = serviceResourceName;
Class<?> serviceResourceType = ResourceFactory.getResourceType(serviceTypeOrImplClass); //serviceResourceType
this.serviceid = Sncp.serviceid(serviceResourceName, serviceResourceType);
final List<SncpAction> methodens = new ArrayList<>();
final List<SncpServiceAction> methodens = new ArrayList<>();
//------------------------------------------------------------------------------
for (Map.Entry<Uint128, Method> en : parseMethodActions(serviceClass).entrySet()) {
methodens.add(new SncpAction(serviceClass, en.getValue(), serviceid, en.getKey()));
for (Map.Entry<Uint128, Method> en : Sncp.loadMethodActions(serviceClass).entrySet()) {
methodens.add(new SncpServiceAction(serviceClass, en.getValue(), serviceid, en.getKey()));
}
this.actions = methodens.toArray(new SncpAction[methodens.size()]);
this.actions = methodens.toArray(new SncpServiceAction[methodens.size()]);
this.addrBytes = clientSncpAddress == null ? new byte[4] : clientSncpAddress.getAddress().getAddress();
this.addrPort = clientSncpAddress == null ? 0 : clientSncpAddress.getPort();
if (this.addrBytes.length != 4) {
@@ -96,15 +96,6 @@ public final class SncpOldClient {
}
}
static List<SncpAction> getSncpActions(final Class serviceClass, Uint128 serviceid) {
final List<SncpAction> actions = new ArrayList<>();
//------------------------------------------------------------------------------
for (Map.Entry<Uint128, Method> en : parseMethodActions(serviceClass).entrySet()) {
actions.add(new SncpAction(serviceClass, en.getValue(), serviceid, en.getKey()));
}
return actions;
}
public MessageAgent getMessageAgent() {
return messageAgent;
}
@@ -168,80 +159,9 @@ public final class SncpOldClient {
+ ", actions.size = " + actions.length + ")";
}
public static LinkedHashMap<Uint128, Method> parseMethodActions(final Class serviceClass) {
final List<Method> list = new ArrayList<>();
final List<Method> multis = new ArrayList<>();
final Map<Uint128, Method> actionids = new LinkedHashMap<>();
for (final java.lang.reflect.Method method : serviceClass.getMethods()) {
if (method.isSynthetic()) {
continue;
}
if (Modifier.isStatic(method.getModifiers())) {
continue;
}
if (Modifier.isFinal(method.getModifiers())) {
continue;
}
if (method.getAnnotation(Local.class) != null) {
continue;
}
if (method.getName().equals("getClass") || method.getName().equals("toString")) {
continue;
}
if (method.getName().equals("equals") || method.getName().equals("hashCode")) {
continue;
}
if (method.getName().equals("notify") || method.getName().equals("notifyAll") || method.getName().equals("wait")) {
continue;
}
if (method.getParameterCount() == 1 && method.getParameterTypes()[0] == AnyValue.class) {
if (method.getName().equals("init") || method.getName().equals("stop") || method.getName().equals("destroy")) {
continue;
}
}
Uint128 actionid = Sncp.actionid(method);
Method old = actionids.get(actionid);
if (old != null) {
if (old.getDeclaringClass().equals(method.getDeclaringClass())) {
throw new SncpException(serviceClass.getName() + " have one more same action(Method=" + method + ", " + old + ", actionid=" + actionid + ")");
}
continue;
}
actionids.put(actionid, method);
if (method.getAnnotation(SncpDyn.class) != null) {
multis.add(method);
} else {
list.add(method);
}
}
multis.sort((m1, m2) -> m1.getAnnotation(SncpDyn.class).index() - m2.getAnnotation(SncpDyn.class).index());
list.sort((Method o1, Method o2) -> {
if (!o1.getName().equals(o2.getName())) {
return o1.getName().compareTo(o2.getName());
}
if (o1.getParameterCount() != o2.getParameterCount()) {
return o1.getParameterCount() - o2.getParameterCount();
}
return 0;
});
//带SncpDyn必须排在前面
multis.addAll(list);
final LinkedHashMap<Uint128, Method> rs = new LinkedHashMap<>();
for (Method method : multis) {
for (Map.Entry<Uint128, Method> en : actionids.entrySet()) {
if (en.getValue() == method) {
rs.put(en.getKey(), en.getValue());
break;
}
}
}
return rs;
}
//只给远程模式调用的
public <T> T remote(final int index, final Object... params) {
final SncpAction action = actions[index];
final SncpServiceAction action = actions[index];
final CompletionHandler handlerFunc = action.handlerFuncParamIndex >= 0 ? (CompletionHandler) params[action.handlerFuncParamIndex] : null;
if (action.handlerFuncParamIndex >= 0) {
params[action.handlerFuncParamIndex] = null;
@@ -295,7 +215,7 @@ public final class SncpOldClient {
}
}
private CompletableFuture<byte[]> remote0(final CompletionHandler handler, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) {
private CompletableFuture<byte[]> remote0(final CompletionHandler handler, final Transport transport, final SocketAddress addr0, final SncpServiceAction action, final Object... params) {
final String traceid = Traces.currTraceid();
final Type[] myparamtypes = action.paramTypes;
final Class[] myparamclass = action.paramClass;
@@ -369,7 +289,6 @@ public final class SncpOldClient {
final AsyncConnection conn = conn0;
final ByteArray array = writer.toByteArray();
fillHeader(array, action, seqid, traceid, reqBodyLength);
conn.write(array, new CompletionHandler<Integer, Void>() {
@Override
@@ -433,6 +352,7 @@ public final class SncpOldClient {
success();
}
} catch (Throwable e) {
e.printStackTrace();
future.completeExceptionally(new RpcRemoteException(action.method + " sncp[" + conn.getRemoteAddress() + "] remote response error, params=" + JsonConvert.root().convertTo(params)));
transport.offerConnection(true, conn);
if (handler != null) {
@@ -496,7 +416,7 @@ public final class SncpOldClient {
});
}
private void checkResult(long seqid, final SncpAction action, ByteBuffer buffer) {
private void checkResult(long seqid, final SncpServiceAction action, ByteBuffer buffer) {
long rseqid = buffer.getLong();
if (rseqid != seqid) {
throw new SncpException("sncp(" + action.method + ") response.seqid = " + seqid + ", but request.seqid =" + rseqid);
@@ -521,111 +441,8 @@ public final class SncpOldClient {
buffer.getChar(); //端口
}
private void fillHeader(ByteArray buffer, SncpAction action, long seqid, String traceid, int bodyLength) {
private void fillHeader(ByteArray buffer, SncpServiceAction action, long seqid, String traceid, int bodyLength) {
action.header.writeTo(buffer, addrBytes, addrPort, seqid, bodyLength, 0); //结果码， 请求方固定传0
}
protected static final class SncpAction {
protected final Uint128 actionid;
protected final Method method;
protected final Type resultTypes; //void 必须设为 null
protected final Type[] paramTypes;
protected final Class[] paramClass;
protected final Attribute[] paramAttrs; // 为null表示无RpcCall处理，index=0固定为null, 其他为参数标记的RpcCall回调方法
protected final int handlerFuncParamIndex;
protected final int handlerAttachParamIndex;
protected final int addressTargetParamIndex;
protected final int addressSourceParamIndex;
protected final int topicTargetParamIndex;
protected final boolean boolReturnTypeFuture; // 返回结果类型是否为 CompletableFuture
protected final Creator<? extends CompletableFuture> futureCreator;
protected final SncpHeader header;
@SuppressWarnings("unchecked")
public SncpAction(final Class clazz, Method method, Uint128 serviceid, Uint128 actionid) {
this.actionid = actionid == null ? Sncp.actionid(method) : actionid;
Type rt = TypeToken.getGenericType(method.getGenericReturnType(), clazz);
this.resultTypes = rt == void.class ? null : rt;
this.boolReturnTypeFuture = CompletableFuture.class.isAssignableFrom(method.getReturnType());
this.futureCreator = boolReturnTypeFuture ? Creator.create((Class<? extends CompletableFuture>) method.getReturnType()) : null;
this.paramTypes = TypeToken.getGenericType(method.getGenericParameterTypes(), clazz);
this.paramClass = method.getParameterTypes();
this.method = method;
Annotation[][] anns = method.getParameterAnnotations();
int tpoicAddrIndex = -1;
int targetAddrIndex = -1;
int sourceAddrIndex = -1;
int handlerAttachIndex = -1;
int handlerFuncIndex = -1;
boolean hasattr = false;
Attribute[] atts = new Attribute[paramTypes.length + 1];
if (anns.length > 0) {
Class<?>[] params = method.getParameterTypes();
for (int i = 0; i < params.length; i++) {
if (CompletionHandler.class.isAssignableFrom(params[i])) {
if (boolReturnTypeFuture) {
throw new SncpException(method + " have both CompletionHandler and CompletableFuture");
}
if (handlerFuncIndex >= 0) {
throw new SncpException(method + " have more than one CompletionHandler type parameter");
}
Sncp.checkAsyncModifier(params[i], method);
handlerFuncIndex = i;
break;
}
}
for (int i = 0; i < anns.length; i++) {
if (anns[i].length > 0) {
for (Annotation ann : anns[i]) {
if (ann.annotationType() == RpcAttachment.class) {
if (handlerAttachIndex >= 0) {
throw new SncpException(method + " have more than one @RpcAttachment parameter");
}
handlerAttachIndex = i;
} else if (ann.annotationType() == RpcTargetAddress.class && SocketAddress.class.isAssignableFrom(params[i])) {
targetAddrIndex = i;
} else if (ann.annotationType() == RpcSourceAddress.class && SocketAddress.class.isAssignableFrom(params[i])) {
sourceAddrIndex = i;
} else if (ann.annotationType() == RpcTargetTopic.class && String.class.isAssignableFrom(params[i])) {
tpoicAddrIndex = i;
}
}
}
}
}
this.topicTargetParamIndex = tpoicAddrIndex;
this.addressTargetParamIndex = targetAddrIndex;
this.addressSourceParamIndex = sourceAddrIndex;
this.handlerFuncParamIndex = handlerFuncIndex;
this.handlerAttachParamIndex = handlerAttachIndex;
this.paramAttrs = hasattr ? atts : null;
this.header = new SncpHeader(null, serviceid, actionid);
if (this.handlerFuncParamIndex >= 0 && method.getReturnType() != void.class) {
throw new SncpException(method + " have CompletionHandler type parameter but return type is not void");
}
}
public String actionName() {
return method.getDeclaringClass().getSimpleName() + "." + method.getName();
}
@Override
public String toString() {
return "{" + actionid + "," + (method == null ? "null" : method.getName()) + "}";
}
}
}

2
src/main/java/org/redkale/net/sncp/SncpRequest.java
View File

@@ -59,7 +59,7 @@ public class SncpRequest extends Request<SncpContext> {
this.header = new SncpHeader();
int headerSize = this.header.read(buffer);
if (headerSize != HEADER_SIZE) {
context.getLogger().log(Level.WARNING, "sncp buffer header.length not " + HEADER_SIZE);
context.getLogger().log(Level.WARNING, "sncp buffer header.length not " + HEADER_SIZE + ", but " + headerSize);
return -1;
}
if (this.header.getRetcode() != 0) { // retcode

218
src/main/java/org/redkale/net/sncp/SncpServiceInfo.java Normal file
View File

@@ -0,0 +1,218 @@
/*
*
*/
package org.redkale.net.sncp;
import java.lang.annotation.Annotation;
import java.lang.reflect.*;
import java.net.*;
import java.nio.channels.CompletionHandler;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import org.redkale.mq.*;
import static org.redkale.net.sncp.Sncp.loadMethodActions;
import org.redkale.service.*;
import org.redkale.util.*;
/**
* 每个Service的client相关信息对象
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
* @param <T> Service泛型
*
* @since 2.8.0
*/
public final class SncpServiceInfo<T extends Service> {
protected final String name;
protected final Class<T> serviceType;
protected final T service;
protected final Uint128 serviceid;
protected final int serviceVersion;
protected final SncpServiceAction[] actions;
protected final String topic;
//MQ模式下此字段才有值
protected final MessageAgent messageAgent;
//MQ模式下此字段才有值
protected final SncpMessageClient messageClient;
//远程模式, 可能为null
protected Set<String> remoteGroups;
//远程模式, 可能为null
protected Set<InetSocketAddress> remoteAddresses;
SncpServiceInfo(String resourceName, Class<T> resourceServiceType, final T service, MessageAgent messageAgent, SncpMessageClient messageClient) {
this.name = resourceName;
this.serviceType = resourceServiceType;
this.serviceid = Sncp.serviceid(name, resourceServiceType);
this.service = service;
this.serviceVersion = 0;
this.messageAgent = messageAgent;
this.messageClient = messageAgent == null ? null : messageAgent.getSncpMessageClient();
this.topic = messageAgent == null ? null : messageAgent.generateSncpReqTopic(service);
final List<SncpServiceAction> serviceActions = new ArrayList<>();
final Class serviceImplClass = service.getClass();
for (Map.Entry<Uint128, Method> en : loadMethodActions(resourceServiceType).entrySet()) {
serviceActions.add(new SncpServiceAction(serviceImplClass, en.getValue(), serviceid, en.getKey()));
}
this.actions = serviceActions.toArray(new SncpServiceAction[serviceActions.size()]);
}
public void updateRemoteAddress(Set<String> remoteGroups, Set<InetSocketAddress> remoteAddresses) {
this.remoteGroups = remoteGroups;
this.remoteAddresses = remoteAddresses;
}
public String getName() {
return name;
}
public Class getServiceClass() {
return serviceType;
}
public T getService() {
return service;
}
public Uint128 getServiceid() {
return serviceid;
}
public int getServiceVersion() {
return serviceVersion;
}
public SncpServiceAction[] getActions() {
return actions;
}
public String getTopic() {
return topic;
}
public Set<String> getRemoteGroups() {
return remoteGroups;
}
public Set<InetSocketAddress> getRemoteAddresses() {
return remoteAddresses;
}
public static final class SncpServiceAction {
protected final Uint128 actionid;
protected final Method method;
protected final Type resultTypes; //void 必须设为 null
protected final Type[] paramTypes;
protected final Class[] paramClass;
protected final Attribute[] paramAttrs; // 为null表示无RpcCall处理，index=0固定为null, 其他为参数标记的RpcCall回调方法
protected final int handlerFuncParamIndex;
protected final int handlerAttachParamIndex;
protected final int addressTargetParamIndex;
protected final int addressSourceParamIndex;
protected final int topicTargetParamIndex;
protected final boolean boolReturnTypeFuture; // 返回结果类型是否为 CompletableFuture
protected final Creator<? extends CompletableFuture> futureCreator;
protected final SncpHeader header;
@SuppressWarnings("unchecked")
SncpServiceAction(final Class serviceImplClass, Method method, Uint128 serviceid, Uint128 actionid) {
this.actionid = actionid == null ? Sncp.actionid(method) : actionid;
Type rt = TypeToken.getGenericType(method.getGenericReturnType(), serviceImplClass);
this.resultTypes = rt == void.class ? null : rt;
this.boolReturnTypeFuture = CompletableFuture.class.isAssignableFrom(method.getReturnType());
this.futureCreator = boolReturnTypeFuture ? Creator.create((Class<? extends CompletableFuture>) method.getReturnType()) : null;
this.paramTypes = TypeToken.getGenericType(method.getGenericParameterTypes(), serviceImplClass);
this.paramClass = method.getParameterTypes();
this.method = method;
Annotation[][] anns = method.getParameterAnnotations();
int tpoicAddrIndex = -1;
int targetAddrIndex = -1;
int sourceAddrIndex = -1;
int handlerAttachIndex = -1;
int handlerFuncIndex = -1;
boolean hasattr = false;
Attribute[] atts = new Attribute[paramTypes.length + 1];
if (anns.length > 0) {
Class<?>[] params = method.getParameterTypes();
for (int i = 0; i < params.length; i++) {
if (CompletionHandler.class.isAssignableFrom(params[i])) {
if (boolReturnTypeFuture) {
throw new SncpException(method + " have both CompletionHandler and CompletableFuture");
}
if (handlerFuncIndex >= 0) {
throw new SncpException(method + " have more than one CompletionHandler type parameter");
}
Sncp.checkAsyncModifier(params[i], method);
handlerFuncIndex = i;
break;
}
}
for (int i = 0; i < anns.length; i++) {
if (anns[i].length > 0) {
for (Annotation ann : anns[i]) {
if (ann.annotationType() == RpcAttachment.class) {
if (handlerAttachIndex >= 0) {
throw new SncpException(method + " have more than one @RpcAttachment parameter");
}
handlerAttachIndex = i;
} else if (ann.annotationType() == RpcTargetAddress.class && SocketAddress.class.isAssignableFrom(params[i])) {
targetAddrIndex = i;
} else if (ann.annotationType() == RpcSourceAddress.class && SocketAddress.class.isAssignableFrom(params[i])) {
sourceAddrIndex = i;
} else if (ann.annotationType() == RpcTargetTopic.class && String.class.isAssignableFrom(params[i])) {
tpoicAddrIndex = i;
}
}
}
}
}
this.topicTargetParamIndex = tpoicAddrIndex;
this.addressTargetParamIndex = targetAddrIndex;
this.addressSourceParamIndex = sourceAddrIndex;
this.handlerFuncParamIndex = handlerFuncIndex;
this.handlerAttachParamIndex = handlerAttachIndex;
this.paramAttrs = hasattr ? atts : null;
this.header = new SncpHeader(null, serviceid, actionid);
if (this.handlerFuncParamIndex >= 0 && method.getReturnType() != void.class) {
throw new SncpException(method + " have CompletionHandler type parameter but return type is not void");
}
}
public String actionName() {
return method.getDeclaringClass().getSimpleName() + "." + method.getName();
}
@Override
public String toString() {
return "{" + actionid + "," + (method == null ? "null" : method.getName()) + "}";
}
}
}

2
src/test/java/org/redkale/test/sncp/SncpClientCodecTest.java
View File

@@ -32,7 +32,7 @@ public class SncpClientCodecTest {
InetSocketAddress sncpAddress = new InetSocketAddress("127.0.0.1", 3389);
InetSocketAddress remoteAddress = new InetSocketAddress("127.0.0.1", 3344);
final AsyncIOGroup asyncGroup = new AsyncIOGroup(8192, 16);
SncpClient client = new SncpClient("test", asyncGroup, sncpAddress, new ClientAddress(remoteAddress), Utility.cpus(), 16);
SncpClient client = new SncpClient("test", asyncGroup, sncpAddress, new ClientAddress(remoteAddress), "TCP", Utility.cpus(), 16);
SncpClientConnection conn = client.createClientConnection(1, asyncGroup.newTCPClientConnection());
SncpClientCodec codec = new SncpClientCodec(conn);
List respResults = new ArrayList();

4
src/test/java/org/redkale/test/sncp/SncpTest.java
View File

@@ -25,9 +25,9 @@ import org.redkale.util.*;
*/
public class SncpTest {
private static final String myhost = Utility.localInetAddress().getHostAddress();
private static final String myhost = "127.0.0.1";
private static int port = 0;
private static int port = 63877;
private static int port2 = 4240;

8
src/test/java/org/redkale/test/sncp/SncpTestServiceImpl.java
View File

@@ -11,7 +11,7 @@ import java.nio.channels.CompletionHandler;
import java.util.concurrent.CompletableFuture;
import org.redkale.annotation.ResourceType;
import org.redkale.net.*;
import org.redkale.net.sncp.*;
import org.redkale.net.sncp.Sncp;
import org.redkale.service.*;
import org.redkale.util.ResourceFactory;
@@ -89,7 +89,7 @@ public class SncpTestServiceImpl implements SncpTestIService {
System.out.println(method);
}
System.out.println("-----------------------------------");
for (Method method : SncpOldClient.parseMethodActions(service.getClass()).values()) {
for (Method method : Sncp.loadMethodActions(service.getClass()).values()) {
System.out.println(method);
}
System.out.println("-----------------------------------");
@@ -98,7 +98,7 @@ public class SncpTestServiceImpl implements SncpTestIService {
System.out.println(method);
}
System.out.println("-----------------------------------");
for (Method method : SncpOldClient.parseMethodActions(service.getClass()).values()) {
for (Method method : Sncp.loadMethodActions(service.getClass()).values()) {
System.out.println(method);
}
System.out.println("-----------------------------------");
@@ -107,7 +107,7 @@ public class SncpTestServiceImpl implements SncpTestIService {
System.out.println(method);
}
System.out.println("-----------------------------------");
for (Method method : SncpOldClient.parseMethodActions(service.getClass()).values()) {
for (Method method : Sncp.loadMethodActions(service.getClass()).values()) {
System.out.println(method);
}
System.out.println("-----------------------------------");