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/examples/CMakeLists.txt b/examples/CMakeLists.txt index a5bf92e..6f64c49 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -22,6 +22,8 @@ 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) else() target_link_libraries(basic_server ${Boost_LIBRARIES}) + target_link_libraries(basic_client ${Boost_LIBRARIES}) endif() diff --git a/examples/client/main.cpp b/examples/client/main.cpp index 6511ae4..05cd558 100644 --- a/examples/client/main.cpp +++ b/examples/client/main.cpp @@ -621,21 +621,20 @@ void benchmark_test(){ } int main() { -// benchmark_test(); - test_sub1(); + benchmark_test(); 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(); + 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 4f7c7f7..249baf0 100644 --- a/examples/server/main.cpp +++ b/examples/server/main.cpp @@ -6,8 +6,8 @@ using namespace rpc_service; #include "qps.h" struct dummy{ - int add(rpc_conn conn, int a, int b) { - return a + b; + int add(rpc_conn conn, int a, int b) { + return a + b; } }; diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp index 576200e..7091d3c 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(); }); @@ -248,6 +283,25 @@ namespace rest_rpc { return future; } + /** + * 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; + } + template void async_call(const std::string& rpc_name, std::function cb, Args&& ... args) { if (!has_connected_) { @@ -521,42 +575,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) { @@ -793,5 +853,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/pom.xml b/java/pom.xml new file mode 100644 index 0000000..7a31d99 --- /dev/null +++ b/java/pom.xml @@ -0,0 +1,45 @@ + + + 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 + + + org.msgpack + msgpack-core + 0.8.21 + + + + + + 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..b6373fe --- /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(Class returnClz); + + + CompletableFuture invoke(Class returnClz, Arg1Type arg1); + + + CompletableFuture invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2); + + + 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 new file mode 100644 index 0000000..07b55d6 --- /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(Class returnClz) { + return internalInvoke(returnClz, new Object[0]); + } + + 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(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) { + Object[] args = new Object[] {arg1, arg2, arg3}; + return internalInvoke(returnClz, 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/Codec.java b/java/src/main/java/org/restrpc/client/Codec.java new file mode 100644 index 0000000..1935441 --- /dev/null +++ b/java/src/main/java/org/restrpc/client/Codec.java @@ -0,0 +1,72 @@ +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. + 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 INT_TYPE_NAME: + messagePacker.packInt((int) arg); + break; + case LONG_TYPE_NAME: + messagePacker.packLong((long) arg); + break; + case STRING_TYPE_NAME: + messagePacker.packString((String) arg); + break; + default: + throw new RuntimeException("Unknown type: " + argTypeName); + } + } + return messagePacker.toByteArray(); + } + + public Object decodeReturnValue(Class returnClz, byte[] encodedBytes) throws IOException { + if (returnClz == null) { + throw new RuntimeException("Internal bug."); + } + + if (encodedBytes == null) { + return null; + } + + // TODO(qwang): unpack nil. + MessageUnpacker messageUnpacker = MessagePack.newDefaultUnpacker(encodedBytes); + // 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); + } +} 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..488507d --- /dev/null +++ b/java/src/main/java/org/restrpc/client/NativeRpcClient.java @@ -0,0 +1,98 @@ +package org.restrpc.client; + + +import java.io.IOException; +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; + + // The map to cache return type. + private ConcurrentHashMap> localFutureReturnTypenameCache = new ConcurrentHashMap<>(); + + private ConcurrentHashMap> localFutureCache = new ConcurrentHashMap<>(); + + public NativeRpcClient() { + rpcClientPointer = nativeNewRpcClient(); + } + + public void connect(String serverAddress) { + if (rpcClientPointer == -1) { + throw new RuntimeException("no init"); + } + nativeConnect(rpcClientPointer, serverAddress); + codec = new Codec(); + } + + 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(Class returnClz, String funcName, Object[] args) { + if (rpcClientPointer == -1) { + throw new RuntimeException("no init"); + } + + byte[] encodedBytes = null; + try { + encodedBytes = codec.encode(funcName, args); + } catch (IOException e) { + throw new RuntimeException("..."); + } + + if (encodedBytes == null) { + return null; + } + + 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() { + if (rpcClientPointer == -1) { + throw new RuntimeException("no init"); + } + + nativeDestroy(rpcClientPointer); + 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[] encodedReturnValue) throws IOException { +// if (requestId not is local_cache) {//error} + synchronized (this) { + final Class returnClz = localFutureReturnTypenameCache.get(requestId); + CompletableFuture future = localFutureCache.get(requestId); + Object o = codec.decodeReturnValue(returnClz, encodedReturnValue); + future.complete(o); + } + } + + 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..3239416 --- /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(Class returnClz, String funcName, Object[] args); + + void close(); +} 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); + } +} 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..03445f7 --- /dev/null +++ b/java/src/test/java/org/restrpc/test/BasicClientTest.java @@ -0,0 +1,24 @@ +package org.restrpc.test; + +import org.restrpc.client.NativeRpcClient; +import org.restrpc.client.RpcClient; +import org.testng.Assert; +import org.testng.annotations.Test; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; + +public class BasicClientTest { + + @Test + 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); + Assert.assertEquals(future.get(), 5); + + CompletableFuture future1 = rpcClient.asyncFunc("echo").invoke(String.class, "hello world"); + Assert.assertEquals(future1.get(), "hello world"); + } + +} diff --git a/jni/CMakeLists.txt b/jni/CMakeLists.txt new file mode 100644 index 0000000..dd2515a --- /dev/null +++ b/jni/CMakeLists.txt @@ -0,0 +1,23 @@ +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" + "../../third/msgpack/include" + "../jni" + ${JNI_INCLUDE_DIRS}) + +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..3541918 --- /dev/null +++ b/jni/org_restrpc_client_NativeRpcClient.cc @@ -0,0 +1,137 @@ +/* DO NOT EDIT THIS FILE - it is machine generated */ + +#include "org_restrpc_client_NativeRpcClient.h" +#include +#include + +#include + +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) {} + +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()); + 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); + std::string str(size, 0); + env->GetByteArrayRegion(bytes, 0, size, reinterpret_cast(&str.front())); + return str; +} + +// TODO(qwang) +// 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 *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; + jvm->AttachCurrentThreadAsDaemon(reinterpret_cast(&env), nullptr); + jbyteArray javaByteArray = NativeStringToJavaByteArray(env, data); + env->CallVoidMethod(java_object_native_rpc_client, java_method_onResultReceived, request_id, javaByteArray); + }; + + rest_rpc::rpc_client *native_rpc_client = new rest_rpc::rpc_client( + rest_rpc::client_language_t::JAVA, on_result_received); + return reinterpret_cast(native_rpc_client); +} + + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeConnect + * Signature: (JLjava/lang/String;)V + */ +JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeConnect +(JNIEnv *env, jobject o, jlong rpcClientPointer, jstring serverAddress) { + auto *native_rpc_client = reinterpret_cast(rpcClientPointer); + // 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)); +} + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeInvoke + * Signature: (J[B)J + */ +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->internal_async_call(encodedFuncNameAndArgs); +} + +/* + * Class: org_restrpc_client_NativeRpcClient + * Method: nativeDestroy + * Signature: (J)V + */ +JNIEXPORT void JNICALL Java_org_restrpc_client_NativeRpcClient_nativeDestroy +(JNIEnv *, jobject, jlong rpcClientPointer) { + 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 +} +#endif diff --git a/jni/org_restrpc_client_NativeRpcClient.h b/jni/org_restrpc_client_NativeRpcClient.h new file mode 100644 index 0000000..e4ab0b4 --- /dev/null +++ b/jni/org_restrpc_client_NativeRpcClient.h @@ -0,0 +1,48 @@ +/* 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 + +extern JavaVM *jvm; + +#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, jbyteArray); + +/* + * 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