From 7e22d4b402b918e61ef43a500620e583c9533fca Mon Sep 17 00:00:00 2001 From: Qing Wang Date: Wed, 21 Oct 2020 20:24:52 +0800 Subject: [PATCH] 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