Page History

public class AsyncScalarFunction<T>AsyncScalarFunction extends UserDefinedFunction {
    @Override
    public final FunctionKind getKind() {
        return FunctionKind.ASYNC_SCALAR;
    }

    @Override
    public TypeInference getTypeInference(DataTypeFactory typeFactory) {
        TypeInference val = TypeInferenceExtractor.forAsyncScalarFunction(typeFactory, getClass());
        return val;
    }
}

public class RemoteCallFunction extends AsyncScalarFunction<String>AsyncScalarFunction {

    private ExternalClient client;
    private ExecutorService executor;

    public RemoteCallFunction() {
    }

    @Override
    public void open(FunctionContext context) throws Exception {
        client = new Client();
        executor = Executors.newFixedThreadPool(
            context.getJobParameter("in-flight-requests", 10));
    }

    @Override
    public void close() throws Exception {
        client.close();
        executor.shutdownNow();
    }

    public final void eval(
            CompletableFuture<String> future,
            String param1,
            int param2) {
        executor.submit(() -> {
            try {
                String resp = client.rpc(param1, param2);
                future.complete(resp);
            } catch (Throwable t) {
                future.completeExceptionally(t);
            }
        });
    }
}