From 9ae931cd65426f484f87ba3757aec13ffb2f3c98 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 12:44:51 +0800 Subject: [PATCH 01/12] Fix --- examples/CMakeLists.txt | 30 ++++++++++++++++++++++++++++++ examples/client/CMakeLists.txt | 26 -------------------------- examples/server/CMakeLists.txt | 26 -------------------------- 3 files changed, 30 insertions(+), 52 deletions(-) create mode 100644 examples/CMakeLists.txt delete mode 100644 examples/client/CMakeLists.txt delete mode 100644 examples/server/CMakeLists.txt diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt new file mode 100644 index 0000000..49a8f06 --- /dev/null +++ b/examples/CMakeLists.txt @@ -0,0 +1,30 @@ +cmake_minimum_required(VERSION 3.1) +project(example) + +set(ASIO_STANDALONE 1) +set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11") + +SET(ENABLE_SSL OFF) + +if (ENABLE_SSL) + add_definitions(-DCINATRA_ENABLE_SSL) + message(STATUS "Use SSL") +endif() + +#find_package(Boost COMPONENTS system filesystem REQUIRED) +include_directories( + # "/usr/local/include" + "../include" + "/Users/qingwang/workspace/opensource/rest_rpc/third/msgpack/include") + +INCLUDE_DIRECTORIES(SYSTEM "/Users/qingwang/workspace/opensource/dousi/core/build/external/boost/src/boost_ep") +INCLUDE_DIRECTORIES(SYSTEM "/Users/qingwang/workspace/opensource/rest_rpc/third/msgpack/include") + +add_executable(basic_server server/main.cpp) +add_executable(basic_client client/main.cpp) + +if (ENABLE_SSL) + target_link_libraries(basic_server ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) +else() + target_link_libraries(basic_server ${Boost_LIBRARIES}) +endif() diff --git a/examples/client/CMakeLists.txt b/examples/client/CMakeLists.txt deleted file mode 100644 index 3e59292..0000000 --- a/examples/client/CMakeLists.txt +++ /dev/null @@ -1,26 +0,0 @@ -cmake_minimum_required(VERSION 3.1) -project(basic_client) - -set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11") - -SET(ENABLE_SSL OFF) - -if (ENABLE_SSL) - add_definitions(-DCINATRA_ENABLE_SSL) - message(STATUS "Use SSL") -endif() - -find_package(Boost COMPONENTS system REQUIRED) -include_directories( - "/usr/local/include" - "../../include" - "../../third/msgpack/include" -) - -add_executable(basic_client main.cpp) - -if (ENABLE_SSL) - target_link_libraries(basic_client ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) -else() - target_link_libraries(basic_client ${Boost_LIBRARIES}) -endif() \ No newline at end of file diff --git a/examples/server/CMakeLists.txt b/examples/server/CMakeLists.txt deleted file mode 100644 index fea606b..0000000 --- a/examples/server/CMakeLists.txt +++ /dev/null @@ -1,26 +0,0 @@ -cmake_minimum_required(VERSION 3.1) -project(basic_server) - -set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11") - -SET(ENABLE_SSL OFF) - -if (ENABLE_SSL) - add_definitions(-DCINATRA_ENABLE_SSL) - message(STATUS "Use SSL") -endif() - -find_package(Boost COMPONENTS system filesystem REQUIRED) -include_directories( - "/usr/local/include" - "../../include" - "../../third/msgpack/include" -) - -add_executable(basic_server main.cpp) - -if (ENABLE_SSL) - target_link_libraries(basic_server ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) -else() - target_link_libraries(basic_server ${Boost_LIBRARIES}) -endif() From b66ca0cde9ecdb2e18fb7a16db9dab344dcedb2a Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 13:53:46 +0800 Subject: [PATCH 02/12] Fix --- .gitignore | 6 ++ java/pom.xml | 40 +++++++++++ .../org/restrpc/client/AsyncRpcFunction.java | 28 ++++++++ .../restrpc/client/AsyncRpcFunctionImpl.java | 59 ++++++++++++++++ .../java/org/restrpc/client/JniUtils.java | 67 +++++++++++++++++++ .../org/restrpc/client/LibraryFileUtils.java | 48 +++++++++++++ .../org/restrpc/client/NativeRpcClient.java | 57 ++++++++++++++++ .../java/org/restrpc/client/RpcClient.java | 14 ++++ .../org/restrpc/test/BasicClientTest.java | 14 ++++ jni/CMakeLists.txt | 27 ++++++++ jni/org_restrpc_client_NativeRpcClient.cc | 52 ++++++++++++++ jni/org_restrpc_client_NativeRpcClient.h | 45 +++++++++++++ 12 files changed, 457 insertions(+) create mode 100644 java/pom.xml create mode 100644 java/src/main/java/org/restrpc/client/AsyncRpcFunction.java create mode 100644 java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java create mode 100644 java/src/main/java/org/restrpc/client/JniUtils.java create mode 100644 java/src/main/java/org/restrpc/client/LibraryFileUtils.java create mode 100644 java/src/main/java/org/restrpc/client/NativeRpcClient.java create mode 100644 java/src/main/java/org/restrpc/client/RpcClient.java create mode 100644 java/src/test/java/org/restrpc/test/BasicClientTest.java create mode 100644 jni/CMakeLists.txt create mode 100644 jni/org_restrpc_client_NativeRpcClient.cc create mode 100644 jni/org_restrpc_client_NativeRpcClient.h diff --git a/.gitignore b/.gitignore index dafb7b0..5b3d0a7 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,9 @@ cmake_install.cmake # idea .idea/ + +.DS_Store +*iml +target/ + +build/ diff --git a/java/pom.xml b/java/pom.xml new file mode 100644 index 0000000..d1bfebf --- /dev/null +++ b/java/pom.xml @@ -0,0 +1,40 @@ + + + 4.0.0 + + org.restrpc + restrpc + 1.0-SNAPSHOT + + + + commons-io + commons-io + 2.5 + + + com.google.guava + guava + 29.0-jre + + + org.slf4j + slf4j-api + 1.7.21 + + + org.testng + testng + 7.0.0 + + + + + + native_dependencies + + + + diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java new file mode 100644 index 0000000..a771616 --- /dev/null +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java @@ -0,0 +1,28 @@ +package org.restrpc.client; + +import java.util.concurrent.CompletableFuture; + +public interface AsyncRpcFunction { + + CompletableFuture invoke(); + + + CompletableFuture invoke(Arg1Type arg1); + + + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2); + + + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3); + + + + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4); + + + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5); + + + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6); + +} diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java new file mode 100644 index 0000000..105b590 --- /dev/null +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java @@ -0,0 +1,59 @@ +package org.restrpc.client; + +import java.util.concurrent.CompletableFuture; + +public class AsyncRpcFunctionImpl implements AsyncRpcFunction { + + private RpcClient rpcClient; + + private String funcName; + + public AsyncRpcFunctionImpl(RpcClient rpcClient, String funcName) { + this.rpcClient = rpcClient; + this.funcName = funcName; + } + + public CompletableFuture invoke() { + return internalInvoke(new Object[0]); + } + + public CompletableFuture invoke(Arg1Type arg1) { + Object[] args = new Object[] {arg1}; + return internalInvoke(args); + } + + public + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2) { + Object[] args = new Object[] {arg1, arg2}; + return internalInvoke(args); + } + + public + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) { + Object[] args = new Object[] {arg1, arg2, arg3}; + return internalInvoke(args); + } + + + public + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) { + Object[] args = new Object[] {arg1, arg2, arg3, arg4}; + return internalInvoke(args); + } + + public + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5) { + Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5}; + return internalInvoke(args); + } + + public + CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6) { + Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5, arg6}; + return internalInvoke(args); + } + + private CompletableFuture internalInvoke(Object[] args) { + return rpcClient.invoke(args); + } +} diff --git a/java/src/main/java/org/restrpc/client/JniUtils.java b/java/src/main/java/org/restrpc/client/JniUtils.java new file mode 100644 index 0000000..3ece0da --- /dev/null +++ b/java/src/main/java/org/restrpc/client/JniUtils.java @@ -0,0 +1,67 @@ +package org.restrpc.client; + +import com.google.common.base.Strings; +import com.google.common.collect.Sets; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.lang.reflect.Field; +import java.util.Set; + +public class JniUtils { + + private static final Logger LOGGER = LoggerFactory.getLogger(JniUtils.class); + + private static Set loadedLibs = Sets.newHashSet(); + + /** + * Loads the native library specified by the libraryName argument. + * The libraryName argument must not contain any platform specific + * prefix, file extension or path. + * + * @param libraryName the name of the library. + */ + public static synchronized void loadLibrary(String libraryName) { + if (!loadedLibs.contains(libraryName)) { + LOGGER.debug("Loading native library {}.", libraryName); + // Load native library. + String fileName = System.mapLibraryName(libraryName); + final File file = LibraryFileUtils.getFile("/tmp/restrpc", fileName); + System.load(file.getAbsolutePath()); + LOGGER.debug("Native library loaded."); + resetLibraryPath(file.getAbsolutePath()); + loadedLibs.add(libraryName); + } + } + + /** + * This is a hack to reset library path at runtime. + */ + public static synchronized void resetLibraryPath(String libPath) { + if (Strings.isNullOrEmpty(libPath)) { + return; + } + String path = System.getProperty("java.library.path"); + String separator = System.getProperty("path.separator"); + if (Strings.isNullOrEmpty(path)) { + path = ""; + } else { + path += separator; + } + path += String.join(separator, libPath); + + // This is a hack to reset library path at runtime, + // see https://stackoverflow.com/questions/15409223/. + System.setProperty("java.library.path", path); + // Set sys_paths to null so that java.library.path will be re-evaluated next time it is needed. + final Field sysPathsField; + try { + sysPathsField = ClassLoader.class.getDeclaredField("sys_paths"); + sysPathsField.setAccessible(true); + sysPathsField.set(null, null); + } catch (NoSuchFieldException | IllegalAccessException e) { + LOGGER.error("Failed to set library path.", e); + } + } +} diff --git a/java/src/main/java/org/restrpc/client/LibraryFileUtils.java b/java/src/main/java/org/restrpc/client/LibraryFileUtils.java new file mode 100644 index 0000000..8d95ab3 --- /dev/null +++ b/java/src/main/java/org/restrpc/client/LibraryFileUtils.java @@ -0,0 +1,48 @@ +package org.restrpc.client; + +import com.google.common.base.Preconditions; +import org.apache.commons.io.FileUtils; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.RandomAccessFile; +import java.nio.channels.FileLock; +import java.nio.file.Files; +import java.nio.file.Paths; + +public class LibraryFileUtils { + + public static final String RESTRPC_LIBRARY_NAME = "restrpc_jni"; + + public static File getFile(String destDir, String fileName) { + final File dir = new File(destDir); + if (!dir.exists()) { + try { + FileUtils.forceMkdir(dir); + } catch (IOException e) { + throw new RuntimeException("Couldn't make directory: " + dir.getAbsolutePath(), e); + } + } + String lockFilePath = destDir + File.separator + "file_lock"; + try (FileLock ignored = new RandomAccessFile(lockFilePath, "rw") + .getChannel().lock()) { + File file = new File(String.format("%s/%s", destDir, fileName)); + if (file.exists()) { + return file; + } + + // File does not exist. + try (InputStream is = LibraryFileUtils.class.getResourceAsStream("/" + fileName)) { + Preconditions.checkNotNull(is, "{} doesn't exist.", fileName); + Files.copy(is, Paths.get(file.getCanonicalPath())); + } catch (IOException e) { + throw new RuntimeException("Couldn't get temp file from resource " + fileName, e); + } + return file; + } catch (IOException e) { + throw new RuntimeException(e); + } + } + +} diff --git a/java/src/main/java/org/restrpc/client/NativeRpcClient.java b/java/src/main/java/org/restrpc/client/NativeRpcClient.java new file mode 100644 index 0000000..7cbd35d --- /dev/null +++ b/java/src/main/java/org/restrpc/client/NativeRpcClient.java @@ -0,0 +1,57 @@ +package org.restrpc.client; + +import java.util.concurrent.CompletableFuture; + +public class NativeRpcClient implements RpcClient { + + static { + JniUtils.loadLibrary("restrpc_jni"); + } + private long rpcClientPointer = -1; + + public NativeRpcClient() { + rpcClientPointer = nativeNewRpcClient(); + } + + public void connect(String serverAddress) { + if (rpcClientPointer == -1) { + throw new RuntimeException("no init"); + } + nativeConnect(rpcClientPointer, serverAddress); + } + + public AsyncRpcFunction asyncFunc(String funcName) { + if (funcName == null) { + throw new NullPointerException("Rpc function name should be null."); + } + return new AsyncRpcFunctionImpl(this, funcName); + } + + public CompletableFuture invoke(Object[] args) { + if (rpcClientPointer == -1) { + throw new RuntimeException("no init"); + } + + nativeInvoke(rpcClientPointer, null); +// return nativeInvoke(rpcClientPointer, ); + return null; + } + + public void close() { + if (rpcClientPointer == -1) { + throw new RuntimeException("no init"); + } + + nativeDestroy(rpcClientPointer); + 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 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 new file mode 100644 index 0000000..c0c5a3e --- /dev/null +++ b/java/src/main/java/org/restrpc/client/RpcClient.java @@ -0,0 +1,14 @@ +package org.restrpc.client; + +import java.util.concurrent.CompletableFuture; + +public interface RpcClient { + + void connect(String serverAddress); + + AsyncRpcFunction asyncFunc(String funcName); + + CompletableFuture invoke(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 new file mode 100644 index 0000000..8516b6c --- /dev/null +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -0,0 +1,14 @@ +package org.restrpc.test; + +import org.restrpc.client.NativeRpcClient; +import org.restrpc.client.RpcClient; +import org.testng.annotations.Test; + +public class BasicClientTest { + + @Test + public void testBasic() { + RpcClient rpcClient = new NativeRpcClient(); + rpcClient.asyncFunc("111").invoke(); + } +} diff --git a/jni/CMakeLists.txt b/jni/CMakeLists.txt new file mode 100644 index 0000000..77e1608 --- /dev/null +++ b/jni/CMakeLists.txt @@ -0,0 +1,27 @@ +cmake_minimum_required(VERSION 3.1) +project(example) + +set(ASIO_STANDALONE 1) +set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11") + +SET(ENABLE_SSL OFF) + +if (ENABLE_SSL) + add_definitions(-DCINATRA_ENABLE_SSL) + message(STATUS "Use SSL") +endif() + +find_package(JNI REQUIRED) +#find_package(Boost COMPONENTS system filesystem REQUIRED) +include_directories( + # "/usr/local/include" + "../include" + "../jni" + ${JNI_INCLUDE_DIRS} +# "/usr/local/opt/openjdk/include/" + "/Users/qingwang/workspace/opensource/rest_rpc/third/msgpack/include") + +INCLUDE_DIRECTORIES(SYSTEM "/Users/qingwang/workspace/opensource/dousi/core/build/external/boost/src/boost_ep") +INCLUDE_DIRECTORIES(SYSTEM "/Users/qingwang/workspace/opensource/rest_rpc/third/msgpack/include") + +add_library(restrpc_jni SHARED org_restrpc_client_NativeRpcClient.cc) diff --git a/jni/org_restrpc_client_NativeRpcClient.cc b/jni/org_restrpc_client_NativeRpcClient.cc new file mode 100644 index 0000000..a94a323 --- /dev/null +++ b/jni/org_restrpc_client_NativeRpcClient.cc @@ -0,0 +1,52 @@ +/* DO NOT EDIT THIS FILE - it is machine generated */ + +#include "org_restrpc_client_NativeRpcClient.h" +#include + +#include + +#ifdef __cplusplus +extern "C" { +#endif +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeNewRpcClient + * Signature: ()J + */ +JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeNewRpcClient + (JNIEnv *, jobject) { + return 10009; +} + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeConnect + * Signature: (JLjava/lang/String;)V + */ +JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeConnect +(JNIEnv *, jobject, jlong, jstring) { +} + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeInvoke + * Signature: (J[[B)J + */ +JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeInvoke + (JNIEnv *, jobject, jlong, jobjectArray) { + return 20009; +} + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeDestroy + * Signature: (J)V + */ +JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeDestroy +(JNIEnv *, jobject, jlong) { + +} + +#ifdef __cplusplus +} +#endif diff --git a/jni/org_restrpc_client_NativeRpcClient.h b/jni/org_restrpc_client_NativeRpcClient.h new file mode 100644 index 0000000..eb7e772 --- /dev/null +++ b/jni/org_restrpc_client_NativeRpcClient.h @@ -0,0 +1,45 @@ +/* DO NOT EDIT THIS FILE - it is machine generated */ +#include +/* Header for class org_restrpc_client_NativeRpcClient */ + +#ifndef _Included_org_restrpc_client_NativeRpcClient +#define _Included_org_restrpc_client_NativeRpcClient +#ifdef __cplusplus +extern "C" { +#endif +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeNewRpcClient + * Signature: ()J + */ +JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeNewRpcClient + (JNIEnv *, jobject); + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeConnect + * Signature: (JLjava/lang/String;)V + */ +JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeConnect + (JNIEnv *, jobject, jlong, jstring); + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeInvoke + * Signature: (J[[B)J + */ +JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeInvoke + (JNIEnv *, jobject, jlong, jobjectArray); + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeDestroy + * Signature: (J)V + */ +JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeDestroy + (JNIEnv *, jobject, jlong); + +#ifdef __cplusplus +} +#endif +#endif From 5785fe6aa60e8f3da01700dc03b54885e4392e05 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 16:11:50 +0800 Subject: [PATCH 03/12] 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 From 7e22d4b402b918e61ef43a500620e583c9533fca Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 20:24:52 +0800 Subject: [PATCH 04/12] Work around. --- examples/client/main.cpp | 16 +-- examples/server/main.cpp | 3 +- examples/server/qps.h | 2 +- include/rest_rpc/rpc_client.hpp | 117 ++++++++++++------ .../org/restrpc/client/AsyncRpcFunction.java | 36 +++--- .../restrpc/client/AsyncRpcFunctionImpl.java | 70 +++++------ .../main/java/org/restrpc/client/Codec.java | 49 +++++++- .../org/restrpc/client/NativeRpcClient.java | 30 ++++- .../java/org/restrpc/client/RestFuture.java | 38 ++++++ .../java/org/restrpc/client/RpcClient.java | 2 +- .../org/restrpc/test/BasicClientTest.java | 33 ++++- jni/org_restrpc_client_NativeRpcClient.cc | 67 +++++++++- jni/org_restrpc_client_NativeRpcClient.h | 3 + 13 files changed, 347 insertions(+), 119 deletions(-) create mode 100644 java/src/main/java/org/restrpc/client/RestFuture.java diff --git a/examples/client/main.cpp b/examples/client/main.cpp index 6511ae4..9c40940 100644 --- a/examples/client/main.cpp +++ b/examples/client/main.cpp @@ -16,8 +16,8 @@ void test_add() { } { - auto result = client.call("add", 1, 2); - std::cout << result << std::endl; +// auto result = client.call("add", 1, 2); +// std::cout << result << std::endl; } { @@ -622,12 +622,12 @@ void benchmark_test(){ int main() { // benchmark_test(); - test_sub1(); - test_connect(); - test_callback(); - test_echo(); - test_sync_client(); - test_async_client(); + test_add(); +// test_connect(); +// test_callback(); +// test_echo(); +// test_sync_client(); +// test_async_client(); //test_threads(); //test_sub(); //test_call_with_timeout(); diff --git a/examples/server/main.cpp b/examples/server/main.cpp index 4f7c7f7..05cfb58 100644 --- a/examples/server/main.cpp +++ b/examples/server/main.cpp @@ -6,7 +6,8 @@ using namespace rpc_service; #include "qps.h" struct dummy{ - int add(rpc_conn conn, int a, int b) { + int add(rpc_conn conn, int a, int b) { + std::cout << "ohhhhhhhhhhh invoked: a=" << a << ", b=" << b << std::endl; return a + b; } }; diff --git a/examples/server/qps.h b/examples/server/qps.h index 408a520..2afd304 100644 --- a/examples/server/qps.h +++ b/examples/server/qps.h @@ -12,7 +12,7 @@ public: qps() : counter_(0) { thd_ = std::thread([this] { while (!stop_) { - std::cout << "qps: " << counter_.load(std::memory_order_acquire) << '\n'; +// std::cout << "qps: " << counter_.load(std::memory_order_acquire) << '\n'; std::this_thread::sleep_for(std::chrono::seconds(1)); //counter_.store(0, std::memory_order_release); } diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp index fde574c..54928cc 100644 --- a/include/rest_rpc/rpc_client.hpp +++ b/include/rest_rpc/rpc_client.hpp @@ -3,14 +3,25 @@ #include #include #include +#include #include "use_asio.hpp" #include "client_util.hpp" #include "const_vars.h" #include "meta_util.hpp" +#include using namespace rest_rpc::rpc_service; namespace rest_rpc { + + /** + * The type to indicate the language of the client. + */ + enum class client_language_t { + CPP = 0, + JAVA = 1, + }; + class req_result { public: req_result() = default; @@ -48,15 +59,39 @@ namespace rest_rpc { class rpc_client : private asio::noncopyable { public: - rpc_client() : socket_(ios_), work_(ios_), + rpc_client() : socket_(ios_), work_(ios_), deadline_(ios_), body_(INIT_BUF_SIZE) { thd_ = std::make_shared([this] { ios_.run(); }); } - rpc_client(const std::string& host, unsigned short port) : socket_(ios_), work_(ios_), - deadline_(ios_), host_(host), port_(port), body_(INIT_BUF_SIZE) { + rpc_client(client_language_t client_language, + std::function on_result_received_callback) + : socket_(ios_), work_(ios_), + deadline_(ios_), body_(INIT_BUF_SIZE), + client_language_(client_language), + on_result_received_callback_(std::move(on_result_received_callback)) { + thd_ = std::make_shared([this] { + ios_.run(); + }); + } + + rpc_client(const std::string& host, unsigned short port) + : rpc_client(client_language_t::CPP, nullptr, host, port) {} + + rpc_client(client_language_t client_language, + std::function on_result_received_callback, + std::string host, + unsigned short port) + : socket_(ios_), + work_(ios_), + deadline_(ios_), + host_(std::move(host)), + port_(port), + body_(INIT_BUF_SIZE), + client_language_(client_language), + on_result_received_callback_(std::move(on_result_received_callback)) { thd_ = std::make_shared([this] { ios_.run(); }); @@ -251,7 +286,6 @@ namespace rest_rpc { 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_); @@ -539,42 +573,48 @@ namespace rest_rpc { } void call_back(uint64_t req_id, const boost::system::error_code& ec, string_view data) { - temp_req_id_ = req_id; - auto cb_flag = req_id >> 63; - if (cb_flag) { - std::shared_ptr cl = nullptr; - { - std::unique_lock lock(cb_mtx_); - cl = std::move(callback_map_[req_id]); - } + if (client_language_ == client_language_t::JAVA) { + // For Java client. + // TODO(qwang): Call java callback. + // handle error. + on_result_received_callback_(req_id, data.to_string()); + } else { + // For CPP client. + temp_req_id_ = req_id; + auto cb_flag = req_id >> 63; + if (cb_flag) { + std::shared_ptr cl = nullptr; + { + std::unique_lock lock(cb_mtx_); + cl = std::move(callback_map_[req_id]); + } - assert(cl); - if (!cl->has_timeout()) { - cl->cancel(); - cl->callback(ec, data); - } - else { - cl->callback(asio::error::make_error_code(asio::error::timed_out), {}); - } + assert(cl); + if (!cl->has_timeout()) { + cl->cancel(); + cl->callback(ec, data); + } else { + cl->callback(asio::error::make_error_code(asio::error::timed_out), {}); + } - std::unique_lock lock(cb_mtx_); - callback_map_.erase(req_id); - } - else { - std::unique_lock lock(cb_mtx_); - auto& f = future_map_[req_id]; - if (ec) { - //LOG< lock(cb_mtx_); + callback_map_.erase(req_id); + } else { + std::unique_lock lock(cb_mtx_); + auto &f = future_map_[req_id]; + if (ec) { + //LOG<set_value(req_result{ data }); - future_map_.erase(req_id); - } + assert(f); + f->set_value(req_result{data}); + future_map_.erase(req_id); + } + } } void callback_sub(const boost::system::error_code& ec, string_view result) { @@ -811,5 +851,8 @@ namespace rest_rpc { std::unordered_map> sub_map_; std::set> key_token_set_; + + client_language_t client_language_ = client_language_t::CPP; + std::function on_result_received_callback_; }; } diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java index a771616..4c3f314 100644 --- a/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java @@ -4,25 +4,25 @@ import java.util.concurrent.CompletableFuture; public interface AsyncRpcFunction { - CompletableFuture invoke(); +// CompletableFuture invoke(); +// +// +// CompletableFuture invoke(Arg1Type arg1); - - CompletableFuture invoke(Arg1Type arg1); + + RestFuture invoke(Arg1Type arg1, Arg2Type arg2); - - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2); - - - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3); - - - - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4); - - - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5); - - - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6); +// +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3); +// +// +// +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4); +// +// +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5); +// +// +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6); } diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java index 7360484..b5e1f17 100644 --- a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java @@ -13,47 +13,47 @@ public class AsyncRpcFunctionImpl implements AsyncRpcFunction { this.funcName = funcName; } - public CompletableFuture invoke() { - return internalInvoke(new Object[0]); - } +// public CompletableFuture invoke() { +// return internalInvoke(new Object[0]); +// } +// +// public CompletableFuture invoke(Arg1Type arg1) { +// Object[] args = new Object[] {arg1}; +// return internalInvoke(args); +// } - public CompletableFuture invoke(Arg1Type arg1) { - Object[] args = new Object[] {arg1}; - return internalInvoke(args); - } - - public - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2) { + public + RestFuture invoke(Arg1Type arg1, Arg2Type arg2) { Object[] args = new Object[] {arg1, arg2}; return internalInvoke(args); } - public - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) { - Object[] args = new Object[] {arg1, arg2, arg3}; - return internalInvoke(args); - } +// public +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) { +// Object[] args = new Object[] {arg1, arg2, arg3}; +// return internalInvoke(args); +// } +// +// +// public +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) { +// Object[] args = new Object[] {arg1, arg2, arg3, arg4}; +// return internalInvoke(args); +// } +// +// public +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5) { +// Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5}; +// return internalInvoke(args); +// } +// +// public +// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6) { +// Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5, arg6}; +// return internalInvoke(args); +// } - - public - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) { - Object[] args = new Object[] {arg1, arg2, arg3, arg4}; - return internalInvoke(args); - } - - public - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5) { - Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5}; - return internalInvoke(args); - } - - public - CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6) { - Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5, arg6}; - return internalInvoke(args); - } - - private CompletableFuture internalInvoke(Object[] args) { + private RestFuture internalInvoke(Object[] 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 index a0d090b..4213e94 100644 --- a/java/src/main/java/org/restrpc/client/Codec.java +++ b/java/src/main/java/org/restrpc/client/Codec.java @@ -2,9 +2,19 @@ package org.restrpc.client; import org.msgpack.core.MessageBufferPacker; import org.msgpack.core.MessagePack; +import org.msgpack.core.MessageUnpacker; + +import java.awt.print.PrinterGraphics; import java.io.IOException; public class Codec { + + private final static String INT_TYPE_NAME = "java.lang.Integer"; + + private final static String LONG_TYPE_NAME = "java.lang.Long"; + + private final static String STRING_TYPE_NAME = "java.lang.String"; + public byte[] encode(String funcName, Object[] args) throws IOException { // assert args != nullptr. @@ -20,19 +30,50 @@ public class Codec { final String argTypeName = arg.getClass().getName(); switch (argTypeName) { - case "java.lang.Integer": + case INT_TYPE_NAME: messagePacker.packInt((int) arg); break; - case "java.lang.Long": + case LONG_TYPE_NAME: messagePacker.packLong((long) arg); break; - case "java.lang.String": + case STRING_TYPE_NAME: messagePacker.packString((String) arg); break; default: - throw new RuntimeException("Unknown type" + argTypeName); + throw new RuntimeException("Unknown type: " + argTypeName); } } return messagePacker.toByteArray(); } + + public Object decodeSingleValue(String targetTypeName, byte[] encodedBytes) throws IOException { + if (targetTypeName == null) { + throw new RuntimeException("Internal bug."); + } + + if (encodedBytes == null) { + return null; + } + + // TODO(qwang): unpack nil. + MessageUnpacker messageUnpacker = MessagePack.newDefaultUnpacker(encodedBytes); + switch (targetTypeName) { + case INT_TYPE_NAME: + return messageUnpacker.unpackInt(); + case LONG_TYPE_NAME: + return messageUnpacker.unpackLong(); + case STRING_TYPE_NAME: + return messageUnpacker.unpackString(); + default: + throw new RuntimeException("Unknown type: " + targetTypeName); + } + + } + + public int myDecodeInt(byte[] encodedBytes) throws IOException { + MessageUnpacker messageUnpacker = MessagePack.newDefaultUnpacker(encodedBytes); + messageUnpacker.unpackArrayHeader(); + messageUnpacker.unpackInt(); + return messageUnpacker.unpackInt(); + } } diff --git a/java/src/main/java/org/restrpc/client/NativeRpcClient.java b/java/src/main/java/org/restrpc/client/NativeRpcClient.java index 7e2a230..69836cb 100644 --- a/java/src/main/java/org/restrpc/client/NativeRpcClient.java +++ b/java/src/main/java/org/restrpc/client/NativeRpcClient.java @@ -1,19 +1,24 @@ package org.restrpc.client; import java.io.IOException; -import java.util.HashMap; +import java.util.Objects; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; public class NativeRpcClient implements RpcClient { static { JniUtils.loadLibrary("restrpc_jni"); } + private long rpcClientPointer = -1; private Codec codec; - private HashMap> localFutureCache = new HashMap<>(); + // The map to cache return type. + private ConcurrentHashMap localFutureReturnTypenameCache = new ConcurrentHashMap<>(); + + private ConcurrentHashMap> localFutureCache = new ConcurrentHashMap<>(); public NativeRpcClient() { rpcClientPointer = nativeNewRpcClient(); @@ -34,7 +39,7 @@ public class NativeRpcClient implements RpcClient { return new AsyncRpcFunctionImpl(this, funcName); } - public CompletableFuture invoke(String funcName, Object[] args) { + public RestFuture invoke(String funcName, Object[] args) { if (rpcClientPointer == -1) { throw new RuntimeException("no init"); } @@ -51,7 +56,7 @@ public class NativeRpcClient implements RpcClient { } final long requestId = nativeInvoke(rpcClientPointer, encodedBytes); - CompletableFuture futureToReturn = new CompletableFuture<>(); + RestFuture futureToReturn = new RestFuture() {}; localFutureCache.put(requestId, futureToReturn); return futureToReturn; } @@ -65,6 +70,23 @@ public class NativeRpcClient implements RpcClient { this.rpcClientPointer = -1; } + /** + * The callback that will be invoked once the reuslt of rpc request received. + * Note that this method will be invoked in JNI. + */ + private void onResultReceived(long requestId, byte[] encodedReturnValueBytes) throws IOException { +// if (requestId not is local_cache) {//error} +// codec.decodeReturnValue(encodedReturnValueBytes); + System.out.println("-----------oo java-----------"); + System.out.println("result is " + codec.myDecodeInt(encodedReturnValueBytes)); + System.out.println("-----------end java-----------"); + + RestFuture future = localFutureCache.get(requestId); + future.getClass().getGenericSuperclass().getTypeName(); + Object o = codec.decodeSingleValue( + future.getClass().getGenericSuperclass().getTypeName(), encodedReturnValueBytes); + } + private native long nativeNewRpcClient(); private native void nativeConnect(long rpcClientPointer, String serverAddress); diff --git a/java/src/main/java/org/restrpc/client/RestFuture.java b/java/src/main/java/org/restrpc/client/RestFuture.java new file mode 100644 index 0000000..c8e0517 --- /dev/null +++ b/java/src/main/java/org/restrpc/client/RestFuture.java @@ -0,0 +1,38 @@ +package org.restrpc.client; + + +import java.io.IOException; + +public class RestFuture { + + private Class metaType; + + private T value; + + public RestFuture(Class metaType) { + this.metaType = metaType; + } + private byte[] encodedData; + + private static Codec codec = new Codec(); + + public RestFuture() { + } + + void complete(Integer value) { + this.value = (T) value; + } + + void complete(Long value) { + this.value = (T) value; + } + + // ... + + // TODO(qwang): Add uncheck!!! + public T get() throws IOException { +// // TODO(qwang): This can be cached. +// return (T) codec.decodeSingleValue(metaType.getTypeName(), encodedData); + return null; + } +} diff --git a/java/src/main/java/org/restrpc/client/RpcClient.java b/java/src/main/java/org/restrpc/client/RpcClient.java index 58b37b6..36903ad 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(String funcName, Object[] args); + RestFuture 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 823d618..38f9335 100644 --- a/java/src/test/java/org/restrpc/test/BasicClientTest.java +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -1,20 +1,28 @@ package org.restrpc.test; -import org.checkerframework.checker.units.qual.C; +import org.msgpack.core.MessageBufferPacker; import org.msgpack.core.MessagePack; +import org.msgpack.core.MessageUnpacker; import org.restrpc.client.Codec; import org.restrpc.client.NativeRpcClient; +import org.restrpc.client.RestFuture; import org.restrpc.client.RpcClient; import org.testng.annotations.Test; - +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.OutputStream; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; public class BasicClientTest { @Test - public void testBasic() { + public void testBasic() throws IOException, InterruptedException { RpcClient rpcClient = new NativeRpcClient(); - rpcClient.asyncFunc("111").invoke(2, 3); + rpcClient.connect("127.0.0.1:9000"); + RestFuture obj = rpcClient.asyncFunc("add").invoke(2, 3); + TimeUnit.SECONDS.sleep(10000); + System.out.println("The result of add(2, 3) is " + obj.get()); } private String getClassStr(Object o) { @@ -39,4 +47,21 @@ public class BasicClientTest { Object[] args = new Object[] {2, 3}; byte[] bs = codec.encode("add", args); } + + @Test + public void testDecode() throws IOException { + CompletableFuture i = new CompletableFuture() {}; +// System.out.println(i.getClass().getDeclare); +// +// OutputStream os = new ByteArrayOutputStream(); +// MessageBufferPacker messagePacker = MessagePack.newDefaultBufferPacker(); +// messagePacker.packArrayHeader(3); +// messagePacker.packString("add"); +// messagePacker.packInt(2); +// messagePacker.packInt(3); +// byte[] bs1 = messagePacker.toByteArray(); +// +// MessageUnpacker messageUnpacker = MessagePack.newDefaultUnpacker(bs1); +// messageUnpacker.unpackInt(); + } } diff --git a/jni/org_restrpc_client_NativeRpcClient.cc b/jni/org_restrpc_client_NativeRpcClient.cc index ec3d155..2548af4 100644 --- a/jni/org_restrpc_client_NativeRpcClient.cc +++ b/jni/org_restrpc_client_NativeRpcClient.cc @@ -2,13 +2,45 @@ #include "org_restrpc_client_NativeRpcClient.h" #include +#include #include -#ifdef __cplusplus -extern "C" { -#endif +JavaVM *jvm; +jclass java_class_NativeRpcClient; +jmethodID java_method_onResultReceived; +jobject java_object_native_rpc_client; + +inline jclass LoadClass(JNIEnv *env, const char *class_name) { + jclass tempLocalClassRef = env->FindClass(class_name); + jclass ret = (jclass)env->NewGlobalRef(tempLocalClassRef); +// assert(ret); + env->DeleteLocalRef(tempLocalClassRef); + return ret; +} + +/// Load and cache frequently-used Java classes and methods +jint JNI_OnLoad(JavaVM *vm, void *reserved) { + JNIEnv *env; + if (vm->GetEnv(reinterpret_cast(&env), 0x00010008) != JNI_OK) { + return JNI_ERR; + } + jvm = vm; + java_class_NativeRpcClient = LoadClass(env, "org/restrpc/client/NativeRpcClient"); + java_method_onResultReceived = env->GetMethodID(java_class_NativeRpcClient, "onResultReceived", "(J[B)V"); + return 0x00010008; +} + +//void JNI_OnUnload(JavaVM *vm, void *reserved) {} + +/// Convert C++ String to a Java ByteArray. +inline jbyteArray NativeStringToJavaByteArray(JNIEnv *env, const std::string &str) { + jbyteArray array = env->NewByteArray(str.size()); + env->SetByteArrayRegion(array, 0, str.size(), + reinterpret_cast(str.c_str())); + return array; +} inline std::string JavaByteArrayToNativeString(JNIEnv *env, const jbyteArray &bytes) { const auto size = env->GetArrayLength(bytes); @@ -21,17 +53,39 @@ inline std::string JavaByteArrayToNativeString(JNIEnv *env, const jbyteArray &by // JavaByteArrayToSBuffer +#ifdef __cplusplus +extern "C" { +#endif + /* * Class: org_restrpc_client_NativeRpcClient * Method: nativeNewRpcClient * Signature: ()J */ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeNewRpcClient - (JNIEnv *, jobject o) { - rest_rpc::rpc_client *native_rpc_client = new rest_rpc::rpc_client(); + (JNIEnv *env, jobject o) { + + java_object_native_rpc_client = (jobject) env->NewGlobalRef(o); + + auto on_result_received = [](long request_id, const std::string &data) { + JNIEnv *env = nullptr; + std::cout << "on_result_received Callback invoked." << std::endl; + jvm->AttachCurrentThreadAsDaemon(reinterpret_cast(&env), nullptr); + std::cout << "---------------env:" << env << std::endl; + jbyteArray javaByteArray = NativeStringToJavaByteArray(env, data); + + std::cout << "---------------a" << std::endl; + env->CallVoidMethod(java_object_native_rpc_client, java_method_onResultReceived, request_id, javaByteArray); + std::cout << "---------------e" << std::endl; + }; + + rest_rpc::rpc_client *native_rpc_client = new rest_rpc::rpc_client( + rest_rpc::client_language_t::JAVA, on_result_received); + std::cout << "------native_rpc_client=" << reinterpret_cast(native_rpc_client) << std::endl; return reinterpret_cast(native_rpc_client); } + /* * Class: org_restrpc_client_NativeRpcClient * Method: nativeConnect @@ -41,7 +95,8 @@ JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeConnect (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); + const bool connected = native_rpc_client->connect("127.0.0.1", 9000); + std::cout << "Connected:" << connected << std::endl; } /* diff --git a/jni/org_restrpc_client_NativeRpcClient.h b/jni/org_restrpc_client_NativeRpcClient.h index 850d61f..e4ab0b4 100644 --- a/jni/org_restrpc_client_NativeRpcClient.h +++ b/jni/org_restrpc_client_NativeRpcClient.h @@ -4,6 +4,9 @@ #ifndef _Included_org_restrpc_client_NativeRpcClient #define _Included_org_restrpc_client_NativeRpcClient + +extern JavaVM *jvm; + #ifdef __cplusplus extern "C" { #endif From f4c9b02908df4bb88060ec27b2922bea132ae9b2 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 21:38:06 +0800 Subject: [PATCH 05/12] FIRST INITIAL --- .../org/restrpc/client/AsyncRpcFunction.java | 2 +- .../restrpc/client/AsyncRpcFunctionImpl.java | 8 ++--- .../main/java/org/restrpc/client/Codec.java | 25 ++++++------- .../org/restrpc/client/NativeRpcClient.java | 36 ++++++++++--------- .../java/org/restrpc/client/RestFuture.java | 2 ++ .../java/org/restrpc/client/RpcClient.java | 2 +- .../org/restrpc/test/BasicClientTest.java | 10 +++--- 7 files changed, 45 insertions(+), 40 deletions(-) diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java index 4c3f314..45baeaf 100644 --- a/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java @@ -10,7 +10,7 @@ public interface AsyncRpcFunction { // CompletableFuture invoke(Arg1Type arg1); - RestFuture invoke(Arg1Type arg1, Arg2Type arg2); + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2); // // CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3); diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java index b5e1f17..e71924a 100644 --- a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java @@ -23,9 +23,9 @@ public class AsyncRpcFunctionImpl implements AsyncRpcFunction { // } public - RestFuture invoke(Arg1Type arg1, Arg2Type arg2) { + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2) { Object[] args = new Object[] {arg1, arg2}; - return internalInvoke(args); + return internalInvoke(returnClz, args); } // public @@ -53,7 +53,7 @@ public class AsyncRpcFunctionImpl implements AsyncRpcFunction { // return internalInvoke(args); // } - private RestFuture internalInvoke(Object[] args) { - return rpcClient.invoke(funcName, args); + private CompletableFuture internalInvoke(Class returnClz, Object[] args) { + return rpcClient.invoke(returnClz, funcName, args); } } diff --git a/java/src/main/java/org/restrpc/client/Codec.java b/java/src/main/java/org/restrpc/client/Codec.java index 4213e94..f19ead3 100644 --- a/java/src/main/java/org/restrpc/client/Codec.java +++ b/java/src/main/java/org/restrpc/client/Codec.java @@ -46,8 +46,8 @@ public class Codec { return messagePacker.toByteArray(); } - public Object decodeSingleValue(String targetTypeName, byte[] encodedBytes) throws IOException { - if (targetTypeName == null) { + public Object decodeReturnValue(Class returnClz, byte[] encodedBytes) throws IOException { + if (returnClz == null) { throw new RuntimeException("Internal bug."); } @@ -57,17 +57,18 @@ public class Codec { // TODO(qwang): unpack nil. MessageUnpacker messageUnpacker = MessagePack.newDefaultUnpacker(encodedBytes); - switch (targetTypeName) { - case INT_TYPE_NAME: - return messageUnpacker.unpackInt(); - case LONG_TYPE_NAME: - return messageUnpacker.unpackLong(); - case STRING_TYPE_NAME: - return messageUnpacker.unpackString(); - default: - throw new RuntimeException("Unknown type: " + targetTypeName); - } + // Unpack unnecessary fields. + messageUnpacker.unpackArrayHeader(); + messageUnpacker.unpackInt(); + if (Integer.class.equals(returnClz)) { + return messageUnpacker.unpackInt(); + } else if (Long.class.equals(returnClz)) { + return messageUnpacker.unpackLong(); + } else if (String.class.equals(returnClz)) { + return messageUnpacker.unpackString(); + } + throw new RuntimeException("Unknown type: " + returnClz); } public int myDecodeInt(byte[] encodedBytes) throws IOException { diff --git a/java/src/main/java/org/restrpc/client/NativeRpcClient.java b/java/src/main/java/org/restrpc/client/NativeRpcClient.java index 69836cb..bda6bab 100644 --- a/java/src/main/java/org/restrpc/client/NativeRpcClient.java +++ b/java/src/main/java/org/restrpc/client/NativeRpcClient.java @@ -1,7 +1,7 @@ package org.restrpc.client; + import java.io.IOException; -import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -16,9 +16,9 @@ public class NativeRpcClient implements RpcClient { private Codec codec; // The map to cache return type. - private ConcurrentHashMap localFutureReturnTypenameCache = new ConcurrentHashMap<>(); + private ConcurrentHashMap> localFutureReturnTypenameCache = new ConcurrentHashMap<>(); - private ConcurrentHashMap> localFutureCache = new ConcurrentHashMap<>(); + private ConcurrentHashMap> localFutureCache = new ConcurrentHashMap<>(); public NativeRpcClient() { rpcClientPointer = nativeNewRpcClient(); @@ -39,7 +39,7 @@ public class NativeRpcClient implements RpcClient { return new AsyncRpcFunctionImpl(this, funcName); } - public RestFuture invoke(String funcName, Object[] args) { + public CompletableFuture invoke(Class returnClz, String funcName, Object[] args) { if (rpcClientPointer == -1) { throw new RuntimeException("no init"); } @@ -55,10 +55,14 @@ public class NativeRpcClient implements RpcClient { return null; } - final long requestId = nativeInvoke(rpcClientPointer, encodedBytes); - RestFuture futureToReturn = new RestFuture() {}; - localFutureCache.put(requestId, futureToReturn); - return futureToReturn; + synchronized (this) { + final long requestId = nativeInvoke(rpcClientPointer, encodedBytes); + CompletableFuture futureToReturn = new CompletableFuture(); + localFutureReturnTypenameCache.put(requestId, returnClz); + + localFutureCache.put(requestId, futureToReturn); + return futureToReturn; + } } public void close() { @@ -76,15 +80,13 @@ public class NativeRpcClient implements RpcClient { */ private void onResultReceived(long requestId, byte[] encodedReturnValueBytes) throws IOException { // if (requestId not is local_cache) {//error} -// codec.decodeReturnValue(encodedReturnValueBytes); - System.out.println("-----------oo java-----------"); - System.out.println("result is " + codec.myDecodeInt(encodedReturnValueBytes)); - System.out.println("-----------end java-----------"); - - RestFuture future = localFutureCache.get(requestId); - future.getClass().getGenericSuperclass().getTypeName(); - Object o = codec.decodeSingleValue( - future.getClass().getGenericSuperclass().getTypeName(), encodedReturnValueBytes); +// System.out.println("result is " + codec.myDecodeInt(encodedReturnValueBytes)); + synchronized (this) { + final Class returnClz = localFutureReturnTypenameCache.get(requestId); + CompletableFuture future = localFutureCache.get(requestId); + Object o = codec.decodeReturnValue(returnClz, encodedReturnValueBytes); + future.complete(o); + } } private native long nativeNewRpcClient(); diff --git a/java/src/main/java/org/restrpc/client/RestFuture.java b/java/src/main/java/org/restrpc/client/RestFuture.java index c8e0517..f0ecbb6 100644 --- a/java/src/main/java/org/restrpc/client/RestFuture.java +++ b/java/src/main/java/org/restrpc/client/RestFuture.java @@ -2,6 +2,7 @@ package org.restrpc.client; import java.io.IOException; +import java.util.concurrent.CompletableFuture; public class RestFuture { @@ -12,6 +13,7 @@ public class RestFuture { public RestFuture(Class metaType) { this.metaType = metaType; } + private byte[] encodedData; private static Codec codec = new Codec(); diff --git a/java/src/main/java/org/restrpc/client/RpcClient.java b/java/src/main/java/org/restrpc/client/RpcClient.java index 36903ad..d8235b8 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); - RestFuture invoke(String funcName, Object[] args); + CompletableFuture invoke(Class returnClz, 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 38f9335..60f42ff 100644 --- a/java/src/test/java/org/restrpc/test/BasicClientTest.java +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -12,17 +12,17 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; public class BasicClientTest { @Test - public void testBasic() throws IOException, InterruptedException { + public void testBasic() throws IOException, InterruptedException, ExecutionException { RpcClient rpcClient = new NativeRpcClient(); rpcClient.connect("127.0.0.1:9000"); - RestFuture obj = rpcClient.asyncFunc("add").invoke(2, 3); - TimeUnit.SECONDS.sleep(10000); - System.out.println("The result of add(2, 3) is " + obj.get()); + CompletableFuture future = rpcClient.asyncFunc("echo").invoke(String.class, 2, 3); + System.out.println("The result of add(2, 3) is " + future.get()); } private String getClassStr(Object o) { @@ -50,7 +50,7 @@ public class BasicClientTest { @Test public void testDecode() throws IOException { - CompletableFuture i = new CompletableFuture() {}; +// CompletableFuture i = new CompletableFuture() {}; // System.out.println(i.getClass().getDeclare); // // OutputStream os = new ByteArrayOutputStream(); From 90e834c960f100ac66efd7af40ae028436c625ec Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 22:11:49 +0800 Subject: [PATCH 06/12] Fix --- include/rest_rpc/rpc_client.hpp | 32 +++++---- .../org/restrpc/client/AsyncRpcFunction.java | 34 +++++----- .../restrpc/client/AsyncRpcFunctionImpl.java | 68 +++++++++---------- .../org/restrpc/client/NativeRpcClient.java | 7 +- .../java/org/restrpc/client/RestFuture.java | 40 ----------- .../java/org/restrpc/client/RpcClient.java | 2 +- .../org/restrpc/test/BasicClientTest.java | 26 +------ jni/org_restrpc_client_NativeRpcClient.cc | 7 +- 8 files changed, 74 insertions(+), 142 deletions(-) delete mode 100644 java/src/main/java/org/restrpc/client/RestFuture.java diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp index 54928cc..76d94d2 100644 --- a/include/rest_rpc/rpc_client.hpp +++ b/include/rest_rpc/rpc_client.hpp @@ -283,23 +283,25 @@ 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)); + /** + * This internal_async_call is used for other language client. + * We use callback to handle the result is received, so we should not + * add the future to the future map. + */ + long internal_async_call(const std::string& encoded_func_name_and_args) { + auto p = std::make_shared>(); + uint64_t fu_id = 0; + { + std::unique_lock lock(cb_mtx_); + fu_id_++; + fu_id = fu_id_; + } + 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; } - 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/src/main/java/org/restrpc/client/AsyncRpcFunction.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java index 45baeaf..b6373fe 100644 --- a/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunction.java @@ -4,25 +4,25 @@ import java.util.concurrent.CompletableFuture; public interface AsyncRpcFunction { -// CompletableFuture invoke(); -// -// -// CompletableFuture invoke(Arg1Type arg1); + CompletableFuture invoke(Class returnClz); - + + CompletableFuture invoke(Class returnClz, Arg1Type arg1); + + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2); -// -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3); -// -// -// -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4); -// -// -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5); -// -// -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6); + + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3); + + + + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4); + + + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5); + + + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6); } diff --git a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java index e71924a..07b55d6 100644 --- a/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java +++ b/java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java @@ -13,47 +13,47 @@ public class AsyncRpcFunctionImpl implements AsyncRpcFunction { this.funcName = funcName; } -// public CompletableFuture invoke() { -// return internalInvoke(new Object[0]); -// } -// -// public CompletableFuture invoke(Arg1Type arg1) { -// Object[] args = new Object[] {arg1}; -// return internalInvoke(args); -// } + public CompletableFuture invoke(Class returnClz) { + return internalInvoke(returnClz, new Object[0]); + } - public + public CompletableFuture invoke(Class returnClz, Arg1Type arg1) { + Object[] args = new Object[] {arg1}; + return internalInvoke(returnClz, args); + } + + public CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2) { Object[] args = new Object[] {arg1, arg2}; return internalInvoke(returnClz, args); } -// public -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) { -// Object[] args = new Object[] {arg1, arg2, arg3}; -// return internalInvoke(args); -// } -// -// -// public -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) { -// Object[] args = new Object[] {arg1, arg2, arg3, arg4}; -// return internalInvoke(args); -// } -// -// public -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5) { -// Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5}; -// return internalInvoke(args); -// } -// -// public -// CompletableFuture invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6) { -// Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5, arg6}; -// return internalInvoke(args); -// } + public + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) { + Object[] args = new Object[] {arg1, arg2, arg3}; + return internalInvoke(returnClz, args); + } - private CompletableFuture internalInvoke(Class returnClz, Object[] args) { + + public + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) { + Object[] args = new Object[] {arg1, arg2, arg3, arg4}; + return internalInvoke(returnClz, args); + } + + public + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5) { + Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5}; + return internalInvoke(returnClz, args); + } + + public + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6) { + Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5, arg6}; + return internalInvoke(returnClz, args); + } + + private CompletableFuture internalInvoke(Class returnClz, Object[] args) { return rpcClient.invoke(returnClz, funcName, args); } } diff --git a/java/src/main/java/org/restrpc/client/NativeRpcClient.java b/java/src/main/java/org/restrpc/client/NativeRpcClient.java index bda6bab..488507d 100644 --- a/java/src/main/java/org/restrpc/client/NativeRpcClient.java +++ b/java/src/main/java/org/restrpc/client/NativeRpcClient.java @@ -39,7 +39,7 @@ public class NativeRpcClient implements RpcClient { return new AsyncRpcFunctionImpl(this, funcName); } - public CompletableFuture invoke(Class returnClz, String funcName, Object[] args) { + public CompletableFuture invoke(Class returnClz, String funcName, Object[] args) { if (rpcClientPointer == -1) { throw new RuntimeException("no init"); } @@ -78,13 +78,12 @@ public class NativeRpcClient implements RpcClient { * The callback that will be invoked once the reuslt of rpc request received. * Note that this method will be invoked in JNI. */ - private void onResultReceived(long requestId, byte[] encodedReturnValueBytes) throws IOException { + private void onResultReceived(long requestId, byte[] encodedReturnValue) throws IOException { // if (requestId not is local_cache) {//error} -// System.out.println("result is " + codec.myDecodeInt(encodedReturnValueBytes)); synchronized (this) { final Class returnClz = localFutureReturnTypenameCache.get(requestId); CompletableFuture future = localFutureCache.get(requestId); - Object o = codec.decodeReturnValue(returnClz, encodedReturnValueBytes); + Object o = codec.decodeReturnValue(returnClz, encodedReturnValue); future.complete(o); } } diff --git a/java/src/main/java/org/restrpc/client/RestFuture.java b/java/src/main/java/org/restrpc/client/RestFuture.java deleted file mode 100644 index f0ecbb6..0000000 --- a/java/src/main/java/org/restrpc/client/RestFuture.java +++ /dev/null @@ -1,40 +0,0 @@ -package org.restrpc.client; - - -import java.io.IOException; -import java.util.concurrent.CompletableFuture; - -public class RestFuture { - - private Class metaType; - - private T value; - - public RestFuture(Class metaType) { - this.metaType = metaType; - } - - private byte[] encodedData; - - private static Codec codec = new Codec(); - - public RestFuture() { - } - - void complete(Integer value) { - this.value = (T) value; - } - - void complete(Long value) { - this.value = (T) value; - } - - // ... - - // TODO(qwang): Add uncheck!!! - public T get() throws IOException { -// // TODO(qwang): This can be cached. -// return (T) codec.decodeSingleValue(metaType.getTypeName(), encodedData); - return null; - } -} diff --git a/java/src/main/java/org/restrpc/client/RpcClient.java b/java/src/main/java/org/restrpc/client/RpcClient.java index d8235b8..3239416 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(Class returnClz, String funcName, Object[] args); + CompletableFuture invoke(Class returnClz, 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 60f42ff..0f887b8 100644 --- a/java/src/test/java/org/restrpc/test/BasicClientTest.java +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -1,19 +1,12 @@ package org.restrpc.test; -import org.msgpack.core.MessageBufferPacker; -import org.msgpack.core.MessagePack; -import org.msgpack.core.MessageUnpacker; import org.restrpc.client.Codec; import org.restrpc.client.NativeRpcClient; -import org.restrpc.client.RestFuture; import org.restrpc.client.RpcClient; import org.testng.annotations.Test; -import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.io.OutputStream; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; public class BasicClientTest { @@ -21,7 +14,7 @@ public class BasicClientTest { public void testBasic() throws IOException, InterruptedException, ExecutionException { RpcClient rpcClient = new NativeRpcClient(); rpcClient.connect("127.0.0.1:9000"); - CompletableFuture future = rpcClient.asyncFunc("echo").invoke(String.class, 2, 3); + CompletableFuture future = rpcClient.asyncFunc("add").invoke(Integer.class, 2, 3); System.out.println("The result of add(2, 3) is " + future.get()); } @@ -47,21 +40,4 @@ public class BasicClientTest { Object[] args = new Object[] {2, 3}; byte[] bs = codec.encode("add", args); } - - @Test - public void testDecode() throws IOException { -// CompletableFuture i = new CompletableFuture() {}; -// System.out.println(i.getClass().getDeclare); -// -// OutputStream os = new ByteArrayOutputStream(); -// MessageBufferPacker messagePacker = MessagePack.newDefaultBufferPacker(); -// messagePacker.packArrayHeader(3); -// messagePacker.packString("add"); -// messagePacker.packInt(2); -// messagePacker.packInt(3); -// byte[] bs1 = messagePacker.toByteArray(); -// -// MessageUnpacker messageUnpacker = MessagePack.newDefaultUnpacker(bs1); -// messageUnpacker.unpackInt(); - } } diff --git a/jni/org_restrpc_client_NativeRpcClient.cc b/jni/org_restrpc_client_NativeRpcClient.cc index 2548af4..25f2fc1 100644 --- a/jni/org_restrpc_client_NativeRpcClient.cc +++ b/jni/org_restrpc_client_NativeRpcClient.cc @@ -69,14 +69,9 @@ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeNewRpcClie auto on_result_received = [](long request_id, const std::string &data) { JNIEnv *env = nullptr; - std::cout << "on_result_received Callback invoked." << std::endl; jvm->AttachCurrentThreadAsDaemon(reinterpret_cast(&env), nullptr); - std::cout << "---------------env:" << env << std::endl; jbyteArray javaByteArray = NativeStringToJavaByteArray(env, data); - - std::cout << "---------------a" << std::endl; env->CallVoidMethod(java_object_native_rpc_client, java_method_onResultReceived, request_id, javaByteArray); - std::cout << "---------------e" << std::endl; }; rest_rpc::rpc_client *native_rpc_client = new rest_rpc::rpc_client( @@ -108,7 +103,7 @@ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeInvoke (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); + return native_rpc_client->internal_async_call(encodedFuncNameAndArgs); } /* From 789705ea99e0dc38ec0b4840bc2f6cecfcb1cb8e Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 22:34:03 +0800 Subject: [PATCH 07/12] Fix --- .../org/restrpc/test/BasicClientTest.java | 5 ++++ jni/org_restrpc_client_NativeRpcClient.cc | 26 ++++++++++++++----- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/java/src/test/java/org/restrpc/test/BasicClientTest.java b/java/src/test/java/org/restrpc/test/BasicClientTest.java index 0f887b8..8ccd5d2 100644 --- a/java/src/test/java/org/restrpc/test/BasicClientTest.java +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -16,6 +16,11 @@ public class BasicClientTest { rpcClient.connect("127.0.0.1:9000"); CompletableFuture future = rpcClient.asyncFunc("add").invoke(Integer.class, 2, 3); System.out.println("The result of add(2, 3) is " + future.get()); + + { + CompletableFuture future1 = rpcClient.asyncFunc("echo").invoke(String.class, "hello world!"); + System.out.println("The result of echo is " + future1.get()); + } } private String getClassStr(Object o) { diff --git a/jni/org_restrpc_client_NativeRpcClient.cc b/jni/org_restrpc_client_NativeRpcClient.cc index 25f2fc1..0e4af5b 100644 --- a/jni/org_restrpc_client_NativeRpcClient.cc +++ b/jni/org_restrpc_client_NativeRpcClient.cc @@ -34,6 +34,13 @@ jint JNI_OnLoad(JavaVM *vm, void *reserved) { //void JNI_OnUnload(JavaVM *vm, void *reserved) {} +inline std::string JavaStringToNativeString(JNIEnv *env, jstring jstr) { + const char *c_str = env->GetStringUTFChars(jstr, nullptr); + std::string result(c_str); + env->ReleaseStringUTFChars(static_cast(jstr), c_str); + return result; +} + /// Convert C++ String to a Java ByteArray. inline jbyteArray NativeStringToJavaByteArray(JNIEnv *env, const std::string &str) { jbyteArray array = env->NewByteArray(str.size()); @@ -76,7 +83,6 @@ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeNewRpcClie rest_rpc::rpc_client *native_rpc_client = new rest_rpc::rpc_client( rest_rpc::client_language_t::JAVA, on_result_received); - std::cout << "------native_rpc_client=" << reinterpret_cast(native_rpc_client) << std::endl; return reinterpret_cast(native_rpc_client); } @@ -87,11 +93,15 @@ 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 o, jlong rpcClientPointer, jstring serverAddress) { +(JNIEnv *env, jobject o, jlong rpcClientPointer, jstring serverAddress) { auto *native_rpc_client = reinterpret_cast(rpcClientPointer); - // TODO(qwang): Do not hard code this. - const bool connected = native_rpc_client->connect("127.0.0.1", 9000); - std::cout << "Connected:" << connected << std::endl; + // TODO(qwang): return a flag or throw exception. + const std::string server_addr = JavaStringToNativeString(env, serverAddress); + // Use a helper to split and handle the exception. + const size_t pos = server_addr.find(":"); + const std::string ip = server_addr.substr(0, pos); + const int port = std::stoi(server_addr.substr(pos + 1, server_addr.size())); + const bool connected = native_rpc_client->connect(ip, static_cast(port)); } /* @@ -112,8 +122,10 @@ JNIEXPORT jlong JNICALL Java_org_restrpc_client_NativeRpcClient_nativeInvoke * Signature: (J)V */ JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeDestroy -(JNIEnv *, jobject, jlong) { - +(JNIEnv *, jobject, jlong rpcClientPointer) { + auto *native_rpc_client = reinterpret_cast(rpcClientPointer); + native_rpc_client->close(); + delete native_rpc_client; } #ifdef __cplusplus From 38b9e78fda864e14d2f3c1ebc641b0deed1c8c15 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 22:35:02 +0800 Subject: [PATCH 08/12] Fix --- jni/org_restrpc_client_NativeRpcClient.cc | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/jni/org_restrpc_client_NativeRpcClient.cc b/jni/org_restrpc_client_NativeRpcClient.cc index 0e4af5b..3541918 100644 --- a/jni/org_restrpc_client_NativeRpcClient.cc +++ b/jni/org_restrpc_client_NativeRpcClient.cc @@ -126,6 +126,10 @@ JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeDestroy auto *native_rpc_client = reinterpret_cast(rpcClientPointer); native_rpc_client->close(); delete native_rpc_client; + + env->DeleteGlobalRef(java_class_NativeRpcClient); + env->DeleteGlobalRef(java_method_onResultReceived); + env->DeleteGlobalRef(java_object_native_rpc_client); } #ifdef __cplusplus From 44efb726d2f27efd45db4665150c6266e75c8e44 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 22:41:56 +0800 Subject: [PATCH 09/12] Fix --- .../main/java/org/restrpc/client/Codec.java | 8 ----- .../org/restrpc/test/BasicClientTest.java | 36 ++++--------------- 2 files changed, 6 insertions(+), 38 deletions(-) diff --git a/java/src/main/java/org/restrpc/client/Codec.java b/java/src/main/java/org/restrpc/client/Codec.java index f19ead3..1935441 100644 --- a/java/src/main/java/org/restrpc/client/Codec.java +++ b/java/src/main/java/org/restrpc/client/Codec.java @@ -17,7 +17,6 @@ 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); @@ -70,11 +69,4 @@ public class Codec { } throw new RuntimeException("Unknown type: " + returnClz); } - - public int myDecodeInt(byte[] encodedBytes) throws IOException { - MessageUnpacker messageUnpacker = MessagePack.newDefaultUnpacker(encodedBytes); - messageUnpacker.unpackArrayHeader(); - messageUnpacker.unpackInt(); - return messageUnpacker.unpackInt(); - } } diff --git a/java/src/test/java/org/restrpc/test/BasicClientTest.java b/java/src/test/java/org/restrpc/test/BasicClientTest.java index 8ccd5d2..03445f7 100644 --- a/java/src/test/java/org/restrpc/test/BasicClientTest.java +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -1,48 +1,24 @@ package org.restrpc.test; -import org.restrpc.client.Codec; import org.restrpc.client.NativeRpcClient; import org.restrpc.client.RpcClient; +import org.testng.Assert; import org.testng.annotations.Test; -import java.io.IOException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; public class BasicClientTest { @Test - public void testBasic() throws IOException, InterruptedException, ExecutionException { + public void testBasic() throws InterruptedException, ExecutionException { + // Note that you must run the `basic_server` first and then run this test. RpcClient rpcClient = new NativeRpcClient(); rpcClient.connect("127.0.0.1:9000"); CompletableFuture future = rpcClient.asyncFunc("add").invoke(Integer.class, 2, 3); - System.out.println("The result of add(2, 3) is " + future.get()); + Assert.assertEquals(future.get(), 5); - { - CompletableFuture future1 = rpcClient.asyncFunc("echo").invoke(String.class, "hello world!"); - System.out.println("The result of echo is " + future1.get()); - } + CompletableFuture future1 = rpcClient.asyncFunc("echo").invoke(String.class, "hello world"); + Assert.assertEquals(future1.get(), "hello world"); } - 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); - } } From 15394b09bf70e4f7cce0a6497f2444f34b6e744f Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 22:49:00 +0800 Subject: [PATCH 10/12] Fix --- examples/CMakeLists.txt | 6 ++++-- examples/client/main.cpp | 35 ++++++++++++++++----------------- examples/server/main.cpp | 3 +-- examples/server/qps.h | 2 +- include/rest_rpc/rpc_client.hpp | 4 ++-- jni/CMakeLists.txt | 12 ++++------- 6 files changed, 29 insertions(+), 33 deletions(-) diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index a5bf92e..4e955cf 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -21,7 +21,9 @@ add_executable(basic_server server/main.cpp) add_executable(basic_client client/main.cpp) if (ENABLE_SSL) - target_link_libraries(basic_server ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) + target_link_libraries(basic_client ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) + target_link_libraries(basic_client ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) else() - target_link_libraries(basic_server ${Boost_LIBRARIES}) + target_link_libraries(basic_client ${Boost_LIBRARIES}) + target_link_libraries(basic_client ${Boost_LIBRARIES}) endif() diff --git a/examples/client/main.cpp b/examples/client/main.cpp index 9c40940..05cd558 100644 --- a/examples/client/main.cpp +++ b/examples/client/main.cpp @@ -16,8 +16,8 @@ void test_add() { } { -// auto result = client.call("add", 1, 2); -// std::cout << result << std::endl; + auto result = client.call("add", 1, 2); + std::cout << result << std::endl; } { @@ -621,21 +621,20 @@ void benchmark_test(){ } int main() { -// benchmark_test(); - test_add(); -// test_connect(); -// test_callback(); -// test_echo(); -// test_sync_client(); -// test_async_client(); - //test_threads(); - //test_sub(); - //test_call_with_timeout(); - //test_connect(); - //test_upload(); - //test_download(); - //multi_client_performance(20); - //test_performance1(); - //test_multiple_thread(); + benchmark_test(); + test_connect(); + test_callback(); + test_echo(); + test_sync_client(); + test_async_client(); + test_threads(); + test_sub1(); + test_call_with_timeout(); + test_connect(); + test_upload(); + test_download(); + multi_client_performance(20); + test_performance1(); + test_multiple_thread(); return 0; } \ No newline at end of file diff --git a/examples/server/main.cpp b/examples/server/main.cpp index 05cfb58..249baf0 100644 --- a/examples/server/main.cpp +++ b/examples/server/main.cpp @@ -7,8 +7,7 @@ using namespace rpc_service; struct dummy{ int add(rpc_conn conn, int a, int b) { - std::cout << "ohhhhhhhhhhh invoked: a=" << a << ", b=" << b << std::endl; - return a + b; + return a + b; } }; diff --git a/examples/server/qps.h b/examples/server/qps.h index 2afd304..408a520 100644 --- a/examples/server/qps.h +++ b/examples/server/qps.h @@ -12,7 +12,7 @@ public: qps() : counter_(0) { thd_ = std::thread([this] { while (!stop_) { -// std::cout << "qps: " << counter_.load(std::memory_order_acquire) << '\n'; + std::cout << "qps: " << counter_.load(std::memory_order_acquire) << '\n'; std::this_thread::sleep_for(std::chrono::seconds(1)); //counter_.store(0, std::memory_order_release); } diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp index 76d94d2..7091d3c 100644 --- a/include/rest_rpc/rpc_client.hpp +++ b/include/rest_rpc/rpc_client.hpp @@ -575,7 +575,7 @@ namespace rest_rpc { } void call_back(uint64_t req_id, const boost::system::error_code& ec, string_view data) { - if (client_language_ == client_language_t::JAVA) { + if (client_language_ == client_language_t::JAVA) { // For Java client. // TODO(qwang): Call java callback. // handle error. @@ -854,7 +854,7 @@ namespace rest_rpc { std::unordered_map> sub_map_; std::set> key_token_set_; - client_language_t client_language_ = client_language_t::CPP; + client_language_t client_language_ = client_language_t::CPP; std::function on_result_received_callback_; }; } diff --git a/jni/CMakeLists.txt b/jni/CMakeLists.txt index 77e1608..dd2515a 100644 --- a/jni/CMakeLists.txt +++ b/jni/CMakeLists.txt @@ -12,16 +12,12 @@ if (ENABLE_SSL) endif() find_package(JNI REQUIRED) -#find_package(Boost COMPONENTS system filesystem REQUIRED) +find_package(Boost COMPONENTS system filesystem REQUIRED) include_directories( - # "/usr/local/include" + "/usr/local/include" "../include" + "../../third/msgpack/include" "../jni" - ${JNI_INCLUDE_DIRS} -# "/usr/local/opt/openjdk/include/" - "/Users/qingwang/workspace/opensource/rest_rpc/third/msgpack/include") - -INCLUDE_DIRECTORIES(SYSTEM "/Users/qingwang/workspace/opensource/dousi/core/build/external/boost/src/boost_ep") -INCLUDE_DIRECTORIES(SYSTEM "/Users/qingwang/workspace/opensource/rest_rpc/third/msgpack/include") + ${JNI_INCLUDE_DIRS}) add_library(restrpc_jni SHARED org_restrpc_client_NativeRpcClient.cc) From c29a798e5bdd8e379df1144da7eeb9b32a3a5722 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Thu, 22 Oct 2020 09:56:39 +0800 Subject: [PATCH 11/12] Fix --- examples/CMakeLists.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 4e955cf..6f64c49 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -21,9 +21,9 @@ add_executable(basic_server server/main.cpp) add_executable(basic_client client/main.cpp) if (ENABLE_SSL) - target_link_libraries(basic_client ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) + target_link_libraries(basic_server ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) target_link_libraries(basic_client ${Boost_LIBRARIES} -lssl -lcrypto -lpthread) else() - target_link_libraries(basic_client ${Boost_LIBRARIES}) + target_link_libraries(basic_server ${Boost_LIBRARIES}) target_link_libraries(basic_client ${Boost_LIBRARIES}) endif() From 611e8757666162a17adfe9d24100fcb7e9096d33 Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Thu, 22 Oct 2020 12:21:21 +0800 Subject: [PATCH 12/12] Add example --- .../restrpc/examples/AsyncInvokeExample.java | 55 +++++++++++++++++++ 1 file changed, 55 insertions(+) create mode 100644 java/src/main/java/org/restrpc/examples/AsyncInvokeExample.java diff --git a/java/src/main/java/org/restrpc/examples/AsyncInvokeExample.java b/java/src/main/java/org/restrpc/examples/AsyncInvokeExample.java new file mode 100644 index 0000000..8634906 --- /dev/null +++ b/java/src/main/java/org/restrpc/examples/AsyncInvokeExample.java @@ -0,0 +1,55 @@ +package org.restrpc.examples; + +import org.restrpc.client.AsyncRpcFunction; +import org.restrpc.client.NativeRpcClient; +import org.restrpc.client.RpcClient; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +public class AsyncInvokeExample { + + public static void main(String[] args) throws InterruptedException { + /** + * An example shows how we use this Java client to connect to + * the C++ RPC server and invoke the C++ RPC methods. + * + * First of all, we should run a C++ rpc server. In this example, + * we first run the `basic_server` which is written here: + * https://github.com/qicosmos/rest_rpc/blob/master/examples/server/main.cpp + * + * And make sure the C++ RPC server is listening on the address "127.0.0.1:9000". + * + * From the above C++ file, we can know that we registered the method `add()`: + * int add(rpc_conn conn, int a, int b) { + * return a + b; + * } + * + * So we can create a Java RPC client to invoke that remote method by the following + * steps below. + */ + + // Create a RPC client instance. + RpcClient rpcClient = new NativeRpcClient(); + // Connect to the C++ RPC server which is listening on `127.0.0.1:9000`. + rpcClient.connect("127.0.0.1:9000"); + // Use `asyncFunc()` method as a proxy of the remote function. This can + // be known as a stub of the remote method. + AsyncRpcFunction remoteFunc = rpcClient.asyncFunc("add"); + // The first argument indicates the return type you expected to return. It + // followed by the other arguments that the remote method takes to perform the + // `add()` method. + CompletableFuture future = remoteFunc.invoke(Integer.class, 100, 230); + // The last, we can use the future as async return value. + future.whenComplete((obj, exception) -> { + if (exception != null) { + System.out.println("Failed to invoke the add(100, 230) with: " + exception.getMessage()); + } else { + System.out.println("The result of add(100, 230) is " + obj); + } + }); + // You also get the return value synchronously: + // Object result = future.get(); + TimeUnit.SECONDS.sleep(10); + } +}