Work around.

examples/client/main.cpp

@@ -16,8 +16,8 @@ void test_add() {
 		}
 
 		{
-			auto result = client.call<int>("add", 1, 2);
-			std::cout << result << std::endl;
+//			auto result = client.call<int>("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();

examples/server/main.cpp

@@ -7,6 +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; 
 	}
 };

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

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;
@@ -55,8 +66,32 @@ namespace rest_rpc {
 			});
 		}
 
-		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();
 			});
@@ -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::promise<req_result>>();
         std::future<req_result> future = p->get_future();
-
         uint64_t fu_id = 0;
         {
             std::unique_lock<std::mutex> lock(cb_mtx_);
@@ -539,6 +573,13 @@ 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) {
+                // 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) {
@@ -552,17 +593,15 @@ namespace rest_rpc {
                     if (!cl->has_timeout()) {
                         cl->cancel();
                         cl->callback(ec, data);
-				}
-				else {
+                    } 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 {
+                } else {
                     std::unique_lock<std::mutex> lock(cb_mtx_);
-				auto& f = future_map_[req_id];
+                    auto &f = future_map_[req_id];
                     if (ec) {
                         //LOG<<ec.message();
                         if (!f) {
@@ -572,9 +611,10 @@ namespace rest_rpc {
                     }
 
                     assert(f);
-				f->set_value(req_result{ data });
+                    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<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/src/main/java/org/restrpc/client/AsyncRpcFunction.java

@@ -4,25 +4,25 @@ import java.util.concurrent.CompletableFuture;
 
 public interface AsyncRpcFunction {
 
-    CompletableFuture<Object> invoke();
+//    CompletableFuture<Object> invoke();
+//
+//    <Arg1Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1);
 
-    <Arg1Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1);
+    <ReturnType, Arg1Type, Arg2Type>
+    RestFuture<ReturnType> invoke(Arg1Type arg1, Arg2Type arg2);
 
-    <Arg1Type, Arg2Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2);
-
-    <Arg1Type, Arg2Type, Arg3Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3);
-
-
-    <Arg1Type, Arg2Type, Arg3Type, Arg4Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4);
-
-    <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5);
-
-    <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type, Arg6Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6);
+//    <Arg1Type, Arg2Type, Arg3Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3);
+//
+//
+//    <Arg1Type, Arg2Type, Arg3Type, Arg4Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4);
+//
+//    <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5);
+//
+//    <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type, Arg6Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5, Arg6Type arg6);
 
 }

java/src/main/java/org/restrpc/client/AsyncRpcFunctionImpl.java

@@ -13,47 +13,47 @@ public class AsyncRpcFunctionImpl implements AsyncRpcFunction {
         this.funcName = funcName;
     }
 
-    public CompletableFuture<Object> invoke() {
-        return internalInvoke(new Object[0]);
-    }
+//    public CompletableFuture<Object> invoke() {
+//        return internalInvoke(new Object[0]);
+//    }
+//
+//    public <Arg1Type> CompletableFuture<Object> invoke(Arg1Type arg1) {
+//        Object[] args = new Object[] {arg1};
+//        return internalInvoke(args);
+//    }
 
-    public <Arg1Type> CompletableFuture<Object> invoke(Arg1Type arg1) {
-        Object[] args = new Object[] {arg1};
-        return internalInvoke(args);
-    }
-
-    public <Arg1Type, Arg2Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2) {
+    public <ReturnType, Arg1Type, Arg2Type>
+    RestFuture<ReturnType> invoke(Arg1Type arg1, Arg2Type arg2) {
         Object[] args = new Object[] {arg1, arg2};
         return internalInvoke(args);
     }
 
-    public <Arg1Type, Arg2Type, Arg3Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) {
-        Object[] args = new Object[] {arg1, arg2, arg3};
-        return internalInvoke(args);
-    }
+//    public <Arg1Type, Arg2Type, Arg3Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3) {
+//        Object[] args = new Object[] {arg1, arg2, arg3};
+//        return internalInvoke(args);
+//    }
+//
+//
+//    public <Arg1Type, Arg2Type, Arg3Type, Arg4Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) {
+//        Object[] args = new Object[] {arg1, arg2, arg3, arg4};
+//        return internalInvoke(args);
+//    }
+//
+//    public <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type>
+//    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5) {
+//        Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5};
+//        return internalInvoke(args);
+//    }
+//
+//    public <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type, Arg6Type>
+//    CompletableFuture<Object> 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 <Arg1Type, Arg2Type, Arg3Type, Arg4Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4) {
-        Object[] args = new Object[] {arg1, arg2, arg3, arg4};
-        return internalInvoke(args);
-    }
-
-    public <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type>
-    CompletableFuture<Object> invoke(Arg1Type arg1, Arg2Type arg2, Arg3Type arg3, Arg4Type arg4, Arg5Type arg5) {
-        Object[] args = new Object[] {arg1, arg2, arg3, arg4, arg5};
-        return internalInvoke(args);
-    }
-
-    public <Arg1Type, Arg2Type, Arg3Type, Arg4Type, Arg5Type, Arg6Type>
-    CompletableFuture<Object> 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<Object> internalInvoke(Object[] args) {
+    private <ReturnType> RestFuture<ReturnType> internalInvoke(Object[] args) {
         return rpcClient.invoke(funcName, args);
     }
 }

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

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<Long, CompletableFuture<Object>> localFutureCache = new HashMap<>();
+    // The map to cache return type.
+    private ConcurrentHashMap<Long, String> localFutureReturnTypenameCache = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<Long, RestFuture<?>> localFutureCache = new ConcurrentHashMap<>();
 
     public NativeRpcClient() {
         rpcClientPointer = nativeNewRpcClient();
@@ -34,7 +39,7 @@ public class NativeRpcClient implements RpcClient {
         return new AsyncRpcFunctionImpl(this, funcName);
     }
 
-    public CompletableFuture<Object> invoke(String funcName, Object[] args) {
+    public <ReturnType> RestFuture<ReturnType> 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<Object> futureToReturn = new CompletableFuture<>();
+        RestFuture<ReturnType> futureToReturn = new RestFuture<ReturnType>() {};
         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);

java/src/main/java/org/restrpc/client/RestFuture.java

@@ -0,0 +1,38 @@
+package org.restrpc.client;
+
+
+import java.io.IOException;
+
+public class RestFuture<T> {
+
+    private Class<T> metaType;
+
+    private T value;
+
+    public RestFuture(Class<T> 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;
+    }
+}

java/src/main/java/org/restrpc/client/RpcClient.java

@@ -8,7 +8,7 @@ public interface RpcClient {
 
     AsyncRpcFunction asyncFunc(String funcName);
 
-    CompletableFuture<Object> invoke(String funcName, Object[] args);
+    <ReturnType> RestFuture<ReturnType> invoke(String funcName, Object[] args);
 
     void close();
 }

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<Integer> 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<Integer>() {};
+//        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();
+    }
 }

jni/org_restrpc_client_NativeRpcClient.cc

@@ -2,13 +2,45 @@
 
 #include "org_restrpc_client_NativeRpcClient.h"
 #include <rest_rpc.hpp>
+#include <iostream>
 
 #include <jni.h>
 
-#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<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) {}
+
+/// 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);
@@ -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<void **>(&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<long>(native_rpc_client) << std::endl;
     return reinterpret_cast<long>(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<rest_rpc::rpc_client *>(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;
 }
 
 /*

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