Couriers

Reference for OctoMY™'s courier system that handles message routing, data serialization, and application-layer communication.

Did You Know?

Couriers use different reliability modes for different data types. Sensor data uses unreliable (UDP-style) delivery because stale readings are useless - you want the latest data, not retransmitted old data. But Plan transfers use reliable delivery because you need every byte to arrive correctly. This hybrid approach gives OctoMY™ the best of both worlds.

Courier overview

Couriers are the application-layer message handlers in OctoMY™:

Courier architecture

Base class

class Courier : public QObject {
    Q_OBJECT

public:
    explicit Courier(QObject* parent = nullptr);
    virtual ~Courier();

    // Identification
    virtual QString name() const = 0;
    virtual quint8 id() const = 0;

    // Message handling
    virtual void onReceive(const QByteArray& data) = 0;
    virtual QByteArray serialize() const = 0;

    // Lifecycle
    virtual void activate();
    virtual void deactivate();

    // Reliability
    bool isReliable() const;
    void setReliable(bool reliable);

    // Timing
    quint32 sendInterval() const;
    void setSendInterval(quint32 ms);

signals:
    void dataReady();
    void error(const QString& message);

protected:
    void scheduleSend();
    CommsSession* session() const;
};

Courier mandate

struct CourierMandate {
    quint8 courierId;
    bool reliable;
    quint32 sendIntervalMs;
    quint32 priority;
    bool active;
};

Built-in couriers

System couriers (0-15)

Application couriers (16-127)

AgentStateCourier

Synchronizes node state between peers.

Message format

struct AgentStateMessage {
    quint64 timestamp;
    quint8 flags;
    quint8 mode;           // AgentMode enum
    quint8 batteryLevel;   // 0-100%
    qint8 signalStrength;  // dBm
    quint16 activePlans;   // Bitmask
    quint16 errorFlags;    // Error bitmask
};

Usage

AgentStateCourier* courier = new AgentStateCourier(session);

// Subscribe to state updates
connect(courier, &AgentStateCourier::stateReceived,
        [](const AgentStateMessage& state) {
    qDebug() << "Remote mode:" << state.mode;
    qDebug() << "Battery:" << state.batteryLevel << "%";
});

// Send state update
courier->sendState(currentState);

Flags

SensorsCourier

Streams sensor data from Agent to Remote.

Message format

struct SensorsMessage {
    quint64 timestamp;
    quint8 sensorCount;
    SensorReading readings[];  // Variable length
};

struct SensorReading {
    quint8 sensorId;
    quint8 type;        // SensorType enum
    quint16 value;      // Scaled value
    quint8 quality;     // 0-100 signal quality
};

Sensor types

Usage

SensorsCourier* courier = new SensorsCourier(session);
courier->setSendInterval(50);  // 20 Hz

// Receive sensor data
connect(courier, &SensorsCourier::sensorsReceived,
        [](const SensorsMessage& msg) {
    for (int i = 0; i < msg.sensorCount; i++) {
        auto& reading = msg.readings[i];
        processSensor(reading.sensorId, reading.value);
    }
});

// Send sensor data (on Agent)
SensorsMessage msg;
msg.timestamp = currentTimeMs();
msg.sensorCount = activeSensors.count();
// ... populate readings
courier->send(msg);

ActuatorsCourier

Sends actuator commands from Remote to Agent.

Message format

struct ActuatorsMessage {
    quint64 timestamp;
    quint8 actuatorCount;
    ActuatorCommand commands[];
};

struct ActuatorCommand {
    quint8 actuatorId;
    quint8 type;        // ActuatorType enum
    qint16 value;       // Target value
    quint8 flags;       // Control flags
};

Actuator types

Usage

ActuatorsCourier* courier = new ActuatorsCourier(session);

// Send motor commands
ActuatorsMessage msg;
msg.timestamp = currentTimeMs();
msg.actuatorCount = 2;
msg.commands[0] = {0, DC_MOTOR, 60, 0};   // Left motor 60%
msg.commands[1] = {1, DC_MOTOR, 60, 0};   // Right motor 60%
courier->send(msg);

// Receive commands (on Agent)
connect(courier, &ActuatorsCourier::commandReceived,
        [](const ActuatorsMessage& msg) {
    for (int i = 0; i < msg.actuatorCount; i++) {
        applyActuator(msg.commands[i]);
    }
});

BlobCourier

Handles large data transfers with chunking and reliability.

Blob types

Transfer protocol

Usage

BlobCourier* courier = new BlobCourier(session);

// Send blob
QByteArray planData = loadPlan("patrol.opal");
BlobFuture* future = courier->sendBlob(PLAN, planData);

connect(future, &BlobFuture::progress, [](float percent) {
    updateProgress(percent);
});

connect(future, &BlobFuture::complete, []() {
    qDebug() << "Plan sent successfully";
});

connect(future, &BlobFuture::failed, [](const QString& error) {
    qDebug() << "Transfer failed:" << error;
});

// Receive blob
connect(courier, &BlobCourier::blobReceived,
        [](BlobType type, const QByteArray& data) {
    if (type == PLAN) {
        installPlan(data);
    }
});

DiscoveryCourier

Handles peer discovery on local network.

Discovery protocol

Usage

DiscoveryCourier* courier = new DiscoveryCourier(channel);

// Start discovery
courier->startDiscovery();

// Handle discovered nodes
connect(courier, &DiscoveryCourier::nodeDiscovered,
        [](const NodeInfo& info) {
    qDebug() << "Found:" << info.name;
    qDebug() << "Type:" << info.nodeType;
    qDebug() << "Address:" << info.address;
});

// Announce ourselves
courier->announce();

CourierSet

Manages a collection of couriers for a node type.

Node-specific courier sets

// Agent side
class AgentCourierSet : public CourierSet {
public:
    SensorsCourier* sensors;      // Sends sensor data
    ActuatorsCourier* actuators;  // Receives commands
    AgentStateCourier* state;     // Sends state
    BlobCourier* blob;            // Receive plans
};

// Remote side
class RemoteCourierSet : public CourierSet {
public:
    SensorsCourier* sensors;      // Receives sensor data
    ActuatorsCourier* actuators;  // Sends commands
    AgentStateCourier* state;     // Receives state
    BlobCourier* blob;            // Send plans
};

Usage

// Create courier set for session
AgentCourierSet* couriers = new AgentCourierSet(session);

// Activate all couriers
couriers->activate();

// Access individual couriers
couriers->sensors->setSendInterval(20);
couriers->state->setSendInterval(100);

// Deactivate when done
couriers->deactivate();

Custom couriers

Creating a custom courier

class MyCourier : public Courier {
    Q_OBJECT

public:
    explicit MyCourier(QObject* parent = nullptr)
        : Courier(parent) {}

    QString name() const override {
        return "MyCourier";
    }

    quint8 id() const override {
        return 128;  // Custom courier ID range
    }

    void onReceive(const QByteArray& data) override {
        // Parse incoming data
        MyMessage msg;
        QDataStream stream(data);
        stream >> msg.field1 >> msg.field2;

        emit messageReceived(msg);
    }

    QByteArray serialize() const override {
        // Serialize outgoing data
        QByteArray data;
        QDataStream stream(&data, QIODevice::WriteOnly);
        stream << mCurrentMessage.field1
               << mCurrentMessage.field2;
        return data;
    }

    void sendMessage(const MyMessage& msg) {
        mCurrentMessage = msg;
        scheduleSend();
    }

signals:
    void messageReceived(const MyMessage& msg);

private:
    MyMessage mCurrentMessage;
};

Registering custom courier

// Register with session
session->registerCourier(myCourier);

// Or add to courier set
class MyCourierSet : public CourierSet {
public:
    MyCourierSet(CommsSession* session)
        : CourierSet(session)
    {
        myCourier = new MyCourier(this);
        addCourier(myCourier);
    }

    MyCourier* myCourier;
};

Courier timing

Send scheduling

Priority and batching

// Couriers are sent in priority order
courier->setPriority(10);  // Higher = first

// Multiple couriers batched in one packet when possible
session->setBatchingEnabled(true);
session->setMaxBatchDelay(5);  // Max wait for batching

Error handling

Courier errors

Error handling

connect(courier, &Courier::error,
        [](const QString& error) {
    qWarning() << "Courier error:" << error;
    // Handle error appropriately
});

// In custom courier
try {
    processData(data);
} catch (const std::exception& e) {
    emit error(QString("Processing failed: %1").arg(e.what()));
}

Debugging couriers

Courier statistics

struct CourierStats {
    quint64 messagesSent;
    quint64 messagesReceived;
    quint64 bytesSent;
    quint64 bytesReceived;
    quint32 errorCount;
    quint32 averageLatencyMs;
};

CourierStats stats = courier->statistics();
qDebug() << "Messages sent:" << stats.messagesSent;
qDebug() << "Average latency:" << stats.averageLatencyMs << "ms";

Debug logging

// Enable courier debug logging
QLoggingCategory::setFilterRules("octomy.courier.debug=true");

// Or per-courier
courier->setDebugEnabled(true);