Merge pull request #26 from jovany-wang/java

Support Java client for rest_rpc

.gitignore

@@ -38,3 +38,9 @@ cmake_install.cmake
 
 # idea
 .idea/
+
+.DS_Store
+*iml
+target/
+
+build/

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()

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;
 }

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;
 	}
 };
 

include/rest_rpc/rpc_client.hpp

@@ -3,14 +3,25 @@
 #include <string>
 #include <deque>
 #include <future>
+#include <utility>
 #include "use_asio.hpp"
 #include "client_util.hpp"
 #include "const_vars.h"
 #include "meta_util.hpp"
+#include <functional>
 
 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<std::thread>([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<void(long, const std::string &)> 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<std::thread>([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<void(long, const std::string&)> 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<std::thread>([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<std::promise<req_result>>();
+            uint64_t fu_id = 0;
+            {
+                std::unique_lock<std::mutex> 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<size_t TIMEOUT = DEFAULT_TIMEOUT, typename... Args>
 		void async_call(const std::string& rpc_name, std::function<void(boost::system::error_code, string_view)> 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<call_t> cl = nullptr;
-				{
-					std::unique_lock<std::mutex> 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<call_t> cl = nullptr;
+                    {
+                        std::unique_lock<std::mutex> 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<std::mutex> lock(cb_mtx_);
-				callback_map_.erase(req_id);
-			}
-			else {
-				std::unique_lock<std::mutex> lock(cb_mtx_);
-				auto& f = future_map_[req_id];
-				if (ec) {
-					//LOG<<ec.message();
-					if (!f) {
-						//std::cout << "invalid req_id" << std::endl;
-						return;
-					}
-				}
+                    std::unique_lock<std::mutex> lock(cb_mtx_);
+                    callback_map_.erase(req_id);
+                } else {
+                    std::unique_lock<std::mutex> lock(cb_mtx_);
+                    auto &f = future_map_[req_id];
+                    if (ec) {
+                        //LOG<<ec.message();
+                        if (!f) {
+                            //std::cout << "invalid req_id" << std::endl;
+                            return;
+                        }
+                    }
 
-				assert(f);
-				f->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<std::string, std::function<void(string_view)>> sub_map_;
 		std::set<std::pair<std::string, std::string>> key_token_set_;
+
+        client_language_t client_language_ = client_language_t::CPP;
+        std::function<void(long, const std::string&)> on_result_received_callback_;
 	};
 }

java/pom.xml

@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.restrpc</groupId>
+    <artifactId>restrpc</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.5</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>29.0-jre</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>1.7.21</version>
+        </dependency>
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>7.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.msgpack</groupId>
+            <artifactId>msgpack-core</artifactId>
+            <version>0.8.21</version>
+        </dependency>
+    </dependencies>
+    <build>
+        <resources>
+            <resource>
+                <directory>native_dependencies</directory>
+            </resource>
+        </resources>
+    </build>
+</project>

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<Object> invoke(Class returnClz);
+
+    <Arg1Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1);
+
+    <Arg1Type, Arg2Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2);
+
+    <Arg1Type, Arg2Type, Arg3Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3);
+
+
+    <Arg1Type, Arg2Type, Arg3Type, Arg4Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4);
+
+    <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5);
+
+    <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type, Arg6Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6);
+
+}

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<Object> invoke(Class returnClz) {
+        return internalInvoke(returnClz, new Object[0]);
+    }
+
+    public <Arg1Type> CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1) {
+        Object[] args = new Object[] {arg1};
+        return internalInvoke(returnClz, args);
+    }
+
+    public <Arg1Type, Arg2Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2) {
+        Object[] args = new Object[] {arg1, arg2};
+        return internalInvoke(returnClz, args);
+    }
+
+    public <Arg1Type, Arg2Type, Arg3Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) {
+        Object[] args = new Object[] {arg1, arg2, arg3};
+        return internalInvoke(returnClz, args);
+    }
+
+
+    public <Arg1Type, Arg2Type, Arg3Type, Arg4Type>
+    CompletableFuture<Object> invoke(Class returnClz, Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) {
+        Object[] args = new Object[] {arg1, arg2, arg3, arg4};
+        return internalInvoke(returnClz, args);
+    }
+
+    public <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type>
+    CompletableFuture<Object> 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 <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type, Arg6Type>
+    CompletableFuture<Object> 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<Object> internalInvoke(Class returnClz, Object[] args) {
+        return rpcClient.invoke(returnClz, funcName, args);
+    }
+}

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);
+    }
+}

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<String> loadedLibs = Sets.newHashSet();
+
+  /**
+   * Loads the native library specified by the <code>libraryName</code> argument.
+   * The <code>libraryName</code> 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);
+    }
+  }
+}

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);
+    }
+  }
+
+}

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<Long, Class<?>> localFutureReturnTypenameCache = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<Long, CompletableFuture<Object>> 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<Object> 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<Object> futureToReturn = new CompletableFuture<Object>();
+            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<Object> 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);
+}

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<Object> invoke(Class returnClz, String funcName, Object[] args);
+
+    void close();
+}

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<Object> 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);
+    }
+}

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<Object> future = rpcClient.asyncFunc("add").invoke(Integer.class, 2, 3);
+        Assert.assertEquals(future.get(), 5);
+
+        CompletableFuture<Object> future1 = rpcClient.asyncFunc("echo").invoke(String.class, "hello world");
+        Assert.assertEquals(future1.get(), "hello world");
+    }
+
+}

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)

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 <rest_rpc.hpp>
+#include <iostream>
+
+#include <jni.h>
+
+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<void **>(&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<jstring>(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<const jbyte *>(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<jbyte *>(&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<void **>(&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<long>(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<rest_rpc::rpc_client *>(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<short>(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<rest_rpc::rpc_client *>(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<rest_rpc::rpc_client *>(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

jni/org_restrpc_client_NativeRpcClient.h

@@ -0,0 +1,48 @@
+/* DO NOT EDIT THIS FILE - it is machine generated */
+#include <jni.h>
+/* 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