C# Client Library

The Virtufin API Client library (Virtufin.Api.Client) provides a high-level C# interface for interacting with the Virtufin API gateway.

Installation

From Gitea NuGet Registry

Add the Virtufin package source with a personal access token (scope: read:packages):

dotnet nuget add source https://nuget.haenerconsulting.com/api/packages/virtufin/nuget/index.json \
  --name Virtufin --username <your-gitea-username> --password <your-gitea-token>

dotnet add package Virtufin.Api.Client

Or configure via NuGet.Config:

<configuration>
  <packageSources>
    <add key="Virtufin" value="https://nuget.haenerconsulting.com/api/packages/virtufin/nuget/index.json" />
  </packageSources>
  <packageSourceCredentials>
    <Virtufin>
      <add key="Username" value="<your-gitea-username>" />
      <add key="ClearTextPassword" value="<your-gitea-token>" />
    </Virtufin>
  </packageSourceCredentials>
</configuration>

From Local Source

<ProjectReference Include="path/to/Virtufin.Api.Client/Virtufin.Api.Client.csproj" />

Overview

The client library provides three main classes:

Class	Description
`ApiClient`	Main entry point with async-only operations
`GatewayClient`	Low-level gateway operations (async-only)
`ServiceClient`	Service-specific method invocation (async-only)

Note: This library is async-only. For synchronous usage patterns, wrap calls in Task.Run() or use a different approach.

Quick Start

using Virtufin.Api.Client;

// Create and dispose when done
        using var client = new ApiClient("localhost", 5002);

        // Discover services
        Console.WriteLine("Available Services:");
        var services = await client.ListServicesAsync();
        foreach (var service in services)
        {
            Console.WriteLine($"  - {service}");

            // List methods for each service
            var methods = await client.ListMethodsAsync(service);
            foreach (var method in methods.Take(3)) // First 3 methods
            {
                var streaming = method["isServerStreaming"] == "True" ? " (stream)" : "";
                Console.WriteLine($"    {method["name"]}{streaming}");
            }
            if (methods.Count > 3)
                Console.WriteLine($"    ... and {methods.Count - 3} more");
        }

        Console.WriteLine();

        // Invoke a method
        Console.WriteLine("Invoking workmanager.ListWorkers:");
        try
        {
            var response = await client.InvokeAsync("workmanager", "ListWorkers");
            Console.WriteLine($"Response: {System.Text.Json.JsonSerializer.Serialize(response)}");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Error: {ex.Message}");
        }
    }
}

Dynamic Dispatch

The C# client uses the dynamic keyword to allow client.Gateway.<service>.<method>() syntax. Under the hood, GatewayClient and ServiceClient both extend System.Dynamic.DynamicObject and override TryGetMember:

client.Gateway.<service> triggers GatewayClient.TryGetMember, which returns a cached ServiceClient for the named service.
<service>.<method> triggers ServiceClient.TryGetMember, which returns a Func<Dictionary<string, object?>, Task<Dictionary<string, object?>>> that invokes the named method via the Gateway's InvokeJson RPC.

The dictionary argument is serialized to JSON and sent to the Gateway, which unmarshals it to the typed proto request, runs the backend RPC, and marshals the response back to JSON.

Dynamic Keyword Semantics

Because the accessor returns a dynamic, you can either:

// Option 1: assign to a 'dynamic' variable, then call methods on it.
dynamic workmanager = client.Gateway.workmanager;
var workers = await workmanager.ListWorkers();

// Option 2: chain the access in a single expression.
var workers = await client.Gateway.workmanager.ListWorkers();

The result of <service>.<method>(...) is Task<Dictionary<string, object?>>, so it must be awaited. The Dictionary<string, object?> is the JSON-unmarshaled response; field access uses string keys (e.g. w["id"]).

WorkManager

using var client = new ApiClient("localhost", 5002);
dynamic workmanager = client.Gateway.workmanager;

// ListWorkers — no request fields; returns {"workers": [...]}
var workers = await workmanager.ListWorkers();
foreach (var w in workers["workers"])
    Console.WriteLine($"  {w["id"]}: {w["status"]} ({w["language"]})");

// CreateWorker — CodeSource oneof (url | content), mime_type, topic, group
var createResult = await workmanager.CreateWorker(new Dictionary<string, object?>
{
    ["code_source"] = new Dictionary<string, object?> { ["url"] = "https://example.com/worker.py" },
    ["mime_type"] = "text/x-python",
    ["topic"] = "worker-commands",
});
var workerId = (string)createResult["id"];

// StartWorker — id field
await workmanager.StartWorker(new Dictionary<string, object?> { ["id"] = workerId });

WebSocketManager

dynamic wsm = client.Gateway.websocketmanager;

// List — no request fields; returns {"connections": [...]}
var connections = await wsm.List();
foreach (var c in connections["connections"])
    Console.WriteLine($"  {c["id"]}: {c["url"]} ({c["status"]})");

// Connect — url, auto_reconnect
var conn = await wsm.Connect(new Dictionary<string, object?>
{
    ["url"] = "wss://stream.example.com/feed",
    ["auto_reconnect"] = true,
});
var connectionId = (string)conn["id"];

// Disconnect — id field
await wsm.Disconnect(new Dictionary<string, object?> { ["id"] = connectionId });

See also: proto-to-client-mapping spec §Layer 3a (C# wrapper) and §Layer 4 (Dynamic dispatch).

Pub/Sub Operations

The ApiClient provides pub/sub messaging via the Pubsub gRPC service.

Methods

Task<PublishResponse> PublishEventAsync(string topic, CloudEvent cloudEvent, CancellationToken cancellationToken = default)
Task<PubsubSubscribeResponse> PublishWithResultAsync(string topic, CloudEvent cloudEvent, string replyTopic, TimeSpan? timeout = null, CancellationToken cancellationToken = default)
AsyncServerStreamingCall<PubsubSubscribeResponse> SubscribeToTopic(string topic, CancellationToken cancellationToken = default)
Task<UnsubscribeResponse> UnsubscribeFromTopicAsync(string subscriptionId, CancellationToken cancellationToken = default)

PublishEventAsync publishes a CloudEvent to the given topic. Parameters:

topic — Topic to publish to
cloudEvent — The CloudEvent message (use CloudNative.CloudEvents.V1.CloudEvent proto)
cancellationToken — Optional cancellation token

PublishWithResultAsync implements the request-reply pattern over pub/sub using correlation IDs. Parameters:

topic — Topic to publish the request to
cloudEvent — The CloudEvent message (use CloudNative.CloudEvents.V1.CloudEvent proto)
replyTopic — Topic to listen for the correlated response on
timeout — Maximum wait time (default: 30s)
cancellationToken — Optional cancellation token

Raises: TimeoutException if no response arrives within the timeout.

Note on broadcast semantics: Pubsub.Subscribe is a per-API-process broadcast subscription — multiple in-process callers on the same topic all receive every message (per the Pub/Sub Topics spec §Subscription Implementation Pattern). If you want a single in-process handler to receive each message, manage the SubscribeToTopic call yourself. If you want multiple in-process handlers each receiving every message, just call SubscribeToTopic multiple times.

Resource usage: PublishWithResultAsync shares one persistent gRPC Subscribe stream per reply topic across all callers (started lazily on first use, reconnecting with backoff if it drops) and demuxes responses by correlationid. The underlying Dapr subscription is created once and stays warm; per-call cost is one pending waiter entry. Readers are cancelled by Dispose()/DisposeAsync().

Example: Publish

var cloudEvent = new CloudEvent
{
    Id = Guid.NewGuid().ToString(),
    Source = new Uri("urn:my-app"),
    Type = "my-event",
    Data = new ByteStringContent(System.Text.Encoding.UTF8.GetBytes("{\"key\": \"value\"}"), "application/json")
};
var response = await client.PublishEventAsync("my-topic", cloudEvent);
Console.WriteLine($"Published: {response.Status.Success}");

Example: Request-Reply

var cloudEvent = new CloudEvent
{
    Id = Guid.NewGuid().ToString(),
    Source = new Uri("urn:my-app"),
    Type = "my-command",
    Data = new ByteStringContent(System.Text.Encoding.UTF8.GetBytes("{\"command\": \"status\"}"), "application/json")
};
var response = await client.PublishWithResultAsync(
    topic: "worker-commands",
    cloudEvent: cloudEvent,
    replyTopic: "worker-responses",
    timeout: TimeSpan.FromSeconds(10)
);
Console.WriteLine($"Response: {Encoding.UTF8.GetString(response.CloudEvent.Data.ToByteArray())}");

using var sub = client.SubscribeToTopic("my-topic");
await foreach (var evt in sub.ResponseStream.ReadAllAsync())
{
    Console.WriteLine($"Event: {evt.Topic} - {Encoding.UTF8.GetString(evt.Data.ToByteArray())}");
}

Protobuf Dependencies

The client library requires the generated protobuf classes. Ensure your project references:

<PackageReference Include="Google.Protobuf" />
<PackageReference Include="Grpc.Net.Client" />
<PackageReference Include="Grpc.Tools" />

The proto files (gateway.proto, config.proto) must be compiled into: - Virtufin.Api.Protos.Gateway - Virtufin.Api.Protos.Config