Skip to content
IndraMQTT

Data Connectors

Data Connectors

The complete MQTT platform in one small binary: broker, SQL rules, data connectors and multi-tenancy in a single process.

List connector kinds and register one. Use it to attach a sink. You get ids, kinds and tiers from the rule engine registry.

flowchart LR
    Op["operator"] -- "POST /api/v1/connectors" --> API["management API"]
    API -- "register" --> C["connectors"]

Before you start

  • Start the broker from First start.
  • Set INDRA_API_KEY to an operator key. The examples read it.

Note: Calls except /healthz need an operator API key. Pass it as Authorization: Bearer $INDRA_API_KEY.

External data sinks (Kafka, Postgres, Redis, MySQL, S3)

GET /api/v1/connectors

List registered data connectors

Operation id: listConnectors.

Parameters:

No parameters.

Request body:

No request body.

Responses:

StatusDescription
200List of active connectors

Response 200 body:

Media type application/json:

{
  "type": "array",
  "items": {
    "type": "object",
    "properties": {
      "id": {
        "type": "string"
      },
      "kind": {
        "type": "string"
      },
      "tier": {
        "type": "string"
      }
    }
  }
}

Notes:

  • Related calls sit on the same tag page.

Verified: the example flow below runs this call.

Example:

# Illustrative request. Values come from the schema above.
# The example flow below runs verified calls on a live broker.
curl -s -X GET 'http://127.0.0.1:18083/api/v1/connectors' \
  -H "Authorization: Bearer $INDRA_API_KEY"

POST /api/v1/connectors

Register new data connector

Operation id: createConnector.

Parameters:

No parameters.

Request body:

Media type application/json:

Required: id, kind.

FieldTypeDescription
idstring
kindstring
configobject

Responses:

StatusDescription
201Connector registered
400Invalid connector

Notes:

  • Send id, kind. An empty required string fails with 400.

  • Success answers 201 with the stored row.

  • Related calls sit on the same tag page.

Verified: the example flow below runs this call.

Example:

# Illustrative request. Values come from the schema above.
# The example flow below runs verified calls on a live broker.
curl -s -X POST 'http://127.0.0.1:18083/api/v1/connectors' \
  -H "Authorization: Bearer $INDRA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
  "id": "doc-conn-1",
  "kind": "logger",
  "config": {}
}'

Example flow

Read real rows through the management API, then read them back. The docs harness runs this flow against a started broker.

Terminal window
python3 - <<'PY'
import json, os, urllib.request, urllib.error
port = os.environ.get("API_PORT", "18083")
key = os.environ.get("API_KEY", "")
base = f"http://127.0.0.1:{port}"
def call(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(base + path, data=data, method=method)
    req.add_header("Authorization", f"Bearer {key}")
    if body is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=5) as r:
            return r.status, r.read().decode()
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode()
s, b = call("POST", "/api/v1/connectors", {"id": "doc-conn-1", "kind": "console", "config": {}})
assert s == 201, b
assert "doc-conn-1" in b, b
s, b = call("GET", "/api/v1/connectors")
assert s == 200, b
assert "doc-conn-1" in b, b
print("OK")
PY
OK

Check that it works

Start the broker with this file. The health endpoint returns OK.

[node]
id = "doc-ref"


[listeners.tcp]
bind = "0.0.0.0:1883"


[listeners.tls]
bind = "0.0.0.0:8883"


[listeners.ws]
bind = "0.0.0.0:8083"


[listeners.wss]
bind = "0.0.0.0:8084"


[listeners.api]
bind = "0.0.0.0:18083"


[logging]
level = "warn"
Terminal window
python3 - <<'PY'
import os, socket
port = int(os.environ.get("API_PORT", "18083"))
sock = socket.create_connection(("127.0.0.1", port), timeout=5)
sock.sendall(b"GET /healthz HTTP/1.0\r\nHost: docs\r\n\r\n")
print(sock.makefile().read().strip().splitlines()[-1])
PY
OK

Check one documented call. GET /api/v1/connectors answers 200 with its rows. It prints OK.

Terminal window
python3 - <<'PY'
import os, urllib.request
port = os.environ.get("API_PORT", "18083")
key = os.environ.get("API_KEY", "")
base = f"http://127.0.0.1:{port}"
req = urllib.request.Request(base + "/api/v1/connectors", method="GET")
req.add_header("Authorization", f"Bearer {key}")
with urllib.request.urlopen(req, timeout=5) as r:
    body = r.read().decode()
    assert r.status == 200, body
assert body.strip().startswith("["), body
print("OK")
PY
OK

Next steps