From 5785fe6aa60e8f3da01700dc03b54885e4392e05 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 16:11:50 +0800 Subject: [PATCH] WIP --- include/rest_rpc/rpc_client.hpp | 18 +++++++++ java/pom.xml | 5 +++ .../restrpc/client/AsyncRpcFunctionImpl.java | 2 +- .../main/java/org/restrpc/client/Codec.java | 38 +++++++++++++++++++ .../org/restrpc/client/NativeRpcClient.java | 30 ++++++++++++--- .../java/org/restrpc/client/RpcClient.java | 2 +- .../org/restrpc/test/BasicClientTest.java | 30 ++++++++++++++- jni/org_restrpc_client_NativeRpcClient.cc | 31 ++++++++++++--- jni/org_restrpc_client_NativeRpcClient.h | 4 +- 9 files changed, 143 insertions(+), 17 deletions(-) create mode 100644 java/src/main/java/org/restrpc/client/Codec.java diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp index 576200e..fde574c 100644 --- a/include/rest_rpc/rpc_client.hpp +++ b/include/rest_rpc/rpc_client.hpp @@ -248,6 +248,24 @@ namespace rest_rpc { return future; } + long wq_async_call(const std::string& encoded_func_name_and_args) { + auto p = std::make_shared>(); + std::future future = p->get_future(); + + uint64_t fu_id = 0; + { + std::unique_lock lock(cb_mtx_); + fu_id_++; + fu_id = fu_id_; + future_map_.emplace(fu_id, std::move(p)); + } + + msgpack::sbuffer sbuffer; + sbuffer.write(encoded_func_name_and_args.data(), encoded_func_name_and_args.size()); + write(fu_id, request_type::req_res, std::move(sbuffer)); + return fu_id; + } + template void async_call(const std::string& rpc_name, std::function cb, Args&& ... args) { if (!has_connected_) { diff --git a/java/pom.xml b/java/pom.xml index d1bfebf..7a31d99 100644 --- a/java/pom.xml +++ b/java/pom.xml @@ -29,6 +29,11 @@ testng 7.0.0 + + org.msgpack + msgpack-core + 0.8.21 + diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java index 105b590..7360484 100644 --- a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java @@ -54,6 +54,6 @@ public class AsyncRpcFunctionImpl implements AsyncRpcFunction { } private CompletableFuture internalInvoke(Object[] args) { - return rpcClient.invoke(args); + return rpcClient.invoke(funcName, args); } } diff --git a/java/src/main/java/org/restrpc/client/Codec.java b/java/src/main/java/org/restrpc/client/Codec.java new file mode 100644 index 0000000..a0d090b --- /dev/null +++ b/java/src/main/java/org/restrpc/client/Codec.java @@ -0,0 +1,38 @@ +package org.restrpc.client; + +import org.msgpack.core.MessageBufferPacker; +import org.msgpack.core.MessagePack; +import java.io.IOException; + +public class Codec { + public byte[] encode(String funcName, Object[] args) throws IOException { + // assert args != nullptr. + + MessageBufferPacker messagePacker = MessagePack.newDefaultBufferPacker(); + messagePacker.packArrayHeader(1 + args.length); + messagePacker.packString(funcName); + + for (Object arg : args) { + if (arg == null) { + messagePacker.packNil(); + continue; + } + + final String argTypeName = arg.getClass().getName(); + switch (argTypeName) { + case "java.lang.Integer": + messagePacker.packInt((int) arg); + break; + case "java.lang.Long": + messagePacker.packLong((long) arg); + break; + case "java.lang.String": + messagePacker.packString((String) arg); + break; + default: + throw new RuntimeException("Unknown type" + argTypeName); + } + } + return messagePacker.toByteArray(); + } +} diff --git a/java/src/main/java/org/restrpc/client/NativeRpcClient.java b/java/src/main/java/org/restrpc/client/NativeRpcClient.java index 7cbd35d..7e2a230 100644 --- a/java/src/main/java/org/restrpc/client/NativeRpcClient.java +++ b/java/src/main/java/org/restrpc/client/NativeRpcClient.java @@ -1,5 +1,7 @@ package org.restrpc.client; +import java.io.IOException; +import java.util.HashMap; import java.util.concurrent.CompletableFuture; public class NativeRpcClient implements RpcClient { @@ -9,6 +11,10 @@ public class NativeRpcClient implements RpcClient { } private long rpcClientPointer = -1; + private Codec codec; + + private HashMap> localFutureCache = new HashMap<>(); + public NativeRpcClient() { rpcClientPointer = nativeNewRpcClient(); } @@ -18,6 +24,7 @@ public class NativeRpcClient implements RpcClient { throw new RuntimeException("no init"); } nativeConnect(rpcClientPointer, serverAddress); + codec = new Codec(); } public AsyncRpcFunction asyncFunc(String funcName) { @@ -27,14 +34,26 @@ public class NativeRpcClient implements RpcClient { return new AsyncRpcFunctionImpl(this, funcName); } - public CompletableFuture invoke(Object[] args) { + public CompletableFuture invoke(String funcName, Object[] args) { if (rpcClientPointer == -1) { throw new RuntimeException("no init"); } - nativeInvoke(rpcClientPointer, null); -// return nativeInvoke(rpcClientPointer, ); - return null; + byte[] encodedBytes = null; + try { + encodedBytes = codec.encode(funcName, args); + } catch (IOException e) { + throw new RuntimeException("..."); + } + + if (encodedBytes == null) { + return null; + } + + final long requestId = nativeInvoke(rpcClientPointer, encodedBytes); + CompletableFuture futureToReturn = new CompletableFuture<>(); + localFutureCache.put(requestId, futureToReturn); + return futureToReturn; } public void close() { @@ -46,12 +65,11 @@ public class NativeRpcClient implements RpcClient { this.rpcClientPointer = -1; } - private native long nativeNewRpcClient(); private native void nativeConnect(long rpcClientPointer, String serverAddress); - private native long nativeInvoke(long rpcClientPointer, byte[][] encodedFuncNameAndArgs); + private native long nativeInvoke(long rpcClientPointer, byte[] encodedFuncNameAndArgs); private native void nativeDestroy(long rpcClientPointer); } diff --git a/java/src/main/java/org/restrpc/client/RpcClient.java b/java/src/main/java/org/restrpc/client/RpcClient.java index c0c5a3e..58b37b6 100644 --- a/java/src/main/java/org/restrpc/client/RpcClient.java +++ b/java/src/main/java/org/restrpc/client/RpcClient.java @@ -8,7 +8,7 @@ public interface RpcClient { AsyncRpcFunction asyncFunc(String funcName); - CompletableFuture invoke(Object[] args); + CompletableFuture invoke(String funcName, Object[] args); void close(); } diff --git a/java/src/test/java/org/restrpc/test/BasicClientTest.java b/java/src/test/java/org/restrpc/test/BasicClientTest.java index 8516b6c..823d618 100644 --- a/java/src/test/java/org/restrpc/test/BasicClientTest.java +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -1,14 +1,42 @@ package org.restrpc.test; +import org.checkerframework.checker.units.qual.C; +import org.msgpack.core.MessagePack; +import org.restrpc.client.Codec; import org.restrpc.client.NativeRpcClient; import org.restrpc.client.RpcClient; import org.testng.annotations.Test; +import java.io.IOException; + public class BasicClientTest { @Test public void testBasic() { RpcClient rpcClient = new NativeRpcClient(); - rpcClient.asyncFunc("111").invoke(); + rpcClient.asyncFunc("111").invoke(2, 3); + } + + private String getClassStr(Object o) { + return o.getClass().getName(); + } + + @Test + public void testClass() { + int a = 3; + System.out.println("int -> " + getClassStr(a)); + long b = 3; + System.out.println("long -> " + getClassStr(b)); + String s = "3"; + System.out.println("string ->" + getClassStr(s)); + String c = null; + System.out.println("null str ->" + getClassStr(c)); + } + + @Test + public void testEncode() throws IOException { + Codec codec = new Codec(); + Object[] args = new Object[] {2, 3}; + byte[] bs = codec.encode("add", args); } } diff --git a/jni/org_restrpc_client_NativeRpcClient.cc b/jni/org_restrpc_client_NativeRpcClient.cc index a94a323..ec3d155 100644 --- a/jni/org_restrpc_client_NativeRpcClient.cc +++ b/jni/org_restrpc_client_NativeRpcClient.cc @@ -8,14 +8,28 @@ #ifdef __cplusplus extern "C" { #endif + + +inline std::string JavaByteArrayToNativeString(JNIEnv *env, const jbyteArray &bytes) { + const auto size = env->GetArrayLength(bytes); + std::string str(size, 0); + env->GetByteArrayRegion(bytes, 0, size, reinterpret_cast(&str.front())); + return str; +} + +// TODO(qwang) +// JavaByteArrayToSBuffer + + /* * Class: org_restrpc_client_NativeRpcClient * Method: nativeNewRpcClient * Signature: ()J */ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeNewRpcClient - (JNIEnv *, jobject) { - return 10009; + (JNIEnv *, jobject o) { + rest_rpc::rpc_client *native_rpc_client = new rest_rpc::rpc_client(); + return reinterpret_cast(native_rpc_client); } /* @@ -24,17 +38,22 @@ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeNewRpcClie * Signature: (JLjava/lang/String;)V */ JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeConnect -(JNIEnv *, jobject, jlong, jstring) { +(JNIEnv *, jobject o, jlong rpcClientPointer, jstring serverAddress) { + auto *native_rpc_client = reinterpret_cast(rpcClientPointer); + // TODO(qwang): Do not hard code this. + native_rpc_client->connect("127.0.0.1", 9000); } /* * Class: org_restrpc_client_NativeRpcClient * Method: nativeInvoke - * Signature: (J[[B)J + * Signature: (J[B)J */ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeInvoke - (JNIEnv *, jobject, jlong, jobjectArray) { - return 20009; + (JNIEnv *env, jobject o, jlong rpcClientPointer, jbyteArray encodedBytes) { + auto *native_rpc_client = reinterpret_cast(rpcClientPointer); + auto encodedFuncNameAndArgs = JavaByteArrayToNativeString(env, encodedBytes); + return native_rpc_client->wq_async_call(encodedFuncNameAndArgs); } /* diff --git a/jni/org_restrpc_client_NativeRpcClient.h b/jni/org_restrpc_client_NativeRpcClient.h index eb7e772..850d61f 100644 --- a/jni/org_restrpc_client_NativeRpcClient.h +++ b/jni/org_restrpc_client_NativeRpcClient.h @@ -26,10 +26,10 @@ JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeConnect /* * Class: org_restrpc_client_NativeRpcClient * Method: nativeInvoke - * Signature: (J[[B)J + * Signature: (J[B)J */ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeInvoke - (JNIEnv *, jobject, jlong, jobjectArray); + (JNIEnv *, jobject, jlong, jbyteArray); /* * Class: org_restrpc_client_NativeRpcClient