#56: Support async/await in Rust -> Java direction.

Implemented the invoke_async function that returns a Future, which is completed via the Receiver of a oneshot channel. In Java side, the methods that are allowed to be invoked by invoke_async, are the ones that return a Java Future. When the Java Future completes, the Java side of j4rs invokes native Rust code that completes the Rust pending Future with either success or failure, using the Sender of the oneshot channel that was created when the invoke_async was called.

Please note that it is best that the Java methods that are invoked by invoke_async function return a CompletableFuture, as this improves performance.
j4rs handles simple Java Futures with polling, using an internal one-threaded ScheduledExecutorService. This has apparent performance issues.

Author: astonbitecode

java/src/main/java/org/astonbitecode/j4rs/api/Instance.java

@@ -41,14 +41,19 @@ public interface Instance<T> extends ObjectValue, JsonValue {
 
     /**
      * Invokes asynchronously a method of the instance of the class that is set for this {@link Instance}.
-     * The result of the invocation should be provided later using the performCallback method of a {@link org.astonbitecode.j4rs.api.invocation.NativeCallbackSupport} class.
-     * Any possible returned objects from the actual synchronous invocation of the defined method will be dropped.
+     * The result of the invocation must be a {@link java.util.concurrent.Future}.
+     * When the Future returned from the invocation completes, j4rs will invoke native Rust code to either send a success value or a failure.
+     * <p>
+     * Please note that it is best that this function returns a {@link java.util.concurrent.CompletableFuture}, as this improves performance.
+     * j4rs handles simple {@link java.util.concurrent.Future}s with polling using an internal {@link java.util.concurrent.ScheduledExecutorService} with one thread and this has apparent performance issues.
+     * You may have a look at {@link org.astonbitecode.j4rs.api.async.J4rsPolledFuture} for more details.
      *
-     * @param functionPointerAddress The address of the function pointer that will be used later in the native side in order to actually paerform the callback.
+     * @param functionPointerAddress The address of the function pointer that will be used when the {@link java.util.concurrent.Future} completes, in the native side, in order to actually perform the callback
+     *                               and complete a Future that is created in Rust and awaits for the Java Future to complete.
      * @param methodName             The method name
      * @param args                   The arguments to use when invoking the callback method (the functionPointer)
      */
-    void invokeAsync(long functionPointerAddress, String methodName, InvocationArg... args);
+    void invokeAsyncToChannel(long functionPointerAddress, String methodName, InvocationArg... args);
 
     /**
      * Invokes a method of the instance of the class that is set for this {@link Instance}.
@@ -108,7 +113,7 @@ static <T> Instance cloneInstance(Instance from) {
     default T getOrDeserializeJavaObject() {
         boolean isSerialized = false;
         if (InvocationArg.class.isAssignableFrom(this.getClass())) {
-            isSerialized = ((InvocationArg)this).isSerialized();
+            isSerialized = ((InvocationArg) this).isSerialized();
         }
         if (!isSerialized) {
             return (T) this.getObject();

java/src/main/java/org/astonbitecode/j4rs/api/async/J4rsAsyncContext.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2023 astonbitecode
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astonbitecode.j4rs.api.async;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+class J4rsAsyncContext {
+    private static final ScheduledExecutorService SERVICE = Executors.newSingleThreadScheduledExecutor();
+
+    static void schedule(Runnable r) {
+        SERVICE.schedule(r, 10, TimeUnit.NANOSECONDS);
+    }
+}

java/src/main/java/org/astonbitecode/j4rs/api/async/J4rsPolledFuture.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2023 astonbitecode
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astonbitecode.j4rs.api.async;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * A {@link CompletableFuture} that completes by polling a {@link Future}.
+ * @param <T>
+ */
+public class J4rsPolledFuture<T> extends CompletableFuture<T> {
+    private Future<T> future;
+
+    public J4rsPolledFuture(Future<T> future) {
+        this.future = future;
+        J4rsAsyncContext.schedule(this::tryToComplete);
+    }
+
+    private void tryToComplete() {
+        if (future.isDone()) {
+            try {
+                complete(future.get());
+            } catch (InterruptedException error) {
+                completeExceptionally(error);
+            } catch (ExecutionException error) {
+                completeExceptionally(error.getCause());
+            }
+            return;
+        }
+
+        if (future.isCancelled()) {
+            cancel(true);
+            return;
+        }
+
+        J4rsAsyncContext.schedule(this::tryToComplete);
+    }
+}

java/src/main/java/org/astonbitecode/j4rs/api/dtos/InvocationArg.java

@@ -126,9 +126,9 @@ public Instance invokeStatic(String methodName, InvocationArg... args) {
     }
 
     @Override
-    public void invokeAsync(long functionPointerAddress, String methodName, InvocationArg... args) {
+    public void invokeAsyncToChannel(long channelAddress, String methodName, InvocationArg... args) {
         if (getInstance() != null) {
-            getInstance().invokeAsync(functionPointerAddress, methodName, args);
+            getInstance().invokeAsyncToChannel(channelAddress, methodName, args);
         }
     }
 

java/src/main/java/org/astonbitecode/j4rs/api/invocation/EagerJsonInvocationImpl.java

@@ -42,7 +42,7 @@ public Instance invokeStatic(String methodName, InvocationArg... arg) {
     }
 
     @Override
-    public void invokeAsync(long functionPointer, String methodName, InvocationArg... args) {
+    public void invokeAsyncToChannel(long channelAddress, String methodName, InvocationArg... args) {
         throw new RuntimeException("Not implemented yet. Please use the JsonInvocationImpl instead");
     }
 

java/src/main/java/org/astonbitecode/j4rs/api/invocation/JavaFxInvocation.java

@@ -60,8 +60,8 @@ public Instance invokeStatic(String methodName, InvocationArg... args) {
     }
 
     @Override
-    public void invokeAsync(long functionPointerAddress, String methodName, InvocationArg... args) {
-        Platform.runLater(() -> jsonInvocation.invokeAsync(functionPointerAddress, methodName, args));
+    public void invokeAsyncToChannel(long channelAddress, String methodName, InvocationArg... args) {
+        Platform.runLater(() -> jsonInvocation.invokeAsyncToChannel(channelAddress, methodName, args));
     }
 
     @Override

java/src/main/java/org/astonbitecode/j4rs/api/invocation/JsonInvocationImpl.java

@@ -16,6 +16,7 @@
 
 import org.astonbitecode.j4rs.api.Instance;
 import org.astonbitecode.j4rs.api.JsonValue;
+import org.astonbitecode.j4rs.api.async.J4rsPolledFuture;
 import org.astonbitecode.j4rs.api.dtos.GeneratedArg;
 import org.astonbitecode.j4rs.api.dtos.InvocationArg;
 import org.astonbitecode.j4rs.api.dtos.InvocationArgGenerator;
@@ -34,6 +35,8 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 
 public class JsonInvocationImpl<T> implements Instance<T> {
@@ -65,7 +68,6 @@ public JsonInvocationImpl(T instance, Class<T> clazz, List<Type> classGenTypes)
 
     @Override
     public Instance invoke(String methodName, InvocationArg... args) {
-        // Invoke the instance
         try {
             CreatedInstance createdInstance = invokeMethod(methodName, gen.generateArgObjects(args));
             return InstanceGenerator.create(createdInstance.object, createdInstance.clazz, createdInstance.classGenTypes);
@@ -85,18 +87,32 @@ public Instance invokeStatic(String methodName, InvocationArg... args) {
     }
 
     @Override
-    public void invokeAsync(long functionPointerAddress, String methodName, InvocationArg... args) {
-        // Check that the class of the invocation extends the NativeCallbackSupport
-        if (!NativeCallbackSupport.class.isAssignableFrom(this.clazz)) {
-            throw new InvocationException("Cannot invoke asynchronously the class " + this.clazz.getName() + ". The class does not extend the class " + NativeCallbackSupport.class.getName());
-        } else {
-            // Initialize the pointer
-            ((NativeCallbackSupport) object).initPointer(new RustPointer(functionPointerAddress));
-            // Invoke (any possible returned objects will be dropped)
-            invoke(methodName, args);
+    public void invokeAsyncToChannel(final long channelAddress, final String methodName, final InvocationArg... args) {
+        try {
+            NativeCallbackToRustFutureSupport callback = newCallbackForAsyncToChannel(channelAddress);
+            invokeAsyncMethod(methodName, gen.generateArgObjects(args)).handle((o, t) -> {
+                if (o != null) {
+                    callback.doCallbackSuccess(o);
+                } else if (t != null) {
+                    callback.doCallbackFailure(t);
+                } else {
+                    String message = String.format("Error while handling the future returned by the invocation of method %s of class %s",
+                            methodName, getObjectClassName());
+                    throw new InvocationException(message);
+                }
+                return Void.TYPE;
+            });
+        } catch (Exception error) {
+            throw new InvocationException("While invoking method " + methodName + " of Class " + getObjectClassName(), error);
         }
     }
 
+    NativeCallbackToRustFutureSupport newCallbackForAsyncToChannel(long channelAddress) {
+        NativeCallbackToRustFutureSupport callback = new NativeCallbackToRustFutureSupport();
+        callback.initPointer(new RustPointer(channelAddress));
+        return callback;
+    }
+
     @Override
     public void invokeToChannel(long channelAddress, String methodName, InvocationArg... args) {
         initializeCallbackChannel(channelAddress);
@@ -185,6 +201,56 @@ CreatedInstance invokeMethod(String methodName, GeneratedArg[] generatedArgs) th
         return new CreatedInstance(invokedMethodReturnType, returnedObject, retClassGenTypes);
     }
 
+    CompletableFuture<Object> invokeAsyncMethod(String methodName, GeneratedArg[] generatedArgs) throws Exception {
+        Class[] argTypes = Arrays.stream(generatedArgs)
+                .map(invGeneratedArg -> {
+                    try {
+                        return invGeneratedArg.getClazz();
+                    } catch (Exception error) {
+                        throw new InvocationException("Cannot parse the parameter types while invoking async method", error);
+                    }
+                })
+                .toArray(size -> new Class[size]);
+        Object[] argObjects = Arrays.stream(generatedArgs)
+                .map(invGeneratedArg -> {
+                    try {
+                        return invGeneratedArg.getObject();
+                    } catch (Exception error) {
+                        throw new InvocationException("Cannot parse the parameter objects while invoking async method", error);
+                    }
+                })
+                .toArray(size -> new Object[size]);
+
+        Method methodToInvoke = findMethodInHierarchy(this.clazz, methodName, argTypes);
+        List<Type> retClassGenTypes = new ArrayList<>();
+
+        Type returnType = methodToInvoke.getGenericReturnType();
+
+        if (returnType instanceof ParameterizedType) {
+            ParameterizedType type = (ParameterizedType) returnType;
+            retClassGenTypes = Arrays.asList(type.getActualTypeArguments());
+        }
+
+        CompletableFuture<Object> future;
+        Class<?> invokedMethodReturnType = methodToInvoke.getReturnType();
+        if (!invokedMethodReturnType.isAssignableFrom(Future.class)) {
+            String message = String.format("Attempted to asynchronously invoke method %s of class %s that returns %s instead of returning Future",
+                    methodName,
+                    this.clazz.getName(),
+                    returnType.getTypeName());
+            throw new InvocationException(message);
+        }
+        Future<Object> invocationReturnedFuture = (Future<Object>) methodToInvoke.invoke(this.object, argObjects);
+
+        if (invocationReturnedFuture instanceof CompletableFuture) {
+            future = (CompletableFuture<Object>) invocationReturnedFuture;
+        } else {
+            future = new J4rsPolledFuture<>(invocationReturnedFuture);
+        }
+
+        return future;
+    }
+
     Method findMethodInHierarchy(Class clazz, String methodName, Class[] argTypes) throws NoSuchMethodException {
         // Get the declared and methods defined in the interfaces of the class.
         Set<Method> methods = new HashSet<>(Arrays.asList(clazz.getDeclaredMethods()));

java/src/main/java/org/astonbitecode/j4rs/api/invocation/NativeCallbackSupport.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2023 astonbitecode
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astonbitecode.j4rs.api.invocation;
+
+import org.astonbitecode.j4rs.api.Instance;
+import org.astonbitecode.j4rs.errors.InvocationException;
+import org.astonbitecode.j4rs.rust.RustPointer;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Optional;
+
+/**
+ * Performs native callbacks to Rust channels that are transformed to Rust Futures
+ */
+class NativeCallbackToRustFutureSupport {
+    private static native int docallbacktochannel(long channelPointerAddress, Instance inv);
+    private static native int failcallbacktochannel(long channelPointerAddress, String stacktrace);
+
+    private Optional<RustPointer> channelPointerOpt = Optional.empty();
+
+    static void initialize(String libname) {
+        try {
+            System.loadLibrary(libname);
+        } catch (UnsatisfiedLinkError error) {
+            System.err.println("The Callbacks are not initialized because the j4rs lib was not found. You may ignore this error if you don't use callbacks.");
+            error.printStackTrace();
+        }
+    }
+
+    /**
+     * Perform a callback to signal successful operation
+     *
+     * @param obj The {@link Object} to pass in the callback.
+     */
+    public void doCallbackSuccess(Object obj) {
+        if (channelPointerOpt.isPresent() && obj != null) {
+            docallbacktochannel(channelPointerOpt.get().getAddress(), InstanceGenerator.create(obj, obj.getClass()));
+        } else {
+            throw new InvocationException("Cannot do callback. Please make sure that you don't try to access this method while being in the constructor of your class (that extends NativeCallbackSupport)");
+        }
+    }
+
+    /**
+     * Perform a callback to signal failure
+     * @param error The error
+     */
+    public void doCallbackFailure(Throwable error) {
+        if (channelPointerOpt.isPresent() && error != null) {
+            StringWriter sw = new StringWriter();
+            PrintWriter pw = new PrintWriter(sw);
+            error.printStackTrace(pw);
+            String stringStackTrace = sw.toString();
+            failcallbacktochannel(channelPointerOpt.get().getAddress(), stringStackTrace);
+        } else {
+            throw new InvocationException("Cannot do callback for failure. Please make sure that you don't try to access this method while being in the constructor of your class (that extends NativeCallbackSupport). The failure was: ", error);
+        }
+    }
+
+    final void initPointer(RustPointer p) {
+        this.channelPointerOpt = Optional.of(p);
+    }
+}