Custom Ingest Modules

The ingest service uses a pluggable module architecture. You can create new modules to process custom MQTT message types and store data in MongoDB.

Module Interface

Each module follows this pattern:

class MyModule {
  constructor({ mqtt, mongo }) {
    this._mqtt = mqtt;
    this._mongoMgr = mongo;
  }

  load(settings = {}) {
    // Register a listener for a subtopic. OroMqtt subscribes to
    // r/+/<subtopic> and routes matching messages to the handler.
    this._mqtt.registerListener('my_custom_topic', this.handleMessage);
  }

  handleMessage = (robotId, msg, packet) => {
    // 1. Decode the protobuf (or raw) message
    // 2. Transform and validate data
    // 3. Write to MongoDB
  };
}

export default MyModule;

Step-by-Step Guide

1. Define Your Protobuf Message (Optional)

If your module uses protobuf, add the message definition to app/private/oro.proto — the only git-tracked copy of the schema. ingest/import.sh (run by ingest/run.sh on every start) copies it into ingest/src/shared/, which is generated and gitignored — edits there are silently overwritten.

message MyCustomMessage {
  string sensorId = 1;
  float value = 2;
  int64 timestamp = 3;
}

There is nothing to regenerate: the proto file is loaded at runtime.

2. Create the Module

Create a new file at ingest/src/server/modules/myModule.js:

export default class MyModule {
  constructor({ mqtt, mongo }) {
    this._mqtt = mqtt;
    this._mongoMgr = mongo;
  }

  load = (settings = {}) => {
    // Register for r/+/my_custom_topic; OroMqtt owns the subscription
    this._mqtt.registerListener('my_custom_topic', this.handleMessage);

    // Look up the message type from the shared protobuf root.
    // Note the mandatory 'oro.' package prefix.
    this._myCustomMessage = this._mqtt.lookupType('oro.MyCustomMessage');

    // Get the MongoDB collection through the manager
    this._collection = this._mongoMgr.getCollection('my_custom_data');
    return this;
  };

  handleMessage = async (robotId, msg, _packet) => {
    try {
      // Decode protobuf
      const message = this._myCustomMessage.decode(msg);

      // Write to MongoDB
      await this._collection.updateOne(
        { _id: robotId },
        {
          $set: {
            sensorId: message.sensorId,
            value: message.value,
            updatedAt: new Date(message.timestamp.toNumber()),
          },
        },
        { upsert: true }
      );
    } catch (err) {
      console.error(`MyModule error for robot ${robotId}:`, err);
    }
  };
}

See ingest/src/server/modules/events.js (RobotEventsModule) for a compact real-world example of exactly this pattern.

3. Register the Module

Export the module from ingest/src/server/modules/index.js:

export { default as MyModule } from './myModule';

Then load it in ingest/src/main.js:

import { MyModule, /* other modules */ } from './server/modules';

// In the run() function:
new MyModule({ mqtt, mongo }).load(moduleSettings.myModule);

Some older modules take positional constructor arguments (e.g. new BasicsModule(mqtt)); prefer the options-object form shown above for new modules — it is what CustomDataModule, RobotEventsModule, and RobotLocalizationModule use.

4. Configure Module Settings (Optional)

Per-module settings live under the modules key of ingest/settings.json:

{
  "modules": {
    "myModule": {
      "enabled": true,
      "customSetting": "value"
    }
  }
}

warning

ingest/settings.json is generated by Terraform (scripts/generate-settings.sh) and gitignored — manual edits are lost on the next run. Add persistent settings through terraform/main.tf / terraform/local.tfvars instead.

5. Make the Agent Publish

Data only arrives if the corresponding agent-side module is enabled at the robot's current run level. Module run states (including minRunlevel) are managed per robot via the ConfigAPI ModuleState kind and the web app's agent manager — make sure the agentlet that publishes your topic is configured to run.

MQTT Topic Conventions

• Topics follow the pattern r/<robot_id>/<message_type>
• Use + wildcard to match any robot ID: r/+/my_topic
• The MQTT client extracts robotId from the topic and passes it to your handler

Best Practices

• Upsert pattern — use updateOne with upsert: true for idempotent writes
• Error handling — catch and log errors in message handlers; don't let one bad message crash the module
• Graceful degradation — check for null/undefined fields in protobuf messages
• Module settings — accept a settings parameter in load() for configurability

Existing Modules for Reference

Module	File	What it does
BasicsModule	modules/basics.js	Robot identity and status
SystemModule	modules/system.js	System resource metrics
CustomDataModule	modules/customData.js	Arbitrary key-value data
RobotEventsModule	modules/events.js	Sampled key-value events (cleanest template)
DiagnosticsModule	modules/diagnostics.js	ROS diagnostics
CustomCommandsModule	modules/customCommands.js	Command execution feedback
RobotLocalizationModule	modules/localization.js	Pose, maps, lasers, paths
UpstreamModule	modules/upstream.js	Upstream telemetry forwarding

Next Steps

• Ingest Pipeline — architecture deep dive
• MQTT & Protocols — topic structure and protobuf messages
• Data Model — MongoDB collections