WIP

java/pom.xml

@@ -29,6 +29,11 @@
             <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>

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

@@ -54,6 +54,6 @@ public class AsyncRpcFunctionImpl implements AsyncRpcFunction {
     }
 
     private CompletableFuture<Object> internalInvoke(Object[] args) {
-        return rpcClient.invoke(args);
+        return rpcClient.invoke(funcName, args);
     }
 }

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

@@ -0,0 +1,38 @@
+package org.restrpc.client;
+
+import org.msgpack.core.MessageBufferPacker;
+import org.msgpack.core.MessagePack;
+import java.io.IOException;
+
+public class Codec {
+    public byte[] encode(String funcName, Object[] args) throws IOException {
+        // assert args != nullptr.
+
+        MessageBufferPacker messagePacker = MessagePack.newDefaultBufferPacker();
+        messagePacker.packArrayHeader(1 + args.length);
+        messagePacker.packString(funcName);
+
+        for (Object arg : args) {
+            if (arg == null) {
+                messagePacker.packNil();
+                continue;
+            }
+
+            final String argTypeName = arg.getClass().getName();
+            switch (argTypeName) {
+                case "java.lang.Integer":
+                    messagePacker.packInt((int) arg);
+                    break;
+                case "java.lang.Long":
+                    messagePacker.packLong((long) arg);
+                    break;
+                case "java.lang.String":
+                    messagePacker.packString((String) arg);
+                    break;
+                default:
+                    throw new RuntimeException("Unknown type" + argTypeName);
+            }
+        }
+        return messagePacker.toByteArray();
+    }
+}

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

@@ -1,5 +1,7 @@
 package org.restrpc.client;
 
+import java.io.IOException;
+import java.util.HashMap;
 import java.util.concurrent.CompletableFuture;
 
 public class NativeRpcClient implements RpcClient {
@@ -9,6 +11,10 @@ public class NativeRpcClient implements RpcClient {
     }
     private long rpcClientPointer = -1;
 
+    private Codec codec;
+
+    private HashMap<Long, CompletableFuture<Object>> localFutureCache = new HashMap<>();
+
     public NativeRpcClient() {
         rpcClientPointer = nativeNewRpcClient();
     }
@@ -18,6 +24,7 @@ public class NativeRpcClient implements RpcClient {
             throw new RuntimeException("no init");
         }
         nativeConnect(rpcClientPointer, serverAddress);
+        codec = new Codec();
     }
 
     public AsyncRpcFunction asyncFunc(String funcName) {
@@ -27,14 +34,26 @@ public class NativeRpcClient implements RpcClient {
         return new AsyncRpcFunctionImpl(this, funcName);
     }
 
-    public CompletableFuture<Object> invoke(Object[] args) {
+    public CompletableFuture<Object> invoke(String funcName, Object[] args) {
         if (rpcClientPointer == -1) {
             throw new RuntimeException("no init");
         }
 
-        nativeInvoke(rpcClientPointer, null);
-//        return nativeInvoke(rpcClientPointer, );
-        return null;
+        byte[] encodedBytes = null;
+        try {
+            encodedBytes = codec.encode(funcName, args);
+        } catch (IOException e) {
+            throw new RuntimeException("...");
+        }
+
+        if (encodedBytes == null) {
+            return null;
+        }
+
+        final long requestId = nativeInvoke(rpcClientPointer, encodedBytes);
+        CompletableFuture<Object> futureToReturn = new CompletableFuture<>();
+        localFutureCache.put(requestId, futureToReturn);
+        return futureToReturn;
     }
 
     public void close() {
@@ -46,12 +65,11 @@ public class NativeRpcClient implements RpcClient {
         this.rpcClientPointer = -1;
     }
 
-
     private native long nativeNewRpcClient();
 
     private native void nativeConnect(long rpcClientPointer, String serverAddress);
 
-    private native long nativeInvoke(long rpcClientPointer, byte[][] encodedFuncNameAndArgs);
+    private native long nativeInvoke(long rpcClientPointer, byte[] encodedFuncNameAndArgs);
 
     private native void nativeDestroy(long rpcClientPointer);
 }

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

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

java/src/test/java/org/restrpc/test/BasicClientTest.java

@@ -1,14 +1,42 @@
 package org.restrpc.test;
 
+import org.checkerframework.checker.units.qual.C;
+import org.msgpack.core.MessagePack;
+import org.restrpc.client.Codec;
 import org.restrpc.client.NativeRpcClient;
 import org.restrpc.client.RpcClient;
 import org.testng.annotations.Test;
 
+import java.io.IOException;
+
 public class BasicClientTest {
 
     @Test
     public void testBasic() {
         RpcClient rpcClient = new NativeRpcClient();
-        rpcClient.asyncFunc("111").invoke();
+        rpcClient.asyncFunc("111").invoke(2, 3);
+    }
+
+    private String getClassStr(Object o) {
+        return o.getClass().getName();
+    }
+
+    @Test
+    public void testClass() {
+        int a = 3;
+        System.out.println("int -> " + getClassStr(a));
+        long b = 3;
+        System.out.println("long -> " + getClassStr(b));
+        String s = "3";
+        System.out.println("string ->" + getClassStr(s));
+        String c = null;
+        System.out.println("null str ->" + getClassStr(c));
+    }
+
+    @Test
+    public void testEncode() throws IOException {
+        Codec codec = new Codec();
+        Object[] args = new Object[] {2, 3};
+        byte[] bs = codec.encode("add", args);
     }
 }