FleetMQ 1.0.45

FleetMQ C# SDK

The FleetMQ C# SDK lets your application send and receive data through the FleetMQ Node running on the same device. Communication uses ZMQ sockets — the SDK handles socket setup and port negotiation automatically.

Installation

Install the SDK from NuGet:

dotnet add package FleetMQSDK

Or in your .csproj:

<PackageReference Include="FleetMQSDK" Version="*" />

Prerequisites

The FleetMQ Node must be running on the device before your application starts. The SDK connects to it on localhost at construction time.

Quick Start

using FleetMQSDK;

using var fleetMQ = FleetMQ.Connect();

fleetMQ.Publish("my-topic", "Hello World!");
string msg = fleetMQ.Receive("my-topic");

Initialization

FleetMQ.Connect()

Creates a FleetMQ client, connects to the FleetMQ Node, fetches the device config, and sets up all publishers and subscribers defined in that config. Blocks until the config is received. All parameters are optional.

using var fleetMQ = FleetMQ.Connect();

Full signature:

FleetMQ.Connect(
    string logPort       = "5550",  // Port for forwarding SDK logs to the FleetMQ Node
    string reqPort       = "5558",  // RPC port for config and publisher/subscriber provisioning
    string pullPort      = "5557",  // Push/pull port for config and metrics delivery
    int    publishRateHz = 200      // Max publish rate in Hz, clamped to 20–200
)

reqPort and pullPort do not normally need to be changed. Only set them if you have a port conflict on your machine — if you do, the corresponding port flags on the FleetMQ Node must be set to match.

FleetMQ implements IDisposable — use using to ensure sockets are cleaned up automatically.

Publishing

Publish(string topic, string message)

Publishes a string message to the given topic. Throws if the publisher does not exist.

fleetMQ.Publish("vehicle/telemetry", "speed=42.5");

PublishBytes(string topic, byte[] message)

Publishes raw bytes to the given topic. Throws if the publisher does not exist.

byte[] payload = BitConverter.GetBytes(42.5f);
fleetMQ.PublishBytes("vehicle/sensors", payload);

Publisher behaviour: Publishers buffer the latest message and send at the configured rate (publishRateHz). If you call Publish or PublishBytes faster than the rate allows, only the most recent value is sent — older values are dropped. This is intentional for sensor/telemetry data where only the latest reading matters.

CreatePublisher(string topic, bool ipc = false)

Manually provisions a publisher for a topic via RPC to the FleetMQ Node. Only needed for topics not in the device config. If a publisher for this topic already exists, this is a no-op.

fleetMQ.CreatePublisher("vehicle/telemetry");

Subscribing

Receive(string topic)

Blocks until the next string message is available for the topic. Throws if the subscriber does not exist.

string msg = fleetMQ.Receive("operator/commands");
Console.WriteLine($"Received: {msg}");

ReceiveBytes(string topic)

Blocks until the next message is available, returning it as byte[]. Throws if the subscriber does not exist.

byte[] raw = fleetMQ.ReceiveBytes("vehicle/sensors");

CreateSubscriber(string topic, bool ipc = false)

Manually provisions a subscriber for a topic via RPC to the FleetMQ Node. Only needed for topics not in the device config. If a subscriber for this topic already exists, this is a no-op.

fleetMQ.CreateSubscriber("operator/commands");

Metrics and Events

PullMetricsAndEvents()

Returns (List<FleetMQMetric> metrics, FleetMQEvent? eventObj)? — null if nothing is available, or a tuple containing a list of metrics and an optional event. Call this periodically to monitor connection quality and adapt your application's behaviour. Metrics include one-way latency between your device and each peer, which your SDK can use to make real-time decisions — for example, reducing publish frequency, switching to a lower-fidelity data stream, or surfacing connection quality indicators in your UI.

var result = fleetMQ.PullMetricsAndEvents();
if (result.HasValue)
{
    var (metrics, eventObj) = result.Value;

    foreach (var metric in metrics)
        Console.WriteLine($"{metric.Type}: {metric.Value} (peer={metric.Peer}, ts={metric.Timestamp})");

    if (eventObj != null)
        Console.WriteLine($"Event: {eventObj.Type} = {eventObj.Value}");
}

FleetMQMetric properties

FleetMQEvent properties

FleetMQLatencyGateEvent (subtype of FleetMQEvent)

Cast to this type when Type == "latency_gate":

if (eventObj is FleetMQLatencyGateEvent lg)
    Console.WriteLine($"Gated={lg.Value}, Timestamp={lg.Timestamp}");

Latency gate behaviour: A latency gate event is fired when the round-trip latency between your device and a peer crosses a configured threshold. When Gated is true, the FleetMQ Node has paused data flow on that connection to protect against sending stale data over a degraded link. When Gated is false, the connection has recovered and data flow has resumed. Use these events to update your UI or trigger application-level responses — for example, alerting the operator that the connection is degraded or suppressing controls that depend on low-latency feedback.

FleetMQVideoQualityEvent (subtype of FleetMQEvent)

Cast to this type when Type == "video_quality". The FleetMQ Node continuously monitors network conditions and adjusts the outgoing video stream to match available bandwidth. This event is fired each time the quality tier changes.

if (eventObj is FleetMQVideoQualityEvent vq)
{
    dynamic val = vq.Value;
    Console.WriteLine($"Video quality: {val.tier} (cause: {val.triggeringMetric})");
    if (val.tier == "CRITICAL")
        Console.WriteLine("Warning: severe network degradation, video resolution reduced");
}

Tier values:

Cleanup

Dispose()

Cancels the publisher background thread, closes all sockets, and releases resources. Called automatically when using using.

fleetMQ.Dispose();
// or just: using var fleetMQ = FleetMQ.Connect();

Full Example

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using FleetMQSDK;

using var fleetMQ = FleetMQ.Connect();
Console.WriteLine("Connected to FleetMQ!");

var cts = new CancellationTokenSource();

var tasks = new List<Task>
{
    Task.Run(() => Publish(fleetMQ, "vehicle/telemetry", cts.Token)),
    Task.Run(() => Subscribe(fleetMQ, "operator/commands", cts.Token)),
    Task.Run(() => PollMetrics(fleetMQ, cts.Token)),
};

Console.ReadLine();
cts.Cancel();
await Task.WhenAll(tasks);

static async Task Publish(FleetMQ fleetMQ, string topic, CancellationToken token)
{
    while (!token.IsCancellationRequested)
    {
        fleetMQ.Publish(topic, "Hello World!");
        await Task.Delay(1000, token);
    }
}

static async Task Subscribe(FleetMQ fleetMQ, string topic, CancellationToken token)
{
    while (!token.IsCancellationRequested)
    {
        string msg = fleetMQ.Receive(topic); // blocks until a message arrives
        Console.WriteLine($"Received: {msg}");
    }
}

static async Task PollMetrics(FleetMQ fleetMQ, CancellationToken token)
{
    while (!token.IsCancellationRequested)
    {
        var result = fleetMQ.PullMetricsAndEvents();
        if (result.HasValue)
        {
            var (metrics, eventObj) = result.Value;
            foreach (var m in metrics)
                Console.WriteLine($"Metric: {m.Type} = {m.Value}");
            if (eventObj != null)
                Console.WriteLine($"Event: {eventObj.Type}");
        }
        await Task.Delay(500, token);
    }
}

Binary data example

using FleetMQSDK;

using var fleetMQ = FleetMQ.Connect();

// Publish bytes
byte[] payload = BitConverter.GetBytes(36.7f);
fleetMQ.PublishBytes("vehicle/sensors", payload);

// Receive bytes
byte[] raw = fleetMQ.ReceiveBytes("vehicle/sensors");
float value = BitConverter.ToSingle(raw, 0);
Console.WriteLine($"Temperature: {value}");

Notes

• Receive() and ReceiveBytes() block until a message arrives. Always run them on a dedicated thread or Task to avoid blocking the rest of your application.
• Topic names must exactly match what is configured in the FleetMQ dashboard. A mismatch will cause Publish, PublishBytes, Receive, or ReceiveBytes to throw.
• On Windows, the SDK automatically connects to tcp://host.docker.internal instead of localhost to support Docker-based deployments.