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

Browse Source
This commit is contained in:
Redkale
2016-08-26 07:16:47 +08:00
parent d5a752dea0
commit 470a82b6e6
2 changed files with 121 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,7 @@ import jdk.internal.org.objectweb.asm.*;
import static jdk.internal.org.objectweb.asm.Opcodes.*;
import jdk.internal.org.objectweb.asm.Type;
import org.redkale.convert.bson.BsonConvert;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.Transport;
import org.redkale.net.sncp.SncpClient.SncpAction;
import org.redkale.service.*;
@@ -25,8 +26,8 @@ import org.redkale.util.*;
/**
* Service Node Communicate Protocol
* 生成Service的本地模式或远程模式Service-Class的工具类
*
*
*
*
* 详情见: http://redkale.org
*
* @author zhangjx
@@ -79,6 +80,7 @@ public abstract class Sncp {
* 对类名或者name字符串进行hash。
*
* @param name String
*
* @return hash值
*/
public static DLong hash(final String name) {
@@ -131,79 +133,83 @@ public abstract class Sncp {
/**
* <blockquote><pre>
* public class TestService implements Service{
*
*
* public String findSomeThing(){
* return "hello";
* }
*
*
* &#64;MultiRun(selfrun = false)
* public void createSomeThing(TestBean bean){
* //do something
* }
*
*
* &#64;MultiRun
* public String updateSomeThing(String id){
* return "hello" + id;
* }
* }
* </pre></blockquote>
*
*
* <blockquote><pre>
* &#64;Resource(name = "")
* &#64;SncpDyn(remote = false)
* &#64;ResourceType({TestService.class})
* public final class _DynLocalTestService extends TestService{
*
*
* &#64;Resource
* private BsonConvert _redkale_convert;
*
* private BsonConvert _redkale_bsonConvert;
*
* &#64;Resource
* private JsonConvert _redkale_jsonConvert;
*
* private Transport _redkale_sameGroupTransport;
*
*
* private Transport[] _redkale_diffGroupTransports;
*
*
* private SncpClient _redkale_client;
*
*
* private String _redkale_selfstring;
*
*
* &#64;Override
* public String toString() {
* return _redkale_selfstring == null ? super.toString() : _redkale_selfstring;
* }
*
*
* &#64;Override
* public void createSomeThing(TestBean bean){
* this._redkale_createSomeThing(false, true, true, bean);
* }
*
*
* &#64;SncpDyn(remote = false, index = 0)
* public void _redkale_createSomeThing(boolean selfrunnable, boolean samerunnable, boolean diffrunnable, TestBean bean){
* if(selfrunnable) super.createSomeThing(bean);
* if (_redkale_client== null) return;
* if (samerunnable) _redkale_client.remoteSameGroup(_redkale_convert, _sameGroupTransport, 0, true, false, false, bean);
* if (diffrunnable) _redkale_client.remoteDiffGroup(_redkale_convert, _diffGroupTransports, 0, true, true, false, bean);
* if (samerunnable) _redkale_client.remoteSameGroup(_redkale_bsonConvert, _redkale_jsonConvert, _sameGroupTransport, 0, true, false, false, bean);
* if (diffrunnable) _redkale_client.remoteDiffGroup(_redkale_bsonConvert, _redkale_jsonConvert, _diffGroupTransports, 0, true, true, false, bean);
* }
*
*
* &#64;Override
* public String updateSomeThing(String id){
* return this._redkale_updateSomeThing(true, true, true, id);
* }
*
*
* &#64;SncpDyn(remote = false, index = 1)
* public String _redkale_updateSomeThing(boolean selfrunnable, boolean samerunnable, boolean diffrunnable, String id){
* String rs = super.updateSomeThing(id);
* if (_redkale_client== null) return;
* if (samerunnable) _redkale_client.remoteSameGroup(_redkale_convert, _sameGroupTransport, 1, true, false, false, id);
* if (diffrunnable) _redkale_client.remoteDiffGroup(_redkale_convert, _diffGroupTransports, 1, true, true, false, id);
* if (samerunnable) _redkale_client.remoteSameGroup(_redkale_bsonConvert, _redkale_jsonConvert, _sameGroupTransport, 1, true, false, false, id);
* if (diffrunnable) _redkale_client.remoteDiffGroup(_redkale_bsonConvert, _redkale_jsonConvert, _diffGroupTransports, 1, true, true, false, id);
* return rs;
* }
* }
* </pre></blockquote>
*
*
* 创建Service的本地模式Class
*
* @param <T> Service子类
* @param name 资源名
* @param serviceClass Service类
*
* @return Service实例
*/
@SuppressWarnings("unchecked")
@@ -217,7 +223,8 @@ public abstract class Sncp {
final String supDynName = serviceClass.getName().replace('.', '/');
final String clientName = SncpClient.class.getName().replace('.', '/');
final String clientDesc = Type.getDescriptor(SncpClient.class);
final String convertDesc = Type.getDescriptor(BsonConvert.class);
final String bsonConvertDesc = Type.getDescriptor(BsonConvert.class);
final String jsonConvertDesc = Type.getDescriptor(JsonConvert.class);
final String sncpDynDesc = Type.getDescriptor(SncpDyn.class);
final String transportDesc = Type.getDescriptor(Transport.class);
final String transportsDesc = Type.getDescriptor(Transport[].class);
@@ -275,7 +282,13 @@ public abstract class Sncp {
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_convert", convertDesc, null, null);
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();
@@ -481,20 +494,24 @@ public abstract class Sncp {
mv.visitVarInsn(ALOAD, 0);//调用 _client
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitVarInsn(ALOAD, 0); //传递 _convert
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_convert", convertDesc);
mv.visitVarInsn(ALOAD, 0); //传递 _bsonConvert
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 = 4; //调用selfrunnable之前的参数个数; _client/_bsonConvert/_jsonConvert/_sameGroupTransport
if (index <= 5) { //第几个 SncpAction
mv.visitInsn(ICONST_0 + index);
} else {
mv.visitIntInsn(BIPUSH, index);
}
if (paramtypes.length + 3 <= 5) { //参数总数量
mv.visitInsn(ICONST_0 + paramtypes.length + 3);
if (paramtypes.length + preparams <= 5) { //参数总数量
mv.visitInsn(ICONST_0 + paramtypes.length + preparams);
} else {
mv.visitIntInsn(BIPUSH, paramtypes.length + 3);
mv.visitIntInsn(BIPUSH, paramtypes.length + preparams);
}
mv.visitTypeInsn(ANEWARRAY, "java/lang/Object");
@@ -517,7 +534,7 @@ public abstract class Sncp {
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
int insn = 3;
int insn = 3; //空3给selfrunnable、samerunnable、diffrunnable
for (int j = 0; j < paramtypes.length; j++) {
final Class pt = paramtypes[j];
mv.visitInsn(DUP);
@@ -544,7 +561,7 @@ public abstract class Sncp {
}
mv.visitInsn(AASTORE);
}
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteSameGroup" : "remoteSameGroup", "(" + convertDesc + transportDesc + "I[Ljava/lang/Object;)V", false);
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteSameGroup" : "remoteSameGroup", "(" + bsonConvertDesc + jsonConvertDesc + transportDesc + "I[Ljava/lang/Object;)V", false);
mv.visitLabel(sameLabel);
//---------------------------- 调用diffrun ---------------------------------
mv.visitVarInsn(ILOAD, 3); //读取 diffrunnable
@@ -554,7 +571,9 @@ public abstract class Sncp {
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_convert", convertDesc);
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);
@@ -563,10 +582,10 @@ public abstract class Sncp {
} else {
mv.visitIntInsn(BIPUSH, index);
}
if (paramtypes.length + 3 <= 5) { //参数总数量
mv.visitInsn(ICONST_0 + paramtypes.length + 3);
if (paramtypes.length + preparams <= 5) { //参数总数量
mv.visitInsn(ICONST_0 + paramtypes.length + preparams);
} else {
mv.visitIntInsn(BIPUSH, paramtypes.length + 3);
mv.visitIntInsn(BIPUSH, paramtypes.length + preparams);
}
mv.visitTypeInsn(ANEWARRAY, "java/lang/Object");
@@ -589,7 +608,7 @@ public abstract class Sncp {
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
insn = 3;
insn = 3;//空3给selfrunnable、samerunnable、diffrunnable
for (int j = 0; j < paramtypes.length; j++) {
final Class pt = paramtypes[j];
mv.visitInsn(DUP);
@@ -616,7 +635,7 @@ public abstract class Sncp {
}
mv.visitInsn(AASTORE);
}
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteDiffGroup" : "remoteDiffGroup", "(" + convertDesc + transportsDesc + "I[Ljava/lang/Object;)V", false);
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteDiffGroup" : "remoteDiffGroup", "(" + bsonConvertDesc + jsonConvertDesc + transportsDesc + "I[Ljava/lang/Object;)V", false);
mv.visitLabel(diffLabel);
if (returnType == void.class) {
@@ -712,11 +731,12 @@ public abstract class Sncp {
* @param clientAddress 本地IP地址
* @param sameGroupTransport 同组的通信组件
* @param diffGroupTransports 异组的通信组件列表
*
* @return Service的本地模式实例
*/
@SuppressWarnings("unchecked")
public static <T extends Service> T createLocalService(final String name, final Consumer<Runnable> executor, final ResourceFactory resourceFactory,
final Class<T> serviceClass, final InetSocketAddress clientAddress, final Transport sameGroupTransport, final Collection<Transport> diffGroupTransports) {
final Class<T> serviceClass, final InetSocketAddress clientAddress, final Transport sameGroupTransport, final Collection<Transport> diffGroupTransports) {
try {
final Class newClazz = createLocalServiceClass(name, serviceClass);
T rs = (T) newClazz.newInstance();
@@ -823,48 +843,51 @@ public abstract class Sncp {
* &#64;SncpDyn(remote = true)
* &#64;ResourceType({TestService.class})
* public final class _DynRemoteTestService extends TestService{
*
*
* &#64;Resource
* private BsonConvert _redkale_convert;
*
* private BsonConvert _redkale_bsonConvert;
*
* &#64;Resource
* private JsonConvert _redkale_jsonConvert;
*
* private Transport _redkale_transport;
*
*
* private SncpClient _redkale_client;
*
*
* private String _redkale_selfstring;
*
*
* &#64;Override
* public String toString() {
* return _redkale_selfstring == null ? super.toString() : _redkale_selfstring;
* }
*
*
* &#64;SncpDyn(remote = false, index = 0)
* public void _redkale_createSomeThing(boolean selfrunnable, boolean samerunnable, boolean diffrunnable, TestBean bean){
* _redkale_client.remote(_redkale_convert, _redkale_transport, 0, selfrunnable, samerunnable, diffrunnable, bean);
* _redkale_client.remote(_redkale_bsonConvert, _redkale_jsonConvert, _redkale_transport, 0, selfrunnable, samerunnable, diffrunnable, bean);
* }
*
*
* &#64;SncpDyn(remote = false, index = 1)
* public String _redkale_updateSomeThing(boolean selfrunnable, boolean samerunnable, boolean diffrunnable, String id){
* return _redkale_client.remote(_redkale_convert, _redkale_transport, 1, selfrunnable, samerunnable, diffrunnable, id);
* return _redkale_client.remote(_redkale_bsonConvert, _redkale_jsonConvert, _redkale_transport, 1, selfrunnable, samerunnable, diffrunnable, id);
* }
*
*
* &#64;Override
* public void createSomeThing(TestBean bean){
* _redkale_client.remote(_redkale_convert, _redkale_transport, 2, bean);
* _redkale_client.remote(_redkale_bsonConvert, _redkale_jsonConvert, _redkale_transport, 2, bean);
* }
*
*
* &#64;Override
* public String findSomeThing(){
* return _redkale_client.remote(_redkale_convert, _redkale_transport, 3);
* return _redkale_client.remote(_redkale_bsonConvert, _redkale_jsonConvert, _redkale_transport, 3);
* }
*
*
* &#64;Override
* public String updateSomeThing(String id){
* return _redkale_client.remote(_redkale_convert, _redkale_transport, 4, id);
* return _redkale_client.remote(_redkale_bsonConvert, _redkale_jsonConvert, _redkale_transport, 4, id);
* }
* }
* </pre></blockquote>
*
*
* 创建远程模式的Service实例
*
* @param <T> Service泛型
@@ -878,7 +901,7 @@ public abstract class Sncp {
*/
@SuppressWarnings("unchecked")
public static <T extends Service> T createRemoteService(final String name, final Consumer<Runnable> executor, final Class<T> serviceClass,
final InetSocketAddress clientAddress, final Transport transport) {
final InetSocketAddress clientAddress, final Transport transport) {
if (serviceClass == null) return null;
if (!Service.class.isAssignableFrom(serviceClass)) return null;
int mod = serviceClass.getModifiers();
@@ -888,7 +911,8 @@ public abstract class Sncp {
final String clientName = SncpClient.class.getName().replace('.', '/');
final String clientDesc = Type.getDescriptor(SncpClient.class);
final String sncpDynDesc = Type.getDescriptor(SncpDyn.class);
final String convertDesc = Type.getDescriptor(BsonConvert.class);
final String bsonConvertDesc = Type.getDescriptor(BsonConvert.class);
final String jsonConvertDesc = Type.getDescriptor(JsonConvert.class);
final String transportDesc = Type.getDescriptor(Transport.class);
final String anyValueDesc = Type.getDescriptor(AnyValue.class);
ClassLoader loader = Sncp.class.getClassLoader();
@@ -958,7 +982,13 @@ public abstract class Sncp {
}
}
{
fv = cw.visitField(ACC_PRIVATE, FIELDPREFIX + "_convert", convertDesc, null, null);
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();
@@ -1032,7 +1062,9 @@ public abstract class Sncp {
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_convert", convertDesc);
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) {
@@ -1079,7 +1111,7 @@ public abstract class Sncp {
}
}
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, "remote", "(" + convertDesc + transportDesc + "I[Ljava/lang/Object;)Ljava/lang/Object;", false);
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, "remote", "(" + bsonConvertDesc + jsonConvertDesc + transportDesc + "I[Ljava/lang/Object;)Ljava/lang/Object;", false);
//mv.visitMethodInsn(INVOKEVIRTUAL, convertName, "convertFrom", convertFromDesc, false);
if (method.getGenericReturnType() == void.class) {
mv.visitInsn(POP);

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

@@ -140,7 +140,7 @@ public final class SncpClient {
protected final Consumer<Runnable> executor;
public <T extends Service> SncpClient(final String serviceName, final Class<T> serviceType, final Consumer<Runnable> executor,
final boolean remote, final Class serviceClass, final InetSocketAddress clientAddress) {
final boolean remote, final Class serviceClass, final InetSocketAddress clientAddress) {
this.remote = remote;
this.executor = executor;
this.serviceClass = serviceClass;
@@ -174,8 +174,8 @@ public final class SncpClient {
String service = serviceClass.getName();
if (remote) service = service.replace(Sncp.LOCALPREFIX, Sncp.REMOTEPREFIX);
return this.getClass().getSimpleName() + "(service = " + service + ", serviceid = " + serviceid + ", name = '" + name
+ "', address = " + (clientAddress == null ? "" : (clientAddress.getHostString() + ":" + clientAddress.getPort()))
+ ", actions.size = " + actions.length + ")";
+ "', address = " + (clientAddress == null ? "" : (clientAddress.getHostString() + ":" + clientAddress.getPort()))
+ ", actions.size = " + actions.length + ")";
}
public static List<Method> parseMethod(final Class serviceClass) {
@@ -218,83 +218,83 @@ public final class SncpClient {
return multis;
}
public void remoteSameGroup(final BsonConvert convert, Transport transport, final int index, final Object... params) {
public void remoteSameGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport transport, final int index, final Object... params) {
final SncpAction action = actions[index];
if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null; //不能让远程调用handler，因为之前本地方法已经调用过了
for (InetSocketAddress addr : transport.getRemoteAddresses()) {
remote0(null, convert, transport, addr, action, params);
remote0(null, bsonConvert, jsonConvert, transport, addr, action, params);
}
}
public void asyncRemoteSameGroup(final BsonConvert convert, Transport transport, final int index, final Object... params) {
public void asyncRemoteSameGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport transport, final int index, final Object... params) {
if (executor != null) {
executor.accept(() -> {
remoteSameGroup(convert, transport, index, params);
remoteSameGroup(bsonConvert, jsonConvert, transport, index, params);
});
} else {
remoteSameGroup(convert, transport, index, params);
remoteSameGroup(bsonConvert, jsonConvert, transport, index, params);
}
}
public void remoteDiffGroup(final BsonConvert convert, Transport[] transports, final int index, final Object... params) {
public void remoteDiffGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport[] transports, final int index, final Object... params) {
if (transports == null || transports.length < 1) return;
final SncpAction action = actions[index];
if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null; //不能让远程调用handler，因为之前本地方法已经调用过了
for (Transport transport : transports) {
remote0(null, convert, transport, null, action, params);
remote0(null, bsonConvert, jsonConvert, transport, null, action, params);
}
}
public void asyncRemoteDiffGroup(final BsonConvert convert, Transport[] transports, final int index, final Object... params) {
public void asyncRemoteDiffGroup(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport[] transports, final int index, final Object... params) {
if (transports == null || transports.length < 1) return;
if (executor != null) {
executor.accept(() -> {
remoteDiffGroup(convert, transports, index, params);
remoteDiffGroup(bsonConvert, jsonConvert, transports, index, params);
});
} else {
remoteDiffGroup(convert, transports, index, params);
remoteDiffGroup(bsonConvert, jsonConvert, transports, index, params);
}
}
//只给远程模式调用的
public <T> T remote(final BsonConvert convert, Transport transport, final int index, final Object... params) {
public <T> T remote(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport transport, final int index, final Object... params) {
final SncpAction action = actions[index];
final CompletionHandler handlerFunc = action.handlerFuncParamIndex >= 0 ? (CompletionHandler) params[action.handlerFuncParamIndex] : null;
if (action.handlerFuncParamIndex >= 0) params[action.handlerFuncParamIndex] = null;
Future<byte[]> future = remote0(handlerFunc, convert, transport, null, action, params);
Future<byte[]> future = remote0(handlerFunc, bsonConvert, jsonConvert, transport, null, action, params);
if (handlerFunc != null) return null;
final BsonReader reader = convert.pollBsonReader();
final BsonReader reader = bsonConvert.pollBsonReader();
try {
reader.setBytes(future.get(5, TimeUnit.SECONDS));
byte i;
while ((i = reader.readByte()) != 0) {
final Attribute attr = action.paramAttrs[i];
attr.set(params[i - 1], convert.convertFrom(attr.type(), reader));
attr.set(params[i - 1], bsonConvert.convertFrom(attr.type(), reader));
}
return convert.convertFrom(action.resultTypes, reader);
return bsonConvert.convertFrom(action.resultTypes, reader);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
logger.log(Level.SEVERE, actions[index].method + " sncp (params: " + jsonConvert.convertTo(params) + ") remote error", e);
throw new RuntimeException(actions[index].method + " sncp remote error", e);
} finally {
convert.offerBsonReader(reader);
bsonConvert.offerBsonReader(reader);
}
}
public <T> void remote(final BsonConvert convert, Transport[] transports, final int index, final Object... params) {
public <T> void remote(final BsonConvert bsonConvert, final JsonConvert jsonConvert, Transport[] transports, final int index, final Object... params) {
if (transports == null || transports.length < 1) return;
remote(convert, transports[0], index, params);
remote(bsonConvert, jsonConvert, transports[0], index, params);
for (int i = 1; i < transports.length; i++) {
remote0(null, convert, transports[i], null, actions[index], params);
remote0(null, bsonConvert, jsonConvert, transports[i], null, actions[index], params);
}
}
private Future<byte[]> remote0(final CompletionHandler handler, final BsonConvert convert, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) {
private Future<byte[]> remote0(final CompletionHandler handler, final BsonConvert bsonConvert, final JsonConvert jsonConvert, final Transport transport, final SocketAddress addr0, final SncpAction action, final Object... params) {
Type[] myparamtypes = action.paramTypes;
if (action.addressSourceParamIndex >= 0) params[action.addressSourceParamIndex] = this.clientAddress;
final BsonWriter writer = convert.pollBsonWriter(transport.getBufferSupplier()); // 将head写入
final BsonWriter writer = bsonConvert.pollBsonWriter(transport.getBufferSupplier()); // 将head写入
writer.writeTo(DEFAULT_HEADER);
for (int i = 0; i < params.length; i++) {
convert.convertTo(writer, myparamtypes[i], params[i]);
bsonConvert.convertTo(writer, myparamtypes[i], params[i]);
}
final int reqBodyLength = writer.count() - HEADER_SIZE; //body总长度
final long seqid = System.nanoTime();
@@ -393,20 +393,20 @@ public final class SncpClient {
transport.offerConnection(false, conn);
if (handler != null) {
final Object handlerAttach = action.handlerAttachParamIndex >= 0 ? params[action.handlerAttachParamIndex] : null;
final BsonReader reader = convert.pollBsonReader();
final BsonReader reader = bsonConvert.pollBsonReader();
try {
reader.setBytes(this.body);
int i;
while ((i = (reader.readByte() & 0xff)) != 0) {
final Attribute attr = action.paramAttrs[i];
attr.set(params[i - 1], convert.convertFrom(attr.type(), reader));
attr.set(params[i - 1], bsonConvert.convertFrom(attr.type(), reader));
}
Object rs = convert.convertFrom(action.resultTypes, reader);
Object rs = bsonConvert.convertFrom(action.resultTypes, reader);
handler.completed(rs, handlerAttach);
} catch (Exception e) {
handler.failed(e, handlerAttach);
} finally {
convert.offerBsonReader(reader);
bsonConvert.offerBsonReader(reader);
}
}
}