Arrow-native RPC.
From subprocess
to service.

from typing import Protocol

class Calculator(Protocol):
    def add(self, a: float, b: float) -> float: ...

from protocol import Calculator
from vgi_rpc import run_server

class CalculatorImpl:
    def add(self, a: float, b: float) -> float:
        return a + b

run_server(Calculator, CalculatorImpl())

from protocol import Calculator
from vgi_rpc import connect

with connect(Calculator, ["python", "worker.py"]) as calc:
    print(calc.add(a=2.0, b=3.0))  # 5.0

import { Protocol, float } from "@query-farm/vgi-rpc";

export const calculator = new Protocol("Calculator");
calculator.unary("add", {
  params: { a: float, b: float },
  result: { result: float },
  handler: async ({ a, b }) => ({ result: a + b }),
});

import { VgiRpcServer } from "@query-farm/vgi-rpc";
import { calculator } from "./protocol";

const server = new VgiRpcServer(calculator);
server.run();

import { subprocessConnect } from "@query-farm/vgi-rpc";

const client = subprocessConnect(
  ["bun", "run", "worker.ts"],
  { protocol: "Calculator" },
);
try {
  console.log(await client.call("add", { a: 2, b: 3 }));
  // { result: 5 }
} finally {
  client.close();
}

type AddParams struct {
    A float64 `vgirpc:"a"`
    B float64 `vgirpc:"b"`
}

var paramsSchema = arrow.NewSchema([]arrow.Field{
    {Name: "a", Type: arrow.PrimitiveTypes.Float64},
    {Name: "b", Type: arrow.PrimitiveTypes.Float64},
}, nil)
var resultSchema = arrow.NewSchema([]arrow.Field{
    {Name: "result", Type: arrow.PrimitiveTypes.Float64},
}, nil)

server := vgirpc.NewServer()
server.SetServiceName("Calculator")
vgirpc.Unary(server, "add", func(
    _ context.Context, _ *vgirpc.CallContext, p AddParams,
) (float64, error) {
    return p.A + p.B, nil
})

log.Fatal(http.ListenAndServe(
    "127.0.0.1:8080", vgirpc.NewHttpServer(server),
))

client, err := vgirpc.NewHttpClient(
    "http://127.0.0.1:8080",
    vgirpc.WithClientProtocol("Calculator"),
)
if err != nil { log.Fatal(err) }
defer client.Close()

builder := array.NewRecordBuilder(memory.DefaultAllocator, paramsSchema)
defer builder.Release()
builder.Field(0).(*array.Float64Builder).Append(2)
builder.Field(1).(*array.Float64Builder).Append(3)
params := builder.NewRecordBatch()
defer params.Release()

result, err := client.CallUnary(
    context.Background(), "add", params, resultSchema,
)
if err != nil { log.Fatal(err) }
defer result.Release()
fmt.Println(result.Batch.Column(0).(*array.Float64).Value(0)) // 5

use vgi_rpc::{service, Result};

pub struct Calculator;

#[service]
impl Calculator {
    #[unary]
    fn add(&self, a: f64, b: f64) -> Result<f64> {
        Ok(a + b)
    }
}

mod protocol;
use protocol::Calculator;
use std::{io, sync::Arc};
use vgi_rpc::RpcServer;

fn main() {
    let mut server = RpcServer::builder()
        .protocol_name("Calculator")
        .build();
    Calculator::register_with(&mut server, Arc::new(Calculator));
    server.serve(io::stdin().lock(), io::stdout().lock());
}

use std::sync::Arc;
use arrow_array::{Float64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema};
use vgi_rpc_client::RpcClient;

fn main() -> vgi_rpc_client::Result<()> {
    let mut client = RpcClient::connect(&["./target/debug/worker"])?
        .protocol("Calculator");
    let schema = Arc::new(Schema::new(vec![
        Field::new("a", DataType::Float64, false),
        Field::new("b", DataType::Float64, false),
    ]));
    let params = RecordBatch::try_new(schema, vec![
        Arc::new(Float64Array::from(vec![2.0])),
        Arc::new(Float64Array::from(vec![3.0])),
    ])?;
    let (result, _) = client.call_unary("add", &params, None)?;
    println!("{:?}", result.column(0)); // [5.0]
    Ok(())
}

public interface Calculator {
    double add(double a, double b);
}

import farm.query.vgirpc.RpcServer;
import farm.query.vgirpc.transport.StdioTransport;

public final class CalculatorWorker {
    public static void main(String[] args) {
        Calculator impl = (a, b) -> a + b;
        var server = new RpcServer(Calculator.class, impl);
        try (var transport = new StdioTransport()) {
            server.serve(transport);
        }
    }
}

import farm.query.vgirpc.RpcConnection;
import farm.query.vgirpc.transport.SubprocessTransport;
import java.util.List;

public final class Client {
    public static void main(String[] args) {
        var transport = new SubprocessTransport(List.of(
            "java", "--add-opens=java.base/java.nio=ALL-UNNAMED",
            "-Dio.netty.noUnsafe=false",
            "-cp", "worker.jar", "CalculatorWorker"));
        try (var conn = new RpcConnection(transport)) {
            Calculator calc = conn.proxy(Calculator.class);
            System.out.println(calc.add(2.0, 3.0)); // 5.0
        }
    }
}

public interface ICalculator
{
    Task<double> AddAsync(double a, double b);
}

using QueryFarm.VgiRpc.Server;
using QueryFarm.VgiRpc.Transport;

var server = new RpcServer(typeof(ICalculator), new Calculator());
await server.ServeAsync(new StdioTransport());

public sealed class Calculator : ICalculator
{
    public Task<double> AddAsync(double a, double b) =>
        Task.FromResult(a + b);
}

using QueryFarm.VgiRpc.Client;

await using var client = RpcClient.StartSubprocess(
    ["dotnet", "Worker.dll"]);
var calc = client.CreateProxy<ICalculator>();

Console.WriteLine(await calc.AddAsync(2, 3)); // 5

#pragma once
#include <arrow/api.h>
#include <vgi_rpc/arrow_utils.h>

inline auto params_schema() {
    return arrow::schema({
        arrow::field("a", arrow::float64(), false),
        arrow::field("b", arrow::float64(), false),
    });
}
inline auto result_schema() {
    return arrow::schema({
        arrow::field("result", arrow::float64(), false),
    });
}
inline auto values(double value) {
    arrow::DoubleBuilder builder;
    VGI_RPC_THROW_NOT_OK(builder.Append(value));
    return vgi_rpc::unwrap(builder.Finish());
}

#include "protocol.h"
#include <vgi_rpc/server.h>
#include <vgi_rpc/request.h>
#include <vgi_rpc/result.h>

int main() {
    auto server = vgi_rpc::ServerBuilder()
        .protocol("Calculator")
        .add_unary("add", params_schema(), result_schema(),
            [](const vgi_rpc::Request& req, vgi_rpc::CallContext&) {
                auto sum = req.get<double>("a") + req.get<double>("b");
                return vgi_rpc::Result::value(result_schema(), {values(sum)});
            })
        .build();
    server->run();
}

#include "protocol.h"
#include <vgi_rpc/client.h>
#include <iostream>

int main() {
    vgi_rpc::RpcClientOptions options;
    options.protocol = "Calculator";
    auto client = vgi_rpc::RpcClient::spawn({"./worker"}, options);
    auto params = arrow::RecordBatch::Make(
        params_schema(), 1, {values(2), values(3)});
    auto result = client.call_unary("add", params);
    auto sum = std::static_pointer_cast<arrow::DoubleArray>(
        result.batch->column(0));
    std::cout << sum->Value(0) << '\n'; // 5
    client.close();
}