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

Browse Source
This commit is contained in:
Redkale
2017-05-29 14:03:11 +08:00
parent 0da74be2fa
commit 06773ccdc0
9 changed files with 599 additions and 682 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
src/org/redkale/boot/Application.java
View File

@@ -5,6 +5,7 @@
*/ */
package org.redkale.boot; package org.redkale.boot;
import org.redkale.net.TransportGroupInfo;
import java.io.*; import java.io.*;
import java.lang.reflect.*; import java.lang.reflect.*;
import java.net.*; import java.net.*;
@@ -89,12 +90,6 @@ public final class Application {
*/ */
public static final String RESNAME_SERVER_ROOT = Server.RESNAME_SERVER_ROOT; public static final String RESNAME_SERVER_ROOT = Server.RESNAME_SERVER_ROOT;
//每个地址对应的Group名
final Map<InetSocketAddress, String> globalNodes = new HashMap<>();
//协议地址的Group集合
final Map<String, GroupInfo> globalGroups = new HashMap<>();
//本地IP地址 //本地IP地址
final InetAddress localAddress; final InetAddress localAddress;
@@ -107,14 +102,8 @@ public final class Application {
//NodeServer 资源 //NodeServer 资源
final List<NodeServer> servers = new CopyOnWriteArrayList<>(); final List<NodeServer> servers = new CopyOnWriteArrayList<>();
//传输端的ByteBuffer对象池 //传输端的TransportFactory
final ObjectPool<ByteBuffer> transportBufferPool; final TransportFactory transportFactory;
//传输端的线程池
final ExecutorService transportExecutor;
//传输端的ChannelGroup
final AsynchronousChannelGroup transportChannelGroup;
//全局根ResourceFactory //全局根ResourceFactory
final ResourceFactory resourceFactory = ResourceFactory.root(); final ResourceFactory resourceFactory = ResourceFactory.root();
@@ -271,18 +260,21 @@ public final class Application {
logger.log(Level.INFO, Transport.class.getSimpleName() + " configure bufferCapacity = " + bufferCapacity + "; bufferPoolSize = " + bufferPoolSize + "; threads = " + threads + ";"); logger.log(Level.INFO, Transport.class.getSimpleName() + " configure bufferCapacity = " + bufferCapacity + "; bufferPoolSize = " + bufferPoolSize + "; threads = " + threads + ";");
} }
} }
this.transportBufferPool = transportPool; this.transportFactory = new TransportFactory(transportExec, transportPool, transportGroup);
this.transportExecutor = transportExec;
this.transportChannelGroup = transportGroup;
} }
public ResourceFactory getResourceFactory() { public ResourceFactory getResourceFactory() {
return resourceFactory; return resourceFactory;
} }
public TransportFactory getTransportFactory() {
return transportFactory;
}
// public WatchFactory getWatchFactory() { // public WatchFactory getWatchFactory() {
// return watchFactory; // return watchFactory;
// } // }
public List<NodeServer> getNodeServers() { public List<NodeServer> getNodeServers() {
return new ArrayList<>(servers); return new ArrayList<>(servers);
} }
@@ -421,25 +413,18 @@ public final class Application {
final AnyValue resources = config.getAnyValue("resources"); final AnyValue resources = config.getAnyValue("resources");
if (resources != null) { if (resources != null) {
//------------------------------------------------------------------------ //------------------------------------------------------------------------
for (AnyValue conf : resources.getAnyValues("group")) { for (AnyValue conf : resources.getAnyValues("group")) {
final String group = conf.getValue("name", ""); final String group = conf.getValue("name", "");
final String protocol = conf.getValue("protocol", Transport.DEFAULT_PROTOCOL).toUpperCase(); final String protocol = conf.getValue("protocol", Transport.DEFAULT_PROTOCOL).toUpperCase();
if (!"TCP".equalsIgnoreCase(protocol) && !"UDP".equalsIgnoreCase(protocol)) { if (!"TCP".equalsIgnoreCase(protocol) && !"UDP".equalsIgnoreCase(protocol)) {
throw new RuntimeException("Not supported Transport Protocol " + conf.getValue("protocol")); throw new RuntimeException("Not supported Transport Protocol " + conf.getValue("protocol"));
} }
GroupInfo ginfo = globalGroups.get(group); TransportGroupInfo ginfo = new TransportGroupInfo(group, protocol, conf.getValue("subprotocol", ""), new LinkedHashSet<>());
if (ginfo == null) {
ginfo = new GroupInfo(group, protocol, conf.getValue("subprotocol", ""), new LinkedHashSet<>());
globalGroups.put(group, ginfo);
}
for (AnyValue node : conf.getAnyValues("node")) { for (AnyValue node : conf.getAnyValues("node")) {
final InetSocketAddress addr = new InetSocketAddress(node.getValue("addr"), node.getIntValue("port")); final InetSocketAddress addr = new InetSocketAddress(node.getValue("addr"), node.getIntValue("port"));
ginfo.addrs.add(addr); ginfo.putAddress(addr);
String oldgroup = globalNodes.get(addr);
if (oldgroup != null) throw new RuntimeException(addr + " had one more group " + (globalNodes.get(addr)));
globalNodes.put(addr, group);
} }
transportFactory.addGroupInfo(ginfo);
} }
} }
//------------------------------------------------------------------------ //------------------------------------------------------------------------
@@ -696,17 +681,6 @@ public final class Application {
System.exit(0); System.exit(0);
} }
Set<String> findSncpGroups(Transport sameGroupTransport, Collection<Transport> diffGroupTransports) {
Set<String> gs = new HashSet<>();
if (sameGroupTransport != null) gs.add(sameGroupTransport.getName());
if (diffGroupTransports != null) {
for (Transport t : diffGroupTransports) {
gs.add(t.getName());
}
}
return gs;
}
NodeSncpServer findNodeSncpServer(final InetSocketAddress sncpAddr) { NodeSncpServer findNodeSncpServer(final InetSocketAddress sncpAddr) {
for (NodeServer node : servers) { for (NodeServer node : servers) {
if (node.isSNCP() && sncpAddr.equals(node.getSncpAddress())) { if (node.isSNCP() && sncpAddr.equals(node.getSncpAddress())) {
@@ -716,11 +690,6 @@ public final class Application {
return null; return null;
} }
GroupInfo findGroupInfo(String group) {
if (group == null) return null;
return globalGroups.get(group);
}
private void shutdown() throws Exception { private void shutdown() throws Exception {
servers.stream().forEach((server) -> { servers.stream().forEach((server) -> {
try { try {
@@ -748,13 +717,7 @@ public final class Application {
logger.log(Level.FINER, source.getClass() + " close CacheSource erroneous", e); logger.log(Level.FINER, source.getClass() + " close CacheSource erroneous", e);
} }
} }
if (this.transportChannelGroup != null) { this.transportFactory.shutdownNow();
try {
this.transportChannelGroup.shutdownNow();
} catch (Exception e) {
logger.log(Level.FINER, "close transportChannelGroup erroneous", e);
}
}
} }
private static AnyValue load(final InputStream in0) { private static AnyValue load(final InputStream in0) {

80
src/org/redkale/boot/GroupInfo.java
View File

@@ -1,80 +0,0 @@
/*
* To change this license header, choose License Headers in Project Properties.
* To change this template file, choose Tools | Templates
* and open the template in the editor.
*/
package org.redkale.boot;
import java.net.InetSocketAddress;
import java.util.*;
import org.redkale.convert.json.JsonConvert;
/**
* 协议地址组合对象, 对应application.xml 中 resources-&#62;group 节点信息
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*/
public class GroupInfo {
protected String name; //地址
protected String protocol; //协议 取值范围: TCP、UDP
protected String subprotocol; //子协议，预留使用
protected Set<InetSocketAddress> addrs; //地址列表， 对应 resources-&#62;group-&#62;node节点信息
public GroupInfo() {
}
public GroupInfo(String name, String protocol, String subprotocol, Set<InetSocketAddress> addrs) {
this.name = name;
this.protocol = protocol;
this.subprotocol = subprotocol;
this.addrs = addrs;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getProtocol() {
return protocol;
}
public void setProtocol(String protocol) {
this.protocol = protocol;
}
public String getSubprotocol() {
return subprotocol;
}
public void setSubprotocol(String subprotocol) {
this.subprotocol = subprotocol;
}
public Set<InetSocketAddress> getAddrs() {
return addrs;
}
public Set<InetSocketAddress> copyAddrs() {
return addrs == null ? null : new LinkedHashSet<>(addrs);
}
public void setAddrs(Set<InetSocketAddress> addrs) {
this.addrs = addrs;
}
@Override
public String toString() {
return JsonConvert.root().convertTo(this);
}
}

2
src/org/redkale/boot/NodeHttpServer.java
View File

@@ -110,7 +110,7 @@ public class NodeHttpServer extends NodeServer {
synchronized (regFactory) { synchronized (regFactory) {
Service nodeService = (Service) rf.find(resourceName, WebSocketNode.class); Service nodeService = (Service) rf.find(resourceName, WebSocketNode.class);
if (nodeService == null) { if (nodeService == null) {
nodeService = Sncp.createLocalService(resourceName, getExecutor(), application.getResourceFactory(), WebSocketNodeService.class, (InetSocketAddress) null, (String) null, (Set<String>) null, (AnyValue) null, (Transport) null, (Collection<Transport>) null); nodeService = Sncp.createLocalService(resourceName, WebSocketNodeService.class, application.getResourceFactory(), application.getTransportFactory(), (InetSocketAddress) null, (Set<String>) null, (AnyValue) null);
regFactory.register(resourceName, WebSocketNode.class, nodeService); regFactory.register(resourceName, WebSocketNode.class, nodeService);
} }
resourceFactory.inject(nodeService, self); resourceFactory.inject(nodeService, self);

145
src/org/redkale/boot/NodeServer.java
View File

@@ -14,7 +14,6 @@ import java.util.*;
import java.util.concurrent.*; import java.util.concurrent.*;
import java.util.function.*; import java.util.function.*;
import java.util.logging.*; import java.util.logging.*;
import java.util.stream.Collectors;
import javax.annotation.Resource; import javax.annotation.Resource;
import javax.persistence.Transient; import javax.persistence.Transient;
import static org.redkale.boot.Application.*; import static org.redkale.boot.Application.*;
@@ -129,7 +128,7 @@ public abstract class NodeServer {
if (isSNCP()) { // SNCP协议 if (isSNCP()) { // SNCP协议
String host = this.serverConf.getValue("host", "0.0.0.0").replace("0.0.0.0", ""); String host = this.serverConf.getValue("host", "0.0.0.0").replace("0.0.0.0", "");
this.sncpAddress = new InetSocketAddress(host.isEmpty() ? application.localAddress.getHostAddress() : host, this.serverConf.getIntValue("port")); this.sncpAddress = new InetSocketAddress(host.isEmpty() ? application.localAddress.getHostAddress() : host, this.serverConf.getIntValue("port"));
this.sncpGroup = application.globalNodes.get(this.sncpAddress); this.sncpGroup = application.transportFactory.findGroupName(this.sncpAddress);
//单向SNCP服务不需要对等group //单向SNCP服务不需要对等group
//if (this.sncpGroup == null) throw new RuntimeException("Server (" + String.valueOf(config).replaceAll("\\s+", " ") + ") not found <group> info"); //if (this.sncpGroup == null) throw new RuntimeException("Server (" + String.valueOf(config).replaceAll("\\s+", " ") + ") not found <group> info");
} }
@@ -189,6 +188,7 @@ public abstract class NodeServer {
final NodeServer self = this; final NodeServer self = this;
//--------------------------------------------------------------------------------------------- //---------------------------------------------------------------------------------------------
final ResourceFactory appResFactory = application.getResourceFactory(); final ResourceFactory appResFactory = application.getResourceFactory();
final TransportFactory appTranFactory = application.getTransportFactory();
final AnyValue resources = application.config.getAnyValue("resources"); final AnyValue resources = application.config.getAnyValue("resources");
final Map<String, AnyValue> cacheResource = new HashMap<>(); final Map<String, AnyValue> cacheResource = new HashMap<>();
//final Map<String, AnyValue> dataResources = new HashMap<>(); //final Map<String, AnyValue> dataResources = new HashMap<>();
@@ -245,13 +245,13 @@ public abstract class NodeServer {
appResFactory.register(resourceName, DataSource.class, source); appResFactory.register(resourceName, DataSource.class, source);
SncpClient client = Sncp.getSncpClient((Service) src); SncpClient client = Sncp.getSncpClient((Service) src);
Transport sameGroupTransport = Sncp.getSameGroupTransport((Service) src);
List<Transport> diffGroupTransports = Arrays.asList(Sncp.getDiffGroupTransports((Service) src));
final InetSocketAddress sncpAddr = client == null ? null : client.getClientAddress(); final InetSocketAddress sncpAddr = client == null ? null : client.getClientAddress();
if ((src instanceof DataSource) && sncpAddr != null && resourceFactory.find(resourceName, DataCacheListener.class) == null) { //只有DataSourceService 才能赋值 DataCacheListener if ((src instanceof DataSource) && sncpAddr != null && resourceFactory.find(resourceName, DataCacheListener.class) == null) { //只有DataSourceService 才能赋值 DataCacheListener
final NodeSncpServer sncpServer = application.findNodeSncpServer(sncpAddr); final NodeSncpServer sncpServer = application.findNodeSncpServer(sncpAddr);
Set<String> gs = application.findSncpGroups(sameGroupTransport, diffGroupTransports); final Set<String> groups = new HashSet<>();
Service cacheListenerService = Sncp.createLocalService(resourceName, getExecutor(), appResFactory, DataCacheListenerService.class, sncpAddr, sncpServer.getSncpGroup(), gs, Sncp.getConf((Service) src), sameGroupTransport, diffGroupTransports); if (client != null && client.getSameGroup() != null) groups.add(client.getSameGroup());
if (client != null && client.getDiffGroups() != null) groups.addAll(client.getDiffGroups());
Service cacheListenerService = Sncp.createLocalService(resourceName, DataCacheListenerService.class, appResFactory, appTranFactory, sncpAddr, groups, Sncp.getConf((Service) src));
appResFactory.register(resourceName, DataCacheListener.class, cacheListenerService); appResFactory.register(resourceName, DataCacheListener.class, cacheListenerService);
localServices.add(cacheListenerService); localServices.add(cacheListenerService);
sncpServer.consumerAccept(cacheListenerService); sncpServer.consumerAccept(cacheListenerService);
@@ -275,15 +275,13 @@ public abstract class NodeServer {
if ((src instanceof Service) && Sncp.isRemote((Service) src)) return; //远程模式不需要注入 CacheSource if ((src instanceof Service) && Sncp.isRemote((Service) src)) return; //远程模式不需要注入 CacheSource
final Service srcService = (Service) src; final Service srcService = (Service) src;
SncpClient client = Sncp.getSncpClient(srcService); SncpClient client = Sncp.getSncpClient(srcService);
Transport sameGroupTransport = Sncp.getSameGroupTransport(srcService);
Transport[] dts = Sncp.getDiffGroupTransports((Service) src);
List<Transport> diffGroupTransports = dts == null ? new ArrayList<>() : Arrays.asList(dts);
final InetSocketAddress sncpAddr = client == null ? null : client.getClientAddress(); final InetSocketAddress sncpAddr = client == null ? null : client.getClientAddress();
final AnyValue sourceConf = cacheResource.get(resourceName); final AnyValue sourceConf = cacheResource.get(resourceName);
final Class sourceType = sourceConf == null ? CacheMemorySource.class : Class.forName(sourceConf.getValue("type")); final Class sourceType = sourceConf == null ? CacheMemorySource.class : Class.forName(sourceConf.getValue("type"));
@SuppressWarnings("unchecked") final Set<String> groups = new HashSet<>();
final CacheSource source = (CacheSource) Sncp.createLocalService(resourceName, getExecutor(), appResFactory, (Class<? extends Service>) sourceType, if (client != null && client.getSameGroup() != null) groups.add(client.getSameGroup());
sncpAddr, Sncp.getSncpGroup(srcService), Sncp.getGroups(srcService), Sncp.getConf(srcService), sameGroupTransport, diffGroupTransports); if (client != null && client.getDiffGroups() != null) groups.addAll(client.getDiffGroups());
final CacheSource source = (CacheSource) Sncp.createLocalService(resourceName, sourceType, appResFactory, appTranFactory, sncpAddr, groups, Sncp.getConf(srcService));
Type genericType = field.getGenericType(); Type genericType = field.getGenericType();
ParameterizedType pt = (genericType instanceof ParameterizedType) ? (ParameterizedType) genericType : null; ParameterizedType pt = (genericType instanceof ParameterizedType) ? (ParameterizedType) genericType : null;
Type valType = pt == null ? null : pt.getActualTypeArguments()[1]; Type valType = pt == null ? null : pt.getActualTypeArguments()[1];
@@ -301,7 +299,6 @@ public abstract class NodeServer {
if ((src instanceof WebSocketNodeService) && sncpAddr != null) { //只有WebSocketNodeService的服务才需要给SNCP服务注入CacheMemorySource if ((src instanceof WebSocketNodeService) && sncpAddr != null) { //只有WebSocketNodeService的服务才需要给SNCP服务注入CacheMemorySource
NodeSncpServer sncpServer = application.findNodeSncpServer(sncpAddr); NodeSncpServer sncpServer = application.findNodeSncpServer(sncpAddr);
Set<String> gs = application.findSncpGroups(sameGroupTransport, diffGroupTransports);
sncpServer.getSncpServer().addSncpServlet((Service) source); sncpServer.getSncpServer().addSncpServlet((Service) source);
//logger.info("[" + Thread.currentThread().getName() + "] Load Service " + source); //logger.info("[" + Thread.currentThread().getName() + "] Load Service " + source);
} }
@@ -323,7 +320,8 @@ public abstract class NodeServer {
final String threadName = "[" + Thread.currentThread().getName() + "] "; final String threadName = "[" + Thread.currentThread().getName() + "] ";
final Set<FilterEntry<? extends Service>> entrys = (Set) serviceFilter.getAllFilterEntrys(); final Set<FilterEntry<? extends Service>> entrys = (Set) serviceFilter.getAllFilterEntrys();
ResourceFactory regFactory = isSNCP() ? application.getResourceFactory() : resourceFactory; ResourceFactory regFactory = isSNCP() ? application.getResourceFactory() : resourceFactory;
final ResourceFactory appResourceFactory = application.getResourceFactory();
final TransportFactory appTransportFactory = application.getTransportFactory();
for (FilterEntry<? extends Service> entry : entrys) { //service实现类 for (FilterEntry<? extends Service> entry : entrys) { //service实现类
final Class<? extends Service> serviceImplClass = entry.getType(); final Class<? extends Service> serviceImplClass = entry.getType();
if (Modifier.isFinal(serviceImplClass.getModifiers())) continue; //修饰final的类跳过 if (Modifier.isFinal(serviceImplClass.getModifiers())) continue; //修饰final的类跳过
@@ -354,13 +352,12 @@ public abstract class NodeServer {
Service service; Service service;
boolean ws = src instanceof WebSocketServlet; boolean ws = src instanceof WebSocketServlet;
if (ws || localed) { //本地模式 if (ws || localed) { //本地模式
service = Sncp.createLocalService(resourceName, getExecutor(), application.getResourceFactory(), serviceImplClass, service = Sncp.createLocalService(resourceName, serviceImplClass, appResourceFactory, appTransportFactory, NodeServer.this.sncpAddress, groups, entry.getProperty());
NodeServer.this.sncpAddress, NodeServer.this.sncpGroup, groups, entry.getProperty(), loadTransport(NodeServer.this.sncpGroup), loadTransports(groups));
} else { } else {
service = Sncp.createRemoteService(resourceName, getExecutor(), serviceImplClass, NodeServer.this.sncpAddress, null, groups, entry.getProperty(), loadTransport(groups)); service = Sncp.createRemoteService(resourceName, serviceImplClass, appTransportFactory, NodeServer.this.sncpAddress, groups, entry.getProperty());
} }
if (SncpClient.parseMethod(serviceImplClass).isEmpty()) return; //class没有可用的方法， 通常为BaseService if (SncpClient.parseMethod(serviceImplClass).isEmpty()) return; //class没有可用的方法， 通常为BaseService
//final ServiceWrapper wrapper = new ServiceWrapper(serviceImplClass, service, resourceName, localed ? NodeServer.this.sncpGroup : null, groups, entry.getProperty());
final Class restype = Sncp.getResourceType(service); final Class restype = Sncp.getResourceType(service);
if (rf.find(resourceName, restype) == null) { if (rf.find(resourceName, restype) == null) {
regFactory.register(resourceName, restype, service); regFactory.register(resourceName, restype, service);
@@ -447,62 +444,62 @@ public abstract class NodeServer {
maxClassNameLength = Math.max(maxClassNameLength, Sncp.getResourceType(y).getName().length() + 1); maxClassNameLength = Math.max(maxClassNameLength, Sncp.getResourceType(y).getName().length() + 1);
} }
protected List<Transport> loadTransports(final HashSet<String> groups) { /*
if (groups == null) return null; * protected List<Transport> loadTransports(final HashSet<String> groups) {
final List<Transport> transports = new ArrayList<>(); * if (groups == null) return null;
for (String group : groups) { * final List<Transport> transports = new ArrayList<>();
if (this.sncpGroup == null || !this.sncpGroup.equals(group)) { * for (String group : groups) {
transports.add(loadTransport(group)); * if (this.sncpGroup == null || !this.sncpGroup.equals(group)) {
} * transports.add(loadTransport(group));
} * }
return transports; * }
} * return transports;
* }
protected Transport loadTransport(final HashSet<String> groups) { * protected Transport loadTransport(final HashSet<String> groups) {
if (groups == null || groups.isEmpty()) return null; * if (groups == null || groups.isEmpty()) return null;
final String groupid = new ArrayList<>(groups).stream().sorted().collect(Collectors.joining(";")); //按字母排列顺序 * final String groupid = new ArrayList<>(groups).stream().sorted().collect(Collectors.joining(";")); //按字母排列顺序
Transport transport = application.resourceFactory.find(groupid, Transport.class); * Transport transport = application.resourceFactory.find(groupid, Transport.class);
if (transport != null) return transport; * if (transport != null) return transport;
final List<Transport> transports = new ArrayList<>(); * final List<Transport> transports = new ArrayList<>();
for (String group : groups) { * for (String group : groups) {
transports.add(loadTransport(group)); * transports.add(loadTransport(group));
} * }
Set<InetSocketAddress> addrs = new HashSet(); * Set<InetSocketAddress> addrs = new HashSet();
transports.forEach(t -> addrs.addAll(Arrays.asList(t.getRemoteAddresses()))); * transports.forEach(t -> addrs.addAll(Arrays.asList(t.getRemoteAddresses())));
Transport first = transports.get(0); * Transport first = transports.get(0);
GroupInfo ginfo = application.findGroupInfo(first.getName()); * TransportGroupInfo ginfo = application.transportFactory.findGroupInfo(first.getName());
Transport newTransport = new Transport(groupid, ginfo.getProtocol(), * Transport newTransport = new Transport(groupid, ginfo.getProtocol(),
ginfo.getSubprotocol(), application.transportBufferPool, application.transportChannelGroup, this.sncpAddress, addrs); * ginfo.getSubprotocol(), application.transportBufferPool, application.transportChannelGroup, this.sncpAddress, addrs);
synchronized (application.resourceFactory) { * synchronized (application.resourceFactory) {
transport = application.resourceFactory.find(groupid, Transport.class); * transport = application.resourceFactory.find(groupid, Transport.class);
if (transport == null) { * if (transport == null) {
transport = newTransport; * transport = newTransport;
application.resourceFactory.register(groupid, transport); * application.resourceFactory.register(groupid, transport);
} * }
} * }
return transport; * return transport;
} * }
*
protected Transport loadTransport(final String group) { * protected Transport loadTransport(final String group) {
if (group == null) return null; * if (group == null) return null;
Transport transport; * Transport transport;
synchronized (application.resourceFactory) { * synchronized (application.resourceFactory) {
transport = application.resourceFactory.find(group, Transport.class); * transport = application.resourceFactory.find(group, Transport.class);
if (transport != null) { * if (transport != null) {
if (this.sncpAddress != null && !this.sncpAddress.equals(transport.getClientAddress())) { * if (this.sncpAddress != null && !this.sncpAddress.equals(transport.getClientAddress())) {
throw new RuntimeException(transport + "repeat create on newClientAddress = " + this.sncpAddress + ", oldClientAddress = " + transport.getClientAddress()); * throw new RuntimeException(transport + "repeat create on newClientAddress = " + this.sncpAddress + ", oldClientAddress = " + transport.getClientAddress());
} * }
return transport; * return transport;
} * }
GroupInfo ginfo = application.findGroupInfo(group); * TransportGroupInfo ginfo = application.findGroupInfo(group);
Set<InetSocketAddress> addrs = ginfo.copyAddrs(); * Set<InetSocketAddress> addrs = ginfo.copyAddresses();
if (addrs == null) throw new RuntimeException("Not found <group> = " + group + " on <resources> "); * if (addrs == null) throw new RuntimeException("Not found <group> = " + group + " on <resources> ");
transport = new Transport(group, ginfo.getProtocol(), ginfo.getSubprotocol(), application.transportBufferPool, application.transportChannelGroup, this.sncpAddress, addrs); * transport = new Transport(group, ginfo.getProtocol(), ginfo.getSubprotocol(), application.transportBufferPool, application.transportChannelGroup, this.sncpAddress, addrs);
application.resourceFactory.register(group, transport); * application.resourceFactory.register(group, transport);
} * }
return transport; * return transport;
} * }
*/
protected abstract ClassFilter<Filter> createFilterClassFilter(); protected abstract ClassFilter<Filter> createFilterClassFilter();
protected abstract ClassFilter<Servlet> createServletClassFilter(); protected abstract ClassFilter<Servlet> createServletClassFilter();

135
src/org/redkale/net/TransportFactory.java Normal file
View File

@@ -0,0 +1,135 @@
/*
* To change this license header, choose License Headers in Project Properties.
* To change this template file, choose Tools | Templates
* and open the template in the editor.
*/
package org.redkale.net;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.logging.*;
import org.redkale.util.ObjectPool;
/**
*
* @author zhangjx
*/
public class TransportFactory {
protected static final Logger logger = Logger.getLogger(TransportFactory.class.getSimpleName());
//传输端的线程池
protected final ExecutorService executor;
//传输端的ByteBuffer对象池
protected final ObjectPool<ByteBuffer> bufferPool;
//传输端的ChannelGroup
protected final AsynchronousChannelGroup channelGroup;
//每个地址对应的Group名
protected final Map<InetSocketAddress, String> groupAddrs = new HashMap<>();
//协议地址的Group集合
protected final Map<String, TransportGroupInfo> groupInfos = new HashMap<>();
public TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup) {
this.executor = executor;
this.bufferPool = bufferPool;
this.channelGroup = channelGroup;
}
public String findGroupName(InetSocketAddress addr) {
if (addr == null) return null;
return groupAddrs.get(addr);
}
public TransportGroupInfo findGroupInfo2(String group) {
if (group == null) return null;
return groupInfos.get(group);
}
public boolean addGroupInfo(TransportGroupInfo info) {
if (info == null) throw new RuntimeException("TransportGroupInfo can not null");
if (info.addresses == null) throw new RuntimeException("TransportGroupInfo.addresses can not null");
if (!checkName(info.name)) throw new RuntimeException("Transport.group.name only 0-9 a-z A-Z _ cannot begin 0-9");
TransportGroupInfo old = groupInfos.get(info.name);
if (old != null && !old.protocol.equals(info.protocol)) throw new RuntimeException("Transport.group.name repeat but protocol is different");
if (old != null && !old.subprotocol.equals(info.subprotocol)) throw new RuntimeException("Transport.group.name repeat but subprotocol is different");
for (InetSocketAddress addr : info.addresses) {
if (!groupAddrs.getOrDefault(addr, info.name).equals(info.name)) throw new RuntimeException(addr + " repeat but different group.name");
}
if (old == null) {
groupInfos.put(info.name, info);
} else {
old.putAddress(info.addresses);
}
for (InetSocketAddress addr : info.addresses) {
groupAddrs.put(addr, info.name);
}
return true;
}
public Transport loadSameGroupTransport(InetSocketAddress sncpAddress) {
return loadTransport(groupAddrs.get(sncpAddress), sncpAddress);
}
public Transport[] loadDiffGroupTransports(InetSocketAddress sncpAddress, final Set<String> diffGroups) {
if (diffGroups == null) return null;
final String sncpGroup = groupAddrs.get(sncpAddress);
final List<Transport> transports = new ArrayList<>();
for (String group : diffGroups) {
if (sncpGroup == null || !sncpGroup.equals(group)) {
transports.add(loadTransport(group, sncpAddress));
}
}
return transports.toArray(new Transport[transports.size()]);
}
public Transport loadRemoteTransport(InetSocketAddress sncpAddress, final Set<String> groups) {
if (groups == null) return null;
Set<InetSocketAddress> addresses = new HashSet<>();
TransportGroupInfo info = null;
for (String group : groups) {
info = groupInfos.get(group);
if (info == null) continue;
addresses.addAll(info.addresses);
}
if (info == null) return null;
if (sncpAddress != null) addresses.remove(sncpAddress);
return new Transport("remotes", info.protocol, info.subprotocol, this.bufferPool, this.channelGroup, sncpAddress, addresses);
}
private Transport loadTransport(final String groupName, InetSocketAddress sncpAddress) {
if (groupName == null) return null;
TransportGroupInfo info = groupInfos.get(groupName);
if (info == null) return null;
return new Transport(groupName, info.protocol, info.subprotocol, this.bufferPool, this.channelGroup, sncpAddress, info.addresses);
}
public ExecutorService getExecutor() {
return executor;
}
public void shutdownNow() {
try {
this.channelGroup.shutdownNow();
} catch (Exception e) {
logger.log(Level.FINER, "close transportChannelGroup erroneous", e);
}
}
private static boolean checkName(String name) { //不能含特殊字符
if (name.isEmpty()) return false;
if (name.charAt(0) >= '0' && name.charAt(0) <= '9') return false;
for (char ch : name.toCharArray()) {
if (!((ch >= '0' && ch <= '9') || ch == '_' || (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z'))) { //不能含特殊字符
return false;
}
}
return true;
}
}

103
src/org/redkale/net/TransportGroupInfo.java Normal file
View File

@@ -0,0 +1,103 @@
/*
* To change this license header, choose License Headers in Project Properties.
* To change this template file, choose Tools | Templates
* and open the template in the editor.
*/
package org.redkale.net;
import java.net.InetSocketAddress;
import java.util.*;
import org.redkale.convert.json.JsonConvert;
/**
* 协议地址组合对象, 对应application.xml 中 resources-&#62;group 节点信息
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*/
public class TransportGroupInfo {
protected String name; //地址
protected String protocol; //协议 取值范围: TCP、UDP
protected String subprotocol; //子协议，预留使用
protected Set<InetSocketAddress> addresses; //地址列表， 对应 resources-&#62;group-&#62;node节点信息
public TransportGroupInfo() {
}
public TransportGroupInfo(String name, String protocol, String subprotocol, Set<InetSocketAddress> addrs) {
Objects.requireNonNull(name, "Transport.group.name can not null");
Objects.requireNonNull(protocol, "Transport.group.protocol can not null");
Objects.requireNonNull(subprotocol, "Transport.group.subprotocol can not null");
this.name = name;
this.protocol = protocol;
this.subprotocol = subprotocol;
this.addresses = addrs;
}
public String getName() {
return name;
}
public void setName(String name) {
Objects.requireNonNull(name, "Transport.group.name can not null");
this.name = name;
}
public String getProtocol() {
return protocol;
}
public void setProtocol(String protocol) {
Objects.requireNonNull(protocol, "Transport.group.protocol can not null");
this.protocol = protocol;
}
public String getSubprotocol() {
return subprotocol;
}
public void setSubprotocol(String subprotocol) {
Objects.requireNonNull(subprotocol, "Transport.group.subprotocol can not null");
this.subprotocol = subprotocol;
}
public Set<InetSocketAddress> getAddresses() {
return addresses;
}
public Set<InetSocketAddress> copyAddresses() {
return addresses == null ? null : new LinkedHashSet<>(addresses);
}
public void setAddresses(Set<InetSocketAddress> addresses) {
this.addresses = addresses;
}
public boolean containsAddress(InetSocketAddress addr) {
if (this.addresses == null) return false;
return this.addresses.contains(addr);
}
public void putAddress(InetSocketAddress addr) {
if (addr == null) return;
if (this.addresses == null) this.addresses = new HashSet<>();
this.addresses.add(addr);
}
public void putAddress(Set<InetSocketAddress> addrs) {
if (addrs == null) return;
if (this.addresses == null) this.addresses = new HashSet<>();
this.addresses.addAll(addrs);
}
@Override
public String toString() {
return JsonConvert.root().convertTo(this);
}
}

6
src/org/redkale/net/http/WebSocket.java
View File

@@ -184,7 +184,7 @@ public abstract class WebSocket<G extends Serializable, T> {
*/ */
CompletableFuture<Integer> sendPacket(WebSocketPacket packet) { CompletableFuture<Integer> sendPacket(WebSocketPacket packet) {
CompletableFuture<Integer> rs = this._runner.sendMessage(packet); CompletableFuture<Integer> rs = this._runner.sendMessage(packet);
if (_engine.finest) _engine.logger.finest("userid:" + userid() + " send websocket result is " + rs + " on " + this + " by message(" + packet + ")"); if (_engine.finest) _engine.logger.finest("userid:" + userid() + " send websocket message(" + packet + ")" + " on " + this);
return rs == null ? CompletableFuture.completedFuture(RETCODE_WSOCKET_CLOSED) : rs; return rs == null ? CompletableFuture.completedFuture(RETCODE_WSOCKET_CLOSED) : rs;
} }
@@ -204,7 +204,7 @@ public abstract class WebSocket<G extends Serializable, T> {
return ((CompletableFuture) message).thenCompose((json) -> _engine.node.sendMessage(json, last, userids)); return ((CompletableFuture) message).thenCompose((json) -> _engine.node.sendMessage(json, last, userids));
} }
CompletableFuture<Integer> rs = _engine.node.sendMessage(message, last, userids); CompletableFuture<Integer> rs = _engine.node.sendMessage(message, last, userids);
if (_engine.finest) _engine.logger.finest("userids:" + Arrays.toString(userids) + " send websocket result is " + rs + " on " + this + " by message(" + _jsonConvert.convertTo(message) + ")"); if (_engine.finest) _engine.logger.finest("userids:" + Arrays.toString(userids) + " send websocket message(" + _jsonConvert.convertTo(message) + ")");
return rs; return rs;
} }
@@ -233,7 +233,7 @@ public abstract class WebSocket<G extends Serializable, T> {
return ((CompletableFuture) message).thenCompose((json) -> _engine.node.broadcastMessage(json, last)); return ((CompletableFuture) message).thenCompose((json) -> _engine.node.broadcastMessage(json, last));
} }
CompletableFuture<Integer> rs = _engine.node.broadcastMessage(message, last); CompletableFuture<Integer> rs = _engine.node.broadcastMessage(message, last);
if (_engine.finest) _engine.logger.finest("broadcast send websocket result is " + rs + " on " + this + " by message(" + _jsonConvert.convertTo(message) + ")"); if (_engine.finest) _engine.logger.finest("broadcast send websocket message(" + _jsonConvert.convertTo(message) + ")");
return rs; return rs;
} }

380
src/org/redkale/net/sncp/Sncp.java
View File

@@ -10,15 +10,12 @@ import java.lang.reflect.*;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.security.*; import java.security.*;
import java.util.*; import java.util.*;
import java.util.function.Consumer;
import javax.annotation.Resource; import javax.annotation.Resource;
import static jdk.internal.org.objectweb.asm.ClassWriter.COMPUTE_FRAMES; import static jdk.internal.org.objectweb.asm.ClassWriter.COMPUTE_FRAMES;
import jdk.internal.org.objectweb.asm.*; import jdk.internal.org.objectweb.asm.*;
import static jdk.internal.org.objectweb.asm.Opcodes.*; import static jdk.internal.org.objectweb.asm.Opcodes.*;
import jdk.internal.org.objectweb.asm.Type; import jdk.internal.org.objectweb.asm.Type;
import org.redkale.convert.bson.BsonConvert; import org.redkale.net.*;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.Transport;
import org.redkale.net.sncp.SncpClient.SncpAction; import org.redkale.net.sncp.SncpClient.SncpAction;
import org.redkale.service.*; import org.redkale.service.*;
import org.redkale.util.*; import org.redkale.util.*;
@@ -99,14 +96,8 @@ public abstract class Sncp {
} }
public static Class getServiceType(Service service) { public static Class getServiceType(Service service) {
if (service == null) return null; ResourceType rt = service.getClass().getAnnotation(ResourceType.class);
try { return rt == null ? service.getClass() : rt.value();
Field ts = service.getClass().getDeclaredField(FIELDPREFIX + "_service_type");
ts.setAccessible(true);
return (Class) ts.get(service);
} catch (Exception e) {
throw new RuntimeException(service + " not found " + FIELDPREFIX + "_service_type");
}
} }
public static Class getResourceType(Service service) { public static Class getResourceType(Service service) {
@@ -115,28 +106,6 @@ public abstract class Sncp {
return type == null ? getServiceType(service) : type.value(); return type == null ? getServiceType(service) : type.value();
} }
public static String getSncpGroup(Service service) {
if (service == null) return null;
try {
Field ts = service.getClass().getDeclaredField(FIELDPREFIX + "_sncpGroup");
ts.setAccessible(true);
return (String) ts.get(service);
} catch (Exception e) {
throw new RuntimeException(service + " not found " + FIELDPREFIX + "_sncpGroup");
}
}
public static Set<String> getGroups(Service service) {
if (service == null) return null;
try {
Field ts = service.getClass().getDeclaredField(FIELDPREFIX + "_groups");
ts.setAccessible(true);
return (Set) ts.get(service);
} catch (Exception e) {
throw new RuntimeException(service + " not found " + FIELDPREFIX + "_groups");
}
}
public static AnyValue getConf(Service service) { public static AnyValue getConf(Service service) {
if (service == null) return null; if (service == null) return null;
try { try {
@@ -159,28 +128,6 @@ public abstract class Sncp {
} }
} }
public static Transport getSameGroupTransport(Service service) {
if (service == null) return null;
try {
Field ts = service.getClass().getDeclaredField(FIELDPREFIX + "_sameGroupTransport");
ts.setAccessible(true);
return (Transport) ts.get(service);
} catch (Exception e) {
throw new RuntimeException(service + " not found " + FIELDPREFIX + "_sameGroupTransport");
}
}
public static Transport[] getDiffGroupTransports(Service service) {
if (service == null) return null;
try {
Field ts = service.getClass().getDeclaredField(FIELDPREFIX + "_diffGroupTransports");
ts.setAccessible(true);
return (Transport[]) ts.get(service);
} catch (Exception e) {
throw new RuntimeException(service + " not found " + FIELDPREFIX + "_diffGroupTransports");
}
}
static void checkAsyncModifier(Class param, Method method) { static void checkAsyncModifier(Class param, Method method) {
if (param == AsyncHandler.class) return; if (param == AsyncHandler.class) return;
if (Modifier.isFinal(param.getModifiers())) { if (Modifier.isFinal(param.getModifiers())) {
@@ -339,13 +286,8 @@ public abstract class Sncp {
final String supDynName = serviceImplClass.getName().replace('.', '/'); final String supDynName = serviceImplClass.getName().replace('.', '/');
final String clientName = SncpClient.class.getName().replace('.', '/'); final String clientName = SncpClient.class.getName().replace('.', '/');
final String clientDesc = Type.getDescriptor(SncpClient.class); final String clientDesc = Type.getDescriptor(SncpClient.class);
final String bsonConvertDesc = Type.getDescriptor(BsonConvert.class);
final String jsonConvertDesc = Type.getDescriptor(JsonConvert.class);
final String stringDesc = Type.getDescriptor(String.class);
final String anyValueDesc = Type.getDescriptor(AnyValue.class); final String anyValueDesc = Type.getDescriptor(AnyValue.class);
final String sncpDynDesc = Type.getDescriptor(SncpDyn.class); final String sncpDynDesc = Type.getDescriptor(SncpDyn.class);
final String transportDesc = Type.getDescriptor(Transport.class);
final String transportsDesc = Type.getDescriptor(Transport[].class);
ClassLoader loader = Sncp.class.getClassLoader(); ClassLoader loader = Sncp.class.getClassLoader();
String newDynName = supDynName.substring(0, supDynName.lastIndexOf('/') + 1) + LOCALPREFIX + serviceImplClass.getSimpleName(); String newDynName = supDynName.substring(0, supDynName.lastIndexOf('/') + 1) + LOCALPREFIX + serviceImplClass.getSimpleName();
if (!name.isEmpty()) { if (!name.isEmpty()) {
@@ -389,59 +331,14 @@ public abstract class Sncp {
av0.visit("value", Type.getType(Type.getDescriptor(rty == null ? serviceImplClass : rty.value()))); av0.visit("value", Type.getType(Type.getDescriptor(rty == null ? serviceImplClass : rty.value())));
av0.visitEnd(); av0.visitEnd();
} }
{
fv = cw.visitField(ACC_PRIVATE + ACC_FINAL + ACC_STATIC, FIELDPREFIX + "_service_type", "Ljava/lang/Class;", null, null);
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_bsonConvert", bsonConvertDesc, null, null);
av0 = fv.visitAnnotation("Ljavax/annotation/Resource;", true);
av0.visitEnd();
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_jsonConvert", jsonConvertDesc, null, null);
av0 = fv.visitAnnotation("Ljavax/annotation/Resource;", true);
av0.visitEnd();
fv.visitEnd();
}
{ {
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_conf", anyValueDesc, null, null); fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_conf", anyValueDesc, null, null);
fv.visitEnd(); fv.visitEnd();
} }
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_sncpGroup", stringDesc, null, null);
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_groups", "Ljava/util/Set;", "Ljava/util/Set<Ljava/lang/String;>;", null);
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_sameGroupTransport", transportDesc, null, null);
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_diffGroupTransports", transportsDesc, null, null);
fv.visitEnd();
}
{ {
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_client", clientDesc, null, null); fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_client", clientDesc, null, null);
fv.visitEnd(); fv.visitEnd();
} }
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_selfstring", "Ljava/lang/String;", null, null);
fv.visitEnd();
}
{//静态构造函数
mv = new AsmMethodVisitor(cw.visitMethod(ACC_STATIC, "<clinit>", "()V", null, null));
mv.visitLdcInsn(Type.getType(Type.getDescriptor(serviceImplClass)));
mv.visitFieldInsn(PUTSTATIC, newDynName, FIELDPREFIX + "_service_type", "Ljava/lang/Class;");
mv.visitInsn(RETURN);
mv.visitMaxs(1, 0);
mv.visitEnd();
}
{ //构造函数 { //构造函数
mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null)); mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null));
//mv.setDebug(true); //mv.setDebug(true);
@@ -454,16 +351,18 @@ public abstract class Sncp {
{ // toString() { // toString()
mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "toString", "()Ljava/lang/String;", null, null)); mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "toString", "()Ljava/lang/String;", null, null));
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_selfstring", "Ljava/lang/String;"); mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
Label l1 = new Label(); Label l1 = new Label();
mv.visitJumpInsn(IFNONNULL, l1); mv.visitJumpInsn(IFNONNULL, l1);
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitMethodInsn(INVOKESPECIAL, "java/lang/Object", "toString", "()Ljava/lang/String;", false); mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/Object", "getClass", "()Ljava/lang/Class;", false);
mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/Class", "getName", "()Ljava/lang/String;", false);
Label l2 = new Label(); Label l2 = new Label();
mv.visitJumpInsn(GOTO, l2); mv.visitJumpInsn(GOTO, l2);
mv.visitLabel(l1); mv.visitLabel(l1);
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_selfstring", "Ljava/lang/String;"); mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, "toSimpleString", "()Ljava/lang/String;", false);
mv.visitLabel(l2); mv.visitLabel(l2);
mv.visitInsn(ARETURN); mv.visitInsn(ARETURN);
mv.visitMaxs(1, 1); mv.visitMaxs(1, 1);
@@ -639,14 +538,7 @@ public abstract class Sncp {
mv.visitVarInsn(ALOAD, 0);//调用 _client mv.visitVarInsn(ALOAD, 0);//调用 _client
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc); mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitVarInsn(ALOAD, 0); //传递 _bsonConvert final int preparams = 3; //调用selfrunnable之前的参数个数; _client
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_bsonConvert", bsonConvertDesc);
mv.visitVarInsn(ALOAD, 0); //传递 _jsonConvert
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_jsonConvert", jsonConvertDesc);
mv.visitVarInsn(ALOAD, 0); //传递 _sameGroupTransport
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_sameGroupTransport", transportDesc);
final int preparams = 3; //调用selfrunnable之前的参数个数; _client/_bsonConvert/_jsonConvert/_sameGroupTransport
if (index <= 5) { //第几个 SncpAction if (index <= 5) { //第几个 SncpAction
mv.visitInsn(ICONST_0 + index); mv.visitInsn(ICONST_0 + index);
@@ -706,7 +598,7 @@ public abstract class Sncp {
} }
mv.visitInsn(AASTORE); mv.visitInsn(AASTORE);
} }
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteSameGroup" : "remoteSameGroup", "(" + bsonConvertDesc + jsonConvertDesc + transportDesc + "I[Ljava/lang/Object;)V", false); mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteSameGroup" : "remoteSameGroup", "(I[Ljava/lang/Object;)V", false);
mv.visitLabel(sameLabel); mv.visitLabel(sameLabel);
//---------------------------- 调用diffrun --------------------------------- //---------------------------- 调用diffrun ---------------------------------
mv.visitVarInsn(ILOAD, 3); //读取 diffrunnable mv.visitVarInsn(ILOAD, 3); //读取 diffrunnable
@@ -715,12 +607,6 @@ public abstract class Sncp {
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc); mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_bsonConvert", bsonConvertDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_jsonConvert", jsonConvertDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_diffGroupTransports", transportsDesc);
if (index <= 5) { //第几个 SncpAction if (index <= 5) { //第几个 SncpAction
mv.visitInsn(ICONST_0 + index); mv.visitInsn(ICONST_0 + index);
@@ -780,7 +666,7 @@ public abstract class Sncp {
} }
mv.visitInsn(AASTORE); mv.visitInsn(AASTORE);
} }
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteDiffGroup" : "remoteDiffGroup", "(" + bsonConvertDesc + jsonConvertDesc + transportsDesc + "I[Ljava/lang/Object;)V", false); mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteDiffGroup" : "remoteDiffGroup", "(I[Ljava/lang/Object;)V", false);
mv.visitLabel(diffLabel); mv.visitLabel(diffLabel);
if (returnType == void.class) { if (returnType == void.class) {
@@ -864,46 +750,35 @@ public abstract class Sncp {
} }
} }
public static <T extends Service> T createSimpleLocalService(final String name, final Class<T> serviceImplClass, final InetSocketAddress clientSncpAddress, final Transport sameGroupTransport) {
return createLocalService(name, null, ResourceFactory.root(), serviceImplClass, clientSncpAddress, null, new HashSet<>(), null, sameGroupTransport, null);
}
/** /**
* *
* 创建本地模式Service实例 * 创建本地模式Service实例
* *
* @param <T> Service泛型 * @param <T> Service泛型
* @param name 资源名 * @param name 资源名
* @param executor 线程池 * @param serviceImplClass Service类
* @param resourceFactory 资源容器 * @param resourceFactory ResourceFactory
* @param serviceImplClass Service类 * @param transportFactory TransportFactory
* @param clientSncpAddress 本地IP地址 * @param clientSncpAddress 本地IP地址
* @param sncpGroup 自身的组节点名 可能为null * @param groups 所有的组节点，包含自身
* @param groups 所有的组节点，包含自身 * @param conf 启动配置项
* @param conf 启动配置项
* @param sameGroupTransport 同组的通信组件
* @param diffGroupTransports 异组的通信组件列表
* *
* @return Service的本地模式实例 * @return Service的本地模式实例
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public static <T extends Service> T createLocalService( public static <T extends Service> T createLocalService(
final String name, final String name,
final Consumer<Runnable> executor,
final ResourceFactory resourceFactory,
final Class<T> serviceImplClass, final Class<T> serviceImplClass,
final ResourceFactory resourceFactory,
final TransportFactory transportFactory,
final InetSocketAddress clientSncpAddress, final InetSocketAddress clientSncpAddress,
final String sncpGroup,
final Set<String> groups, final Set<String> groups,
final AnyValue conf, final AnyValue conf) {
final Transport sameGroupTransport,
final Collection<Transport> diffGroupTransports) {
try { try {
final Class newClazz = createLocalServiceClass(name, serviceImplClass); final Class newClazz = createLocalServiceClass(name, serviceImplClass);
T rs = (T) newClazz.newInstance(); T rs = (T) newClazz.newInstance();
//-------------------------------------- //--------------------------------------
Service remoteService = null; Service remoteService = null;
Transport remoteTransport = null;
{ {
Class loop = newClazz; Class loop = newClazz;
do { do {
@@ -913,26 +788,7 @@ public abstract class Sncp {
if (field.getAnnotation(RpcRemote.class) == null) continue; if (field.getAnnotation(RpcRemote.class) == null) continue;
if (!field.getType().isAssignableFrom(newClazz)) continue; if (!field.getType().isAssignableFrom(newClazz)) continue;
field.setAccessible(true); field.setAccessible(true);
if (remoteTransport == null) { if (remoteService == null) remoteService = createRemoteService(name, serviceImplClass, transportFactory, clientSncpAddress, groups, conf);
List<Transport> list = new ArrayList<>();
if (sameGroupTransport != null) list.add(sameGroupTransport);
if (diffGroupTransports != null) list.addAll(diffGroupTransports);
if (!list.isEmpty()) {
Transport tmp = new Transport(list);
synchronized (resourceFactory) {
Transport old = resourceFactory.find(tmp.getName(), Transport.class);
if (old != null) {
remoteTransport = old;
} else {
remoteTransport = tmp;
resourceFactory.register(tmp.getName(), tmp);
}
}
}
}
if (remoteService == null && remoteTransport != null) {
remoteService = createRemoteService(name, executor, serviceImplClass, clientSncpAddress, sncpGroup, groups, conf, remoteTransport);
}
if (remoteService != null) field.set(rs, remoteService); if (remoteService != null) field.set(rs, remoteService);
} }
} while ((loop = loop.getSuperclass()) != Object.class); } while ((loop = loop.getSuperclass()) != Object.class);
@@ -942,80 +798,31 @@ public abstract class Sncp {
try { try {
Field e = newClazz.getDeclaredField(FIELDPREFIX + "_client"); Field e = newClazz.getDeclaredField(FIELDPREFIX + "_client");
e.setAccessible(true); e.setAccessible(true);
client = new SncpClient(name, serviceImplClass, rs, executor, false, newClazz, clientSncpAddress); client = new SncpClient(name, serviceImplClass, rs, transportFactory, false, newClazz, clientSncpAddress);
Set<String> diffGroups = groups == null ? new HashSet<>() : new HashSet<>(groups);
String sameGroup = transportFactory.findGroupName(clientSncpAddress);
if (sameGroup != null) diffGroups.remove(sameGroup);
client.setSameGroup(sameGroup);
client.setDiffGroups(diffGroups);
client.setSameGroupTransport(transportFactory.loadSameGroupTransport(clientSncpAddress));
client.setDiffGroupTransports(transportFactory.loadDiffGroupTransports(clientSncpAddress, diffGroups));
e.set(rs, client); e.set(rs, client);
} catch (NoSuchFieldException ne) { } catch (NoSuchFieldException ne) {
ne.printStackTrace();
} }
} }
{
StringBuilder sb = new StringBuilder();
sb.append(newClazz.getName()).append("{name = '").append(name).append("'");
if (client != null) {
sb.append(", serviceid = ").append(client.getServiceid());
sb.append(", serviceversion = ").append(client.getServiceversion());
sb.append(", action.size = ").append(client.getActionCount());
// List<String> groups = new ArrayList<>();
// if (sameGroupTransport != null) groups.add(sameGroupTransport.getName());
// if (diffGroupTransports != null) {
// for (Transport t : diffGroupTransports) {
// groups.add(t.getName());
// }
// }
sb.append(", address = ").append(clientSncpAddress).append(", groups = ").append(groups);
sb.append(", sameaddrs = ").append(sameGroupTransport == null ? null : Arrays.asList(sameGroupTransport.getRemoteAddresses()));
List<InetSocketAddress> addrs = new ArrayList<>();
if (diffGroupTransports != null) {
for (Transport t : diffGroupTransports) {
addrs.addAll(Arrays.asList(t.getRemoteAddresses()));
}
}
sb.append(", diffaddrs = ").append(addrs);
} else {
sb.append(", ").append(RpcMultiRun.class.getSimpleName().toLowerCase()).append(" = false");
}
sb.append("}");
Field s = newClazz.getDeclaredField(FIELDPREFIX + "_selfstring");
s.setAccessible(true);
s.set(rs, sb.toString());
}
if (client == null) return rs; if (client == null) return rs;
{
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_sncpGroup");
c.setAccessible(true);
c.set(rs, sncpGroup);
}
{
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_groups");
c.setAccessible(true);
c.set(rs, groups);
}
{ {
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_conf"); Field c = newClazz.getDeclaredField(FIELDPREFIX + "_conf");
c.setAccessible(true); c.setAccessible(true);
c.set(rs, conf); c.set(rs, conf);
} }
{
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_sameGroupTransport");
c.setAccessible(true);
c.set(rs, sameGroupTransport);
}
if (diffGroupTransports != null) {
Field t = newClazz.getDeclaredField(FIELDPREFIX + "_diffGroupTransports");
t.setAccessible(true);
t.set(rs, diffGroupTransports.toArray(new Transport[diffGroupTransports.size()]));
}
return rs; return rs;
} catch (RuntimeException rex) { } catch (RuntimeException rex) {
throw rex; throw rex;
} catch (Exception ex) { } catch (Exception ex) {
throw new RuntimeException(ex); throw new RuntimeException(ex);
} }
}
public static <T extends Service> T createSimpleRemoteService(final String name, final Class<T> serviceTypeOrImplClass, final InetSocketAddress clientAddress, final Transport transport) {
return createRemoteService(name, null, serviceTypeOrImplClass, clientAddress, (String) null, new HashSet<>(), (AnyValue) null, transport);
} }
/** /**
@@ -1081,26 +888,22 @@ public abstract class Sncp {
* *
* @param <T> Service泛型 * @param <T> Service泛型
* @param name 资源名 * @param name 资源名
* @param executor 线程池
* @param serviceTypeOrImplClass Service类 * @param serviceTypeOrImplClass Service类
* @param transportFactory TransportFactory
* @param clientAddress 本地IP地址 * @param clientAddress 本地IP地址
* @param sncpGroup 自身的组节点名 可能为null
* @param groups 所有的组节点，包含自身 * @param groups 所有的组节点，包含自身
* @param conf 启动配置项 * @param conf 启动配置项
* @param transport 通信组件
* *
* @return Service的远程模式实例 * @return Service的远程模式实例
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public static <T extends Service> T createRemoteService( public static <T extends Service> T createRemoteService(
final String name, final String name,
final Consumer<Runnable> executor,
final Class<T> serviceTypeOrImplClass, final Class<T> serviceTypeOrImplClass,
final TransportFactory transportFactory,
final InetSocketAddress clientAddress, final InetSocketAddress clientAddress,
final String sncpGroup,
final Set<String> groups, final Set<String> groups,
final AnyValue conf, final AnyValue conf) {
final Transport transport) {
if (serviceTypeOrImplClass == null) return null; if (serviceTypeOrImplClass == null) return null;
if (!Service.class.isAssignableFrom(serviceTypeOrImplClass)) return null; if (!Service.class.isAssignableFrom(serviceTypeOrImplClass)) return null;
int mod = serviceTypeOrImplClass.getModifiers(); int mod = serviceTypeOrImplClass.getModifiers();
@@ -1110,36 +913,18 @@ public abstract class Sncp {
final String clientName = SncpClient.class.getName().replace('.', '/'); final String clientName = SncpClient.class.getName().replace('.', '/');
final String clientDesc = Type.getDescriptor(SncpClient.class); final String clientDesc = Type.getDescriptor(SncpClient.class);
final String sncpDynDesc = Type.getDescriptor(SncpDyn.class); final String sncpDynDesc = Type.getDescriptor(SncpDyn.class);
final String bsonConvertDesc = Type.getDescriptor(BsonConvert.class);
final String jsonConvertDesc = Type.getDescriptor(JsonConvert.class);
final String transportDesc = Type.getDescriptor(Transport.class);
final String stringDesc = Type.getDescriptor(String.class);
final String anyValueDesc = Type.getDescriptor(AnyValue.class); final String anyValueDesc = Type.getDescriptor(AnyValue.class);
ClassLoader loader = Sncp.class.getClassLoader(); ClassLoader loader = Sncp.class.getClassLoader();
String newDynName = supDynName.substring(0, supDynName.lastIndexOf('/') + 1) + REMOTEPREFIX + serviceTypeOrImplClass.getSimpleName(); String newDynName = supDynName.substring(0, supDynName.lastIndexOf('/') + 1) + REMOTEPREFIX + serviceTypeOrImplClass.getSimpleName();
try { try {
Class newClazz = Class.forName(newDynName.replace('/', '.')); Class newClazz = Class.forName(newDynName.replace('/', '.'));
T rs = (T) newClazz.newInstance(); T rs = (T) newClazz.newInstance();
SncpClient client = new SncpClient(name, serviceTypeOrImplClass, rs, executor, true, realed ? createLocalServiceClass(name, serviceTypeOrImplClass) : serviceTypeOrImplClass, clientAddress); SncpClient client = new SncpClient(name, serviceTypeOrImplClass, rs, transportFactory, true, realed ? createLocalServiceClass(name, serviceTypeOrImplClass) : serviceTypeOrImplClass, clientAddress);
client.setRemoteGroups(groups);
client.setRemoteGroupTransport(transportFactory.loadRemoteTransport(clientAddress, groups));
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_client"); Field c = newClazz.getDeclaredField(FIELDPREFIX + "_client");
c.setAccessible(true); c.setAccessible(true);
c.set(rs, client); c.set(rs, client);
Field t = newClazz.getDeclaredField(FIELDPREFIX + "_transport");
t.setAccessible(true);
t.set(rs, transport);
{
StringBuilder sb = new StringBuilder();
sb.append(newClazz.getName()).append("{name = '").append(name);
sb.append("', serviceid = '").append(client.getServiceid());
sb.append("', serviceversion = ").append(client.getServiceversion());
sb.append(", action.size = ").append(client.getActionCount());
sb.append(", address = ").append(clientAddress).append(", groups = ").append(transport == null ? null : transport.getName());
sb.append(", remoteaddrs = ").append(transport == null ? null : Arrays.asList(transport.getRemoteAddresses()));
sb.append("}");
Field s = newClazz.getDeclaredField(FIELDPREFIX + "_selfstring");
s.setAccessible(true);
s.set(rs, sb.toString());
}
return rs; return rs;
} catch (Throwable ex) { } catch (Throwable ex) {
} }
@@ -1172,54 +957,14 @@ public abstract class Sncp {
visitAnnotation(cw.visitAnnotation(Type.getDescriptor(ann.annotationType()), true), ann); visitAnnotation(cw.visitAnnotation(Type.getDescriptor(ann.annotationType()), true), ann);
} }
} }
{
fv = cw.visitField(ACC_PRIVATE + ACC_FINAL + ACC_STATIC, FIELDPREFIX + "_service_type", "Ljava/lang/Class;", null, null);
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_bsonConvert", bsonConvertDesc, null, null);
av0 = fv.visitAnnotation("Ljavax/annotation/Resource;", true);
av0.visitEnd();
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_jsonConvert", jsonConvertDesc, null, null);
av0 = fv.visitAnnotation("Ljavax/annotation/Resource;", true);
av0.visitEnd();
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_sncpGroup", stringDesc, null, null);
fv.visitEnd();
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_groups", "Ljava/util/Set;", "Ljava/util/Set<Ljava/lang/String;>;", null);
fv.visitEnd();
}
{ {
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_conf", anyValueDesc, null, null); fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_conf", anyValueDesc, null, null);
fv.visitEnd(); fv.visitEnd();
} }
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_transport", transportDesc, null, null);
fv.visitEnd();
}
{ {
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_client", clientDesc, null, null); fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_client", clientDesc, null, null);
fv.visitEnd(); fv.visitEnd();
} }
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_selfstring", "Ljava/lang/String;", null, null);
fv.visitEnd();
}
{//静态构造函数
mv = new AsmMethodVisitor(cw.visitMethod(ACC_STATIC, "<clinit>", "()V", null, null));
mv.visitLdcInsn(Type.getType(Type.getDescriptor(serviceTypeOrImplClass)));
mv.visitFieldInsn(PUTSTATIC, newDynName, FIELDPREFIX + "_service_type", "Ljava/lang/Class;");
mv.visitInsn(RETURN);
mv.visitMaxs(1, 0);
mv.visitEnd();
}
{ //构造函数 { //构造函数
mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null)); mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "<init>", "()V", null, null));
//mv.setDebug(true); //mv.setDebug(true);
@@ -1244,16 +989,18 @@ public abstract class Sncp {
{ // toString() { // toString()
mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "toString", "()Ljava/lang/String;", null, null)); mv = new AsmMethodVisitor(cw.visitMethod(ACC_PUBLIC, "toString", "()Ljava/lang/String;", null, null));
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_selfstring", "Ljava/lang/String;"); mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
Label l1 = new Label(); Label l1 = new Label();
mv.visitJumpInsn(IFNONNULL, l1); mv.visitJumpInsn(IFNONNULL, l1);
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitMethodInsn(INVOKESPECIAL, "java/lang/Object", "toString", "()Ljava/lang/String;", false); mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/Object", "getClass", "()Ljava/lang/Class;", false);
mv.visitMethodInsn(INVOKEVIRTUAL, "java/lang/Class", "getName", "()Ljava/lang/String;", false);
Label l2 = new Label(); Label l2 = new Label();
mv.visitJumpInsn(GOTO, l2); mv.visitJumpInsn(GOTO, l2);
mv.visitLabel(l1); mv.visitLabel(l1);
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_selfstring", "Ljava/lang/String;"); mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, "toSimpleString", "()Ljava/lang/String;", false);
mv.visitLabel(l2); mv.visitLabel(l2);
mv.visitInsn(ARETURN); mv.visitInsn(ARETURN);
mv.visitMaxs(1, 1); mv.visitMaxs(1, 1);
@@ -1276,12 +1023,7 @@ public abstract class Sncp {
} }
mv.visitVarInsn(ALOAD, 0); mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc); mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_bsonConvert", bsonConvertDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_jsonConvert", jsonConvertDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_transport", transportDesc);
if (index <= 5) { if (index <= 5) {
mv.visitInsn(ICONST_0 + index); mv.visitInsn(ICONST_0 + index);
} else { } else {
@@ -1326,7 +1068,7 @@ public abstract class Sncp {
} }
} }
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, "remote", "(" + bsonConvertDesc + jsonConvertDesc + transportDesc + "I[Ljava/lang/Object;)Ljava/lang/Object;", false); mv.visitMethodInsn(INVOKEVIRTUAL, clientName, "remote", "(I[Ljava/lang/Object;)Ljava/lang/Object;", false);
//mv.visitMethodInsn(INVOKEVIRTUAL, convertName, "convertFrom", convertFromDesc, false); //mv.visitMethodInsn(INVOKEVIRTUAL, convertName, "convertFrom", convertFromDesc, false);
if (method.getGenericReturnType() == void.class) { if (method.getGenericReturnType() == void.class) {
mv.visitInsn(POP); mv.visitInsn(POP);
@@ -1369,7 +1111,9 @@ public abstract class Sncp {
}.loadClass(newDynName.replace('/', '.'), bytes); }.loadClass(newDynName.replace('/', '.'), bytes);
try { try {
T rs = (T) newClazz.newInstance(); T rs = (T) newClazz.newInstance();
SncpClient client = new SncpClient(name, serviceTypeOrImplClass, rs, executor, true, realed ? createLocalServiceClass(name, serviceTypeOrImplClass) : serviceTypeOrImplClass, clientAddress); SncpClient client = new SncpClient(name, serviceTypeOrImplClass, rs, transportFactory, true, realed ? createLocalServiceClass(name, serviceTypeOrImplClass) : serviceTypeOrImplClass, clientAddress);
client.setRemoteGroups(groups);
client.setRemoteGroupTransport(transportFactory.loadRemoteTransport(clientAddress, groups));
{ {
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_client"); Field c = newClazz.getDeclaredField(FIELDPREFIX + "_client");
c.setAccessible(true); c.setAccessible(true);
@@ -1380,34 +1124,6 @@ public abstract class Sncp {
c.setAccessible(true); c.setAccessible(true);
c.set(rs, conf); c.set(rs, conf);
} }
{
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_sncpGroup");
c.setAccessible(true);
c.set(rs, sncpGroup);
}
{
Field c = newClazz.getDeclaredField(FIELDPREFIX + "_groups");
c.setAccessible(true);
c.set(rs, groups);
}
{
Field t = newClazz.getDeclaredField(FIELDPREFIX + "_transport");
t.setAccessible(true);
t.set(rs, transport);
}
{
StringBuilder sb = new StringBuilder();
sb.append(newClazz.getName()).append("{name = '").append(name);
sb.append("', serviceid = ").append(client.getServiceid());
sb.append(", serviceversion = ").append(client.getServiceversion());
sb.append(", action.size = ").append(client.getActionCount());
sb.append(", address = ").append(clientAddress).append(", groups = ").append(transport == null ? null : transport.getName());
sb.append(", remotes = ").append(transport == null ? null : Arrays.asList(transport.getRemoteAddresses()));
sb.append("}");
Field s = newClazz.getDeclaredField(FIELDPREFIX + "_selfstring");
s.setAccessible(true);
s.set(rs, sb.toString());
}
return rs; return rs;
} catch (Exception ex) { } catch (Exception ex) {
throw new RuntimeException(ex); throw new RuntimeException(ex);

367
src/org/redkale/net/sncp/SncpClient.java
View File

@@ -12,8 +12,8 @@ import java.nio.*;
import java.nio.channels.*; import java.nio.channels.*;
import java.util.*; import java.util.*;
import java.util.concurrent.*; import java.util.concurrent.*;
import java.util.function.*;
import java.util.logging.*; import java.util.logging.*;
import javax.annotation.Resource;
import org.redkale.convert.bson.*; import org.redkale.convert.bson.*;
import org.redkale.convert.json.*; import org.redkale.convert.json.*;
import org.redkale.net.*; import org.redkale.net.*;
@@ -31,111 +31,6 @@ import org.redkale.service.RpcCall;
*/ */
public final class SncpClient { public final class SncpClient {
protected static final class SncpAction {
protected final DLong 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 boolean boolReturnTypeFuture; // 返回结果类型是否为 CompletableFuture
protected final Creator<? extends CompletableFuture> futureCreator;
public SncpAction(final Class clazz, Method method, DLong actionid) {
this.actionid = actionid == null ? Sncp.hash(method) : actionid;
Type rt = method.getGenericReturnType();
if (rt instanceof TypeVariable) {
TypeVariable tv = (TypeVariable) rt;
if (tv.getBounds().length == 1) rt = tv.getBounds()[0];
}
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 = method.getGenericParameterTypes();
this.paramClass = method.getParameterTypes();
this.method = method;
Annotation[][] anns = method.getParameterAnnotations();
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 (AsyncHandler.class.isAssignableFrom(params[i])) {
if (boolReturnTypeFuture) {
throw new RuntimeException(method + " have both AsyncHandler and CompletableFuture");
}
if (handlerFuncIndex >= 0) {
throw new RuntimeException(method + " have more than one AsyncHandler 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 RuntimeException(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;
}
}
for (Annotation ann : anns[i]) {
if (ann.annotationType() == RpcCall.class) {
try {
atts[i + 1] = ((RpcCall) ann).value().newInstance();
hasattr = true;
} catch (Exception e) {
logger.log(Level.SEVERE, RpcCall.class.getSimpleName() + ".attribute cannot a newInstance for" + method, e);
}
break;
}
}
}
}
}
this.addressTargetParamIndex = targetAddrIndex;
this.addressSourceParamIndex = sourceAddrIndex;
this.handlerFuncParamIndex = handlerFuncIndex;
this.handlerAttachParamIndex = handlerAttachIndex;
this.paramAttrs = hasattr ? atts : null;
if (this.handlerFuncParamIndex >= 0 && method.getReturnType() != void.class) {
throw new RuntimeException(method + " have AsyncHandler type parameter but return type is not void");
}
}
@Override
public String toString() {
return "{" + actionid + "," + (method == null ? "null" : method.getName()) + "}";
}
}
protected static final Logger logger = Logger.getLogger(SncpClient.class.getSimpleName()); protected static final Logger logger = Logger.getLogger(SncpClient.class.getSimpleName());
protected final boolean finest = logger.isLoggable(Level.FINEST); protected final boolean finest = logger.isLoggable(Level.FINEST);
@@ -160,12 +55,36 @@ public final class SncpClient {
protected final SncpAction[] actions; protected final SncpAction[] actions;
protected final Consumer<Runnable> executor; protected final ExecutorService executor;
public <T extends Service> SncpClient(final String serviceName, final Class<T> serviceTypeOrImplClass, final T service, final Consumer<Runnable> executor, @Resource
protected JsonConvert jsonConvert;
@Resource
protected BsonConvert bsonConvert;
//远程模式
protected Set<String> remoteGroups;
//远程模式
protected Transport remoteGroupTransport;
//本地模式
protected String sameGroup;
//本地模式
protected Transport sameGroupTransport;
//本地模式
protected Set<String> diffGroups;
//本地模式
protected Transport[] diffGroupTransports;
public <T extends Service> SncpClient(final String serviceName, final Class<T> serviceTypeOrImplClass, final T service, final TransportFactory factory,
final boolean remote, final Class serviceClass, final InetSocketAddress clientAddress) { final boolean remote, final Class serviceClass, final InetSocketAddress clientAddress) {
this.remote = remote; this.remote = remote;
this.executor = executor; this.executor = factory.getExecutor();
this.serviceClass = serviceClass; this.serviceClass = serviceClass;
this.serviceversion = 0; this.serviceversion = 0;
this.clientAddress = clientAddress; this.clientAddress = clientAddress;
@@ -209,6 +128,54 @@ public final class SncpClient {
return actions.length; return actions.length;
} }
public Set<String> getRemoteGroups() {
return remoteGroups;
}
public void setRemoteGroups(Set<String> remoteGroups) {
this.remoteGroups = remoteGroups;
}
public Transport getRemoteGroupTransport() {
return remoteGroupTransport;
}
public void setRemoteGroupTransport(Transport remoteGroupTransport) {
this.remoteGroupTransport = remoteGroupTransport;
}
public String getSameGroup() {
return sameGroup;
}
public void setSameGroup(String sameGroup) {
this.sameGroup = sameGroup;
}
public Transport getSameGroupTransport() {
return sameGroupTransport;
}
public void setSameGroupTransport(Transport sameGroupTransport) {
this.sameGroupTransport = sameGroupTransport;
}
public Set<String> getDiffGroups() {
return diffGroups;
}
public void setDiffGroups(Set<String> diffGroups) {
this.diffGroups = diffGroups;
}
public Transport[] getDiffGroupTransports() {
return diffGroupTransports;
}
public void setDiffGroupTransports(Transport[] diffGroupTransports) {
this.diffGroupTransports = diffGroupTransports;
}
@Override @Override
public String toString() { public String toString() {
String service = serviceClass.getName(); String service = serviceClass.getName();
@@ -218,6 +185,26 @@ public final class SncpClient {
+ ", actions.size = " + actions.length + ")"; + ", actions.size = " + actions.length + ")";
} }
public String toSimpleString() { //给Sncp产生的Service用
String service = serviceClass.getName();
if (remote) service = service.replace(Sncp.LOCALPREFIX, Sncp.REMOTEPREFIX);
List<InetSocketAddress> diffaddrs = new ArrayList<>();
if (diffGroupTransports != null && diffGroupTransports.length > 1) {
for (Transport t : diffGroupTransports) {
diffaddrs.addAll(Utility.ofList(t.getRemoteAddresses()));
}
}
return service + "(name = '" + name + "', serviceid = " + serviceid + ", serviceversion = " + serviceversion
+ ", clientaddr = " + (clientAddress == null ? "" : (clientAddress.getHostString() + ":" + clientAddress.getPort()))
+ ((sameGroup == null || sameGroup.isEmpty()) ? "" : ", sameGroup = " + sameGroup)
+ (sameGroupTransport == null ? "" : ", sameGroupTransport = " + Arrays.toString(sameGroupTransport.getRemoteAddresses()))
+ ((diffGroups == null || diffGroups.isEmpty()) ? "" : ", diffGroups = " + diffGroups)
+ ((diffGroupTransports == null || diffGroupTransports.length < 1) ? "" : ", diffGroupTransports = " + diffaddrs)
+ ((remoteGroups == null || remoteGroups.isEmpty()) ? "" : ", remoteGroups = " + remoteGroups)
+ (remoteGroupTransport == null ? "" : ", remoteGroupTransport = " + Arrays.toString(remoteGroupTransport.getRemoteAddresses()))
+ ", actions.size = " + actions.length + ")";
}
public static List<Method> parseMethod(final Class serviceClass) { public static List<Method> parseMethod(final Class serviceClass) {
final List<Method> list = new ArrayList<>(); final List<Method> list = new ArrayList<>();
final List<Method> multis = new ArrayList<>(); final List<Method> multis = new ArrayList<>();
@@ -260,53 +247,53 @@ public final class SncpClient {
return multis; return multis;
} }
public void remoteSameGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport transport, final int index, final Object... params) { public void remoteSameGroup(final int index, final Object... params) {
if (transport == null) return; if (sameGroupTransport == null) return;
final SncpAction action = actions[index]; final SncpAction action = actions[index];
if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null; //不能让远程调用handler，因为之前本地方法已经调用过了 if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null; //不能让远程调用handler，因为之前本地方法已经调用过了
for (InetSocketAddress addr : transport.getRemoteAddresses()) { for (InetSocketAddress addr : sameGroupTransport.getRemoteAddresses()) {
remote0(null, bsonConvert, jsonConvert, transport, addr, action, params); remote0(null, sameGroupTransport, addr, action, params);
} }
} }
public void asyncRemoteSameGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport transport, final int index, final Object... params) { public void asyncRemoteSameGroup(final int index, final Object... params) {
if (transport == null) return; if (sameGroupTransport == null) return;
if (executor != null) { if (executor != null) {
executor.accept(() -> { executor.execute(() -> {
remoteSameGroup(bsonConvert, jsonConvert, transport, index, params); remoteSameGroup(index, params);
}); });
} else { } else {
remoteSameGroup(bsonConvert, jsonConvert, transport, index, params); remoteSameGroup(index, params);
} }
} }
public void remoteDiffGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport[] transports, final int index, final Object... params) { public void remoteDiffGroup(final int index, final Object... params) {
if (transports == null || transports.length < 1) return; if (diffGroupTransports == null || diffGroupTransports.length < 1) return;
final SncpAction action = actions[index]; final SncpAction action = actions[index];
if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null; //不能让远程调用handler，因为之前本地方法已经调用过了 if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null; //不能让远程调用handler，因为之前本地方法已经调用过了
for (Transport transport : transports) { for (Transport transport : diffGroupTransports) {
remote0(null, bsonConvert, jsonConvert, transport, null, action, params); remote0(null, transport, null, action, params);
} }
} }
public void asyncRemoteDiffGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport[] transports, final int index, final Object... params) { public void asyncRemoteDiffGroup(final int index, final Object... params) {
if (transports == null || transports.length < 1) return; if (diffGroupTransports == null || diffGroupTransports.length < 1) return;
if (executor != null) { if (executor != null) {
executor.accept(() -> { executor.execute(() -> {
remoteDiffGroup(bsonConvert, jsonConvert, transports, index, params); remoteDiffGroup(index, params);
}); });
} else { } else {
remoteDiffGroup(bsonConvert, jsonConvert, transports, index, params); remoteDiffGroup(index, params);
} }
} }
//只给远程模式调用的 //只给远程模式调用的
public <T> T remote(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport transport, final int index, final Object... params) { public <T> T remote(final int index, final Object... params) {
final SncpAction action = actions[index]; final SncpAction action = actions[index];
final AsyncHandler handlerFunc = action.handlerFuncParamIndex >= 0 ? (AsyncHandler) params[action.handlerFuncParamIndex] : null; final AsyncHandler handlerFunc = action.handlerFuncParamIndex >= 0 ? (AsyncHandler) params[action.handlerFuncParamIndex] : null;
if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null; if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null;
final BsonReader reader = bsonConvert.pollBsonReader(); final BsonReader reader = bsonConvert.pollBsonReader();
CompletableFuture<byte[]> future = remote0(handlerFunc, bsonConvert, jsonConvert, transport, null, action, params); CompletableFuture<byte[]> future = remote0(handlerFunc, remoteGroupTransport, null, action, params);
if (action.boolReturnTypeFuture) { if (action.boolReturnTypeFuture) {
CompletableFuture result = action.futureCreator.create(); CompletableFuture result = action.futureCreator.create();
future.whenComplete((v, e) -> { future.whenComplete((v, e) -> {
@@ -349,27 +336,19 @@ public final class SncpClient {
} }
} }
public <T> void remote(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport[] transports, final int index, final Object... params) { private CompletableFuture<byte[]> remote0(final AsyncHandler handler, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) {
if (transports == null || transports.length < 1) return;
remote(bsonConvert, jsonConvert, transports[0], index, params);
for (int i = 1; i < transports.length; i++) {
remote0(null, bsonConvert, jsonConvert, transports[i], null, actions[index], params);
}
}
private CompletableFuture<byte[]> remote0(final AsyncHandler handler, final BsonConvert bsonConvert, final JsonConvert jsonConvert, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) {
if ("rest".equalsIgnoreCase(transport.getSubprotocol())) { if ("rest".equalsIgnoreCase(transport.getSubprotocol())) {
return remoteRest0(handler, jsonConvert, transport, addr0, action, params); return remoteRest0(handler, transport, addr0, action, params);
} }
return remoteSncp0(handler, bsonConvert, transport, addr0, action, params); return remoteSncp0(handler, transport, addr0, action, params);
} }
//尚未实现 //尚未实现
private CompletableFuture<byte[]> remoteRest0(final AsyncHandler handler, final JsonConvert jsonConvert, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) { private CompletableFuture<byte[]> remoteRest0(final AsyncHandler handler, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) {
return null; return null;
} }
private CompletableFuture<byte[]> remoteSncp0(final AsyncHandler handler, final BsonConvert bsonConvert, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) { private CompletableFuture<byte[]> remoteSncp0(final AsyncHandler handler, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) {
final Type[] myparamtypes = action.paramTypes; final Type[] myparamtypes = action.paramTypes;
final Class[] myparamclass = action.paramClass; final Class[] myparamclass = action.paramClass;
if (action.addressSourceParamIndex >= 0) params[action.addressSourceParamIndex] = this.clientAddress; if (action.addressSourceParamIndex >= 0) params[action.addressSourceParamIndex] = this.clientAddress;
@@ -555,4 +534,108 @@ public final class SncpClient {
buffer.position(currentpos); buffer.position(currentpos);
} }
protected static final class SncpAction {
protected final DLong 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 boolean boolReturnTypeFuture; // 返回结果类型是否为 CompletableFuture
protected final Creator<? extends CompletableFuture> futureCreator;
public SncpAction(final Class clazz, Method method, DLong actionid) {
this.actionid = actionid == null ? Sncp.hash(method) : actionid;
Type rt = method.getGenericReturnType();
if (rt instanceof TypeVariable) {
TypeVariable tv = (TypeVariable) rt;
if (tv.getBounds().length == 1) rt = tv.getBounds()[0];
}
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 = method.getGenericParameterTypes();
this.paramClass = method.getParameterTypes();
this.method = method;
Annotation[][] anns = method.getParameterAnnotations();
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 (AsyncHandler.class.isAssignableFrom(params[i])) {
if (boolReturnTypeFuture) {
throw new RuntimeException(method + " have both AsyncHandler and CompletableFuture");
}
if (handlerFuncIndex >= 0) {
throw new RuntimeException(method + " have more than one AsyncHandler 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 RuntimeException(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;
}
}
for (Annotation ann : anns[i]) {
if (ann.annotationType() == RpcCall.class) {
try {
atts[i + 1] = ((RpcCall) ann).value().newInstance();
hasattr = true;
} catch (Exception e) {
logger.log(Level.SEVERE, RpcCall.class.getSimpleName() + ".attribute cannot a newInstance for" + method, e);
}
break;
}
}
}
}
}
this.addressTargetParamIndex = targetAddrIndex;
this.addressSourceParamIndex = sourceAddrIndex;
this.handlerFuncParamIndex = handlerFuncIndex;
this.handlerAttachParamIndex = handlerAttachIndex;
this.paramAttrs = hasattr ? atts : null;
if (this.handlerFuncParamIndex >= 0 && method.getReturnType() != void.class) {
throw new RuntimeException(method + " have AsyncHandler type parameter but return type is not void");
}
}
@Override
public String toString() {
return "{" + actionid + "," + (method == null ? "null" : method.getName()) + "}";
}
}
} }