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

移除RpcMultiRun功能

Browse Source
This commit is contained in:
Redkale
2020-05-15 22:02:54 +08:00
parent f1a97c0219
commit 292ff63699
5 changed files with 65 additions and 448 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -230,33 +230,6 @@ public abstract class Sncp {
* public String toString() {
* return _redkale_selfstring == null ? super.toString() : _redkale_selfstring;
* }
*
* @Override
* public void createSomeThing(TestBean bean){
* this._redkale_createSomeThing(false, true, true, bean);
* }
*
* @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(0, true, false, false, bean);
* if (diffrunnable) _redkale_client.remoteDiffGroup(0, true, true, false, bean);
* }
*
* @Override
* public String updateSomeThing(String id){
* return this._redkale_updateSomeThing(true, true, true, id);
* }
*
* @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 rs;
* if (samerunnable) _redkale_client.remoteSameGroup(1, true, false, false, id);
* if (diffrunnable) _redkale_client.remoteDiffGroup(1, true, true, false, id);
* return rs;
* }
* }
* </pre></blockquote>
*
@@ -364,308 +337,6 @@ public abstract class Sncp {
mv.visitMaxs(1, 1);
mv.visitEnd();
}
int i = - 1;
for (final Method method : methods) {
final RpcMultiRun mrun = method.getAnnotation(RpcMultiRun.class);
if (mrun == null) continue;
final Class returnType = method.getReturnType();
final String methodDesc = Type.getMethodDescriptor(method);
final Class[] paramtypes = method.getParameterTypes();
final int index = ++i;
{ //原始方法
mv = new MethodDebugVisitor(cw.visitMethod(ACC_PUBLIC + (method.isVarArgs() ? ACC_VARARGS : 0), method.getName(), methodDesc, null, null));
//mv.setDebug(true);
{ //给参数加上 Annotation
final Annotation[][] anns = method.getParameterAnnotations();
for (int k = 0; k < anns.length; k++) {
for (Annotation ann : anns[k]) {
if (ann instanceof SncpDyn || ann instanceof RpcMultiRun) continue; //必须过滤掉 RpcMultiRun、SncpDyn，否则生成远程模式Service时会出错
visitAnnotation(mv.visitParameterAnnotation(k, Type.getDescriptor(ann.annotationType()), true), ann);
}
}
}
mv.visitVarInsn(ALOAD, 0);
mv.visitInsn(mrun.selfrun() ? ICONST_1 : ICONST_0);
mv.visitInsn(mrun.samerun() ? ICONST_1 : ICONST_0);
mv.visitInsn(mrun.diffrun() ? ICONST_1 : ICONST_0);
int varindex = 0;
boolean handlerFuncFlag = false;
for (Class pt : paramtypes) {
if (CompletionHandler.class.isAssignableFrom(pt)) {
if (handlerFuncFlag) throw new RuntimeException(method + " have more than one CompletionHandler type parameter");
checkAsyncModifier(pt, method);
handlerFuncFlag = true;
}
if (pt.isPrimitive()) {
if (pt == long.class) {
mv.visitVarInsn(LLOAD, ++varindex);
++varindex;
} else if (pt == double.class) {
mv.visitVarInsn(DLOAD, ++varindex);
++varindex;
} else if (pt == float.class) {
mv.visitVarInsn(FLOAD, ++varindex);
} else {
mv.visitVarInsn(ILOAD, ++varindex);
}
} else {
mv.visitVarInsn(ALOAD, ++varindex);
}
}
mv.visitMethodInsn(INVOKEVIRTUAL, newDynName, FIELDPREFIX + "_" + method.getName(), "(ZZZ" + methodDesc.substring(1), false);
if (returnType == void.class) {
mv.visitInsn(RETURN);
} else if (returnType.isPrimitive()) {
if (returnType == long.class) {
mv.visitInsn(LRETURN);
} else if (returnType == float.class) {
mv.visitInsn(FRETURN);
} else if (returnType == double.class) {
mv.visitInsn(DRETURN);
} else {
mv.visitInsn(IRETURN);
}
} else {
mv.visitInsn(ARETURN);
}
mv.visitMaxs(varindex + 3, varindex + 1);
mv.visitEnd();
}
{ // _方法 _方法比无_方法多了三个参数
mv = new MethodDebugVisitor(cw.visitMethod(ACC_PUBLIC + (method.isVarArgs() ? ACC_VARARGS : 0), FIELDPREFIX + "_" + method.getName(), "(ZZZ" + methodDesc.substring(1), null, null));
//mv.setDebug(true);
{ //给参数加上 Annotation
final Annotation[][] anns = method.getParameterAnnotations();
boolean handlerAttachFlag = false;
for (int k = 0; k < anns.length; k++) {
for (Annotation ann : anns[k]) {
if (ann.annotationType() == RpcAttachment.class) {
if (handlerAttachFlag) {
throw new RuntimeException(method + " have more than one @RpcAttachment parameter");
}
handlerAttachFlag = true;
}
if (ann instanceof SncpDyn || ann instanceof RpcMultiRun) continue; //必须过滤掉 RpcMultiRun、SncpDyn，否则生成远程模式Service时会出错
visitAnnotation(mv.visitParameterAnnotation(k, Type.getDescriptor(ann.annotationType()), true), ann);
}
}
}
av0 = mv.visitAnnotation(sncpDynDesc, true);
av0.visit("remote", Boolean.FALSE);
av0.visit("index", index);
av0.visitEnd();
//---------------------------- 调用selfrun ---------------------------------
Label selfLabel = new Label();
if (returnType == void.class) { // if
mv.visitVarInsn(ILOAD, 1);
mv.visitJumpInsn(IFEQ, selfLabel);
}
mv.visitVarInsn(ALOAD, 0);
int varindex = 3; //空3给selfrunnable、samerunnable、diffrunnable
for (Class pt : paramtypes) {
if (pt.isPrimitive()) {
if (pt == long.class) {
mv.visitVarInsn(LLOAD, ++varindex);
++varindex;
} else if (pt == double.class) {
mv.visitVarInsn(DLOAD, ++varindex);
++varindex;
} else if (pt == float.class) {
mv.visitVarInsn(FLOAD, ++varindex);
} else {
mv.visitVarInsn(ILOAD, ++varindex);
}
} else {
mv.visitVarInsn(ALOAD, ++varindex);
}
}
mv.visitMethodInsn(INVOKESPECIAL, supDynName, method.getName(), methodDesc, false);
if (returnType == void.class) { // end if
mv.visitLabel(selfLabel);
}
if (returnType == void.class) {
} else if (returnType.isPrimitive()) {
if (returnType == long.class) {
mv.visitVarInsn(LSTORE, ++varindex);
//++varindex; //多加1
} else if (returnType == float.class) {
mv.visitVarInsn(FSTORE, ++varindex);
} else if (returnType == double.class) {
mv.visitVarInsn(DSTORE, ++varindex);
//++varindex; //多加1
} else {
mv.visitVarInsn(ISTORE, ++varindex);
}
} else {
mv.visitVarInsn(ASTORE, ++varindex);
}
final int rsindex = varindex; //
//---------------------------if (_redkale_client== null) return ----------------------------------
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
Label clientLabel = new Label();
mv.visitJumpInsn(IFNONNULL, clientLabel);
if (returnType == void.class) {
mv.visitInsn(RETURN);
} else if (returnType.isPrimitive()) {
if (returnType == long.class) {
mv.visitVarInsn(LLOAD, rsindex);
mv.visitInsn(LRETURN);
} else if (returnType == float.class) {
mv.visitVarInsn(FLOAD, rsindex);
mv.visitInsn(FRETURN);
} else if (returnType == double.class) {
mv.visitVarInsn(DLOAD, rsindex);
mv.visitInsn(DRETURN);
} else {
mv.visitVarInsn(ILOAD, rsindex);
mv.visitInsn(IRETURN);
}
} else {
mv.visitVarInsn(ALOAD, rsindex);
mv.visitInsn(ARETURN);
}
mv.visitLabel(clientLabel);
//---------------------------- 调用samerun ---------------------------------
mv.visitVarInsn(ILOAD, 2); //读取 samerunnable
Label sameLabel = new Label();
mv.visitJumpInsn(IFEQ, sameLabel); //判断 samerunnable
mv.visitVarInsn(ALOAD, 0);//调用 _client
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
final int preparams = 3; //调用selfrunnable之前的参数个数; _client
pushInt(mv, index); //第几个 SncpAction
pushInt(mv, paramtypes.length + preparams); //参数总数量
mv.visitTypeInsn(ANEWARRAY, "java/lang/Object");
mv.visitInsn(DUP);
mv.visitInsn(ICONST_0);
mv.visitInsn(ICONST_1); //第一个参数 selfrunnable
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
mv.visitInsn(DUP);
mv.visitInsn(ICONST_1);
mv.visitInsn(ICONST_0); //第二个参数 samerunnable
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
mv.visitInsn(DUP);
mv.visitInsn(ICONST_2);
mv.visitInsn(ICONST_0); //第三个参数 diffrunnable
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
int insn = 3; //空3给selfrunnable、samerunnable、diffrunnable
for (int j = 0; j < paramtypes.length; j++) {
final Class pt = paramtypes[j];
mv.visitInsn(DUP);
insn++;
pushInt(mv, j + 3);
if (pt.isPrimitive()) {
if (pt == long.class) {
mv.visitVarInsn(LLOAD, insn++);
} else if (pt == float.class) {
mv.visitVarInsn(FLOAD, insn++);
} else if (pt == double.class) {
mv.visitVarInsn(DLOAD, insn++);
} else {
mv.visitVarInsn(ILOAD, insn);
}
Class bigclaz = java.lang.reflect.Array.get(java.lang.reflect.Array.newInstance(pt, 1), 0).getClass();
mv.visitMethodInsn(INVOKESTATIC, bigclaz.getName().replace('.', '/'), "valueOf", "(" + Type.getDescriptor(pt) + ")" + Type.getDescriptor(bigclaz), false);
} else {
mv.visitVarInsn(ALOAD, insn);
}
mv.visitInsn(AASTORE);
}
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteSameGroup" : "remoteSameGroup", "(I[Ljava/lang/Object;)V", false);
mv.visitLabel(sameLabel);
//---------------------------- 调用diffrun ---------------------------------
mv.visitVarInsn(ILOAD, 3); //读取 diffrunnable
Label diffLabel = new Label();
mv.visitJumpInsn(IFEQ, diffLabel); //判断 diffrunnable
mv.visitVarInsn(ALOAD, 0);
mv.visitFieldInsn(GETFIELD, newDynName, FIELDPREFIX + "_client", clientDesc);
pushInt(mv, index); //第几个 SncpAction
pushInt(mv, paramtypes.length + preparams); //参数总数量
mv.visitTypeInsn(ANEWARRAY, "java/lang/Object");
mv.visitInsn(DUP);
mv.visitInsn(ICONST_0);
mv.visitInsn(ICONST_1); //第一个参数 samerunnable
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
mv.visitInsn(DUP);
mv.visitInsn(ICONST_1);
mv.visitInsn(ICONST_1); //第二个参数 diffrunnable
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
mv.visitInsn(DUP);
mv.visitInsn(ICONST_2);
mv.visitInsn(ICONST_0); //第二个参数 diffrunnable
mv.visitMethodInsn(INVOKESTATIC, "java/lang/Boolean", "valueOf", "(Z)Ljava/lang/Boolean;", false);
mv.visitInsn(AASTORE);
insn = 3;//空3给selfrunnable、samerunnable、diffrunnable
for (int j = 0; j < paramtypes.length; j++) {
final Class pt = paramtypes[j];
mv.visitInsn(DUP);
insn++;
pushInt(mv, j + 3);
if (pt.isPrimitive()) {
if (pt == long.class) {
mv.visitVarInsn(LLOAD, insn++);
} else if (pt == float.class) {
mv.visitVarInsn(FLOAD, insn++);
} else if (pt == double.class) {
mv.visitVarInsn(DLOAD, insn++);
} else {
mv.visitVarInsn(ILOAD, insn);
}
Class bigclaz = java.lang.reflect.Array.get(java.lang.reflect.Array.newInstance(pt, 1), 0).getClass();
mv.visitMethodInsn(INVOKESTATIC, bigclaz.getName().replace('.', '/'), "valueOf", "(" + Type.getDescriptor(pt) + ")" + Type.getDescriptor(bigclaz), false);
} else {
mv.visitVarInsn(ALOAD, insn);
}
mv.visitInsn(AASTORE);
}
mv.visitMethodInsn(INVOKEVIRTUAL, clientName, mrun.async() ? "asyncRemoteDiffGroup" : "remoteDiffGroup", "(I[Ljava/lang/Object;)V", false);
mv.visitLabel(diffLabel);
if (returnType == void.class) {
mv.visitInsn(RETURN);
} else if (returnType.isPrimitive()) {
if (returnType == long.class) {
mv.visitVarInsn(LLOAD, rsindex);
mv.visitInsn(LRETURN);
} else if (returnType == float.class) {
mv.visitVarInsn(FLOAD, rsindex);
mv.visitInsn(FRETURN);
} else if (returnType == double.class) {
mv.visitVarInsn(DLOAD, rsindex);
mv.visitInsn(DRETURN);
} else {
mv.visitVarInsn(ILOAD, rsindex);
mv.visitInsn(IRETURN);
}
} else {
mv.visitVarInsn(ALOAD, rsindex);
mv.visitInsn(ARETURN);
}
mv.visitMaxs(Math.max(varindex, 10), varindex + 4);
mv.visitEnd();
}
}
cw.visitEnd();
byte[] bytes = cw.toByteArray();
Class<?> newClazz = new ClassLoader(loader) {
@@ -823,29 +494,19 @@ public abstract class Sncp {
*
* private SncpClient _redkale_client;
*
* &#64;SncpDyn(remote = false, index = 0)
* public void _redkale_createSomeThing(boolean selfrunnable, boolean samerunnable, boolean diffrunnable, TestBean bean){
* _redkale_client.remote(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(1, selfrunnable, samerunnable, diffrunnable, id);
* }
*
* &#64;Override
* public void createSomeThing(TestBean bean){
* _redkale_client.remote(2, bean);
* _redkale_client.remote(0, bean);
* }
*
* &#64;Override
* public String findSomeThing(){
* return _redkale_client.remote(3);
* return _redkale_client.remote(1);
* }
*
* &#64;Override
* public String updateSomeThing(String id){
* return _redkale_client.remote(4, id);
* return _redkale_client.remote(2, id);
* }
* }
* </pre></blockquote>

33
src/org/redkale/service/RpcMultiRun.java
View File

@@ -1,33 +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.service;
import java.lang.annotation.*;
import static java.lang.annotation.ElementType.*;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
/**
* RpcMultiRun 只对本地模式Service有效
*
* <p>
* 详情见: https://redkale.org
*
* @author zhangjx
*/
@Inherited
@Documented
@Target({METHOD})
@Retention(RUNTIME)
public @interface RpcMultiRun {
boolean selfrun() default true; //当前本地实例是否运行指定操作；只有当指定操作的方法的返回值为void时，该值才能为true，否则忽略。
boolean samerun() default true; //是否同组节点运行指定操作
boolean diffrun() default true; //是否不同组节点运行指定操作
boolean async() default true; //分布式运行是否采用异步模式
}

124
src/org/redkale/source/CacheMemorySource.java
View File

@@ -381,7 +381,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
@SuppressWarnings("unchecked")
public V getAndRefresh(String key, final int expireSeconds) {
if (key == null) return null;
@@ -400,7 +400,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
@SuppressWarnings("unchecked")
public String getStringAndRefresh(String key, final int expireSeconds) {
if (key == null) return null;
@@ -412,7 +412,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public long getLongAndRefresh(String key, final int expireSeconds, long defValue) {
if (key == null) return defValue;
CacheEntry entry = container.get(key);
@@ -424,7 +424,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<V> getAndRefreshAsync(final String key, final int expireSeconds) {
return CompletableFuture.supplyAsync(() -> getAndRefresh(key, expireSeconds), getExecutor());
}
@@ -435,19 +435,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<String> getStringAndRefreshAsync(final String key, final int expireSeconds) {
return CompletableFuture.supplyAsync(() -> getStringAndRefresh(key, expireSeconds), getExecutor());
}
@Override
@RpcMultiRun
public CompletableFuture<Long> getLongAndRefreshAsync(final String key, final int expireSeconds, long defValue) {
return CompletableFuture.supplyAsync(() -> getLongAndRefresh(key, expireSeconds, defValue), getExecutor());
}
@Override
@RpcMultiRun
public void refresh(String key, final int expireSeconds) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -457,7 +457,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> refreshAsync(final String key, final int expireSeconds) {
return CompletableFuture.runAsync(() -> refresh(key, expireSeconds), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -476,7 +476,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void set(String key, V value) {
set(CacheEntryType.OBJECT, key, value);
}
@@ -487,19 +487,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void setString(String key, String value) {
set(CacheEntryType.STRING, key, value);
}
@Override
@RpcMultiRun
public void setLong(String key, long value) {
set(CacheEntryType.LONG, key, value);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> setAsync(String key, V value) {
return CompletableFuture.runAsync(() -> set(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -510,13 +510,13 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> setStringAsync(String key, String value) {
return CompletableFuture.runAsync(() -> setString(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> setLongAsync(String key, long value) {
return CompletableFuture.runAsync(() -> setLong(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -535,7 +535,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void set(int expireSeconds, String key, V value) {
set(CacheEntryType.OBJECT, expireSeconds, key, value);
}
@@ -546,19 +546,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void setString(int expireSeconds, String key, String value) {
set(CacheEntryType.STRING, expireSeconds, key, value);
}
@Override
@RpcMultiRun
public void setLong(int expireSeconds, String key, long value) {
set(CacheEntryType.LONG, expireSeconds, key, value);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> setAsync(int expireSeconds, String key, V value) {
return CompletableFuture.runAsync(() -> set(expireSeconds, key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -569,19 +569,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> setStringAsync(int expireSeconds, String key, String value) {
return CompletableFuture.runAsync(() -> setString(expireSeconds, key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> setLongAsync(int expireSeconds, String key, long value) {
return CompletableFuture.runAsync(() -> setLong(expireSeconds, key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public void setExpireSeconds(String key, int expireSeconds) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -590,32 +590,32 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> setExpireSecondsAsync(final String key, final int expireSeconds) {
return CompletableFuture.runAsync(() -> setExpireSeconds(key, expireSeconds), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public void remove(String key) {
if (key == null) return;
container.remove(key);
}
@Override
@RpcMultiRun
public long incr(final String key) {
return incr(key, 1);
}
@Override
@RpcMultiRun
public CompletableFuture<Long> incrAsync(final String key) {
return CompletableFuture.supplyAsync(() -> incr(key), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public long incr(final String key, long num) {
CacheEntry entry = container.get(key);
if (entry == null) {
@@ -631,37 +631,37 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Long> incrAsync(final String key, long num) {
return CompletableFuture.supplyAsync(() -> incr(key, num), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public long decr(final String key) {
return incr(key, -1);
}
@Override
@RpcMultiRun
public CompletableFuture<Long> decrAsync(final String key) {
return CompletableFuture.supplyAsync(() -> decr(key), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public long decr(final String key, long num) {
return incr(key, -num);
}
@Override
@RpcMultiRun
public CompletableFuture<Long> decrAsync(final String key, long num) {
return CompletableFuture.supplyAsync(() -> decr(key, num), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> removeAsync(final String key) {
return CompletableFuture.runAsync(() -> remove(key), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -763,7 +763,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public Collection<V> getCollectionAndRefresh(final String key, final int expireSeconds) {
return (Collection<V>) getAndRefresh(key, expireSeconds);
}
@@ -774,7 +774,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public Collection<String> getStringCollectionAndRefresh(final String key, final int expireSeconds) {
return (Collection<String>) getAndRefresh(key, expireSeconds);
}
@@ -824,13 +824,13 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public Collection<Long> getLongCollectionAndRefresh(final String key, final int expireSeconds) {
return (Collection<Long>) getAndRefresh(key, expireSeconds);
}
@Override
@RpcMultiRun
public CompletableFuture<Collection<V>> getCollectionAndRefreshAsync(final String key, final int expireSeconds) {
return CompletableFuture.supplyAsync(() -> getCollectionAndRefresh(key, expireSeconds), getExecutor());
}
@@ -841,13 +841,13 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Collection<String>> getStringCollectionAndRefreshAsync(final String key, final int expireSeconds) {
return CompletableFuture.supplyAsync(() -> getStringCollectionAndRefresh(key, expireSeconds), getExecutor());
}
@Override
@RpcMultiRun
public CompletableFuture<Collection<Long>> getLongCollectionAndRefreshAsync(final String key, final int expireSeconds) {
return CompletableFuture.supplyAsync(() -> getLongCollectionAndRefresh(key, expireSeconds), getExecutor());
}
@@ -867,7 +867,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void appendListItem(String key, V value) {
appendListItem(CacheEntryType.OBJECT_LIST, key, value);
}
@@ -878,19 +878,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void appendStringListItem(String key, String value) {
appendListItem(CacheEntryType.STRING_LIST, key, value);
}
@Override
@RpcMultiRun
public void appendLongListItem(String key, long value) {
appendListItem(CacheEntryType.LONG_LIST, key, value);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> appendListItemAsync(final String key, final V value) {
return CompletableFuture.runAsync(() -> appendListItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -901,19 +901,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> appendStringListItemAsync(final String key, final String value) {
return CompletableFuture.runAsync(() -> appendStringListItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> appendLongListItemAsync(final String key, final long value) {
return CompletableFuture.runAsync(() -> appendLongListItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public void removeListItem(String key, V value) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -930,7 +930,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void removeStringListItem(String key, String value) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -939,7 +939,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void removeLongListItem(String key, long value) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -948,7 +948,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> removeListItemAsync(final String key, final V value) {
return CompletableFuture.runAsync(() -> removeListItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -959,13 +959,13 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> removeStringListItemAsync(final String key, final String value) {
return CompletableFuture.runAsync(() -> removeStringListItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> removeLongListItemAsync(final String key, final long value) {
return CompletableFuture.runAsync(() -> removeLongListItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -985,7 +985,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void appendSetItem(String key, V value) {
appendSetItem(CacheEntryType.OBJECT_SET, key, value);
}
@@ -996,19 +996,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void appendStringSetItem(String key, String value) {
appendSetItem(CacheEntryType.OBJECT_SET, key, value);
}
@Override
@RpcMultiRun
public void appendLongSetItem(String key, long value) {
appendSetItem(CacheEntryType.OBJECT_SET, key, value);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> appendSetItemAsync(final String key, final V value) {
return CompletableFuture.runAsync(() -> appendSetItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -1019,19 +1019,19 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> appendStringSetItemAsync(final String key, final String value) {
return CompletableFuture.runAsync(() -> appendStringSetItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> appendLongSetItemAsync(final String key, final long value) {
return CompletableFuture.runAsync(() -> appendLongSetItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public void removeSetItem(String key, V value) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -1048,7 +1048,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void removeStringSetItem(String key, String value) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -1057,7 +1057,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public void removeLongSetItem(String key, long value) {
if (key == null) return;
CacheEntry entry = container.get(key);
@@ -1066,7 +1066,7 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> removeSetItemAsync(final String key, final V value) {
return CompletableFuture.runAsync(() -> removeSetItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@@ -1077,13 +1077,13 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
}
@Override
@RpcMultiRun
public CompletableFuture<Void> removeStringSetItemAsync(final String key, final String value) {
return CompletableFuture.runAsync(() -> removeStringSetItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}
@Override
@RpcMultiRun
public CompletableFuture<Void> removeLongSetItemAsync(final String key, final long value) {
return CompletableFuture.runAsync(() -> removeLongSetItem(key, value), getExecutor()).whenComplete(futureCompleteConsumer);
}

3
test/org/redkale/test/sncp/SncpTestServiceImpl.java
View File

@@ -43,13 +43,11 @@ public class SncpTestServiceImpl implements SncpTestIService {
}
@Override
@RpcMultiRun
public long queryLongResult(String a, int b, long value) {
return value + 1;
}
@Override
@RpcMultiRun
public double queryDoubleResult(String a, int b, double value) {
return value + 1;
}
@@ -103,7 +101,6 @@ public class SncpTestServiceImpl implements SncpTestIService {
if (handler != null) handler.completed("result: " + bean, bean);
}
@RpcMultiRun
@Override
public String updateBean(@RpcCall(CallAttribute.class) SncpTestBean bean) {
bean.setId(System.currentTimeMillis());

8
test/org/redkale/test/sncp/_DynLocalSncpTestService.java
View File

@@ -17,12 +17,4 @@ public class _DynLocalSncpTestService extends SncpTestServiceImpl {
private SncpClient _redkale_client;
@SncpDyn(remote = false, index = 1)
public long _redkale_queryLongResult(boolean selfrunnable, boolean samerunnable, boolean diffrunnable, String a, int b, long value) {
long rs = super.queryLongResult(a, b, value);
if (_redkale_client == null) return rs;
if (samerunnable) _redkale_client.remoteSameGroup(1, true, false, false, a, b, value);
if (diffrunnable) _redkale_client.remoteDiffGroup(1, true, true, false, a, b, value);
return rs;
}
}