Stream

Real-Time Data Stream API - Operations Guide

This article describes the real-time data stream feature, which provides near real-time access to device data. This can be used to replace the REST Search API approach which has inherent delays.

Overview

Two APIs are available for consuming the broker-backed data stream:

API	Use Case	Data Delivery
REST Stream API	Batch polling, periodic data retrieval	Returns batch of messages per request
WebSocket API	Continuous real-time streaming	Pushes individual messages as they arrive

Stream Retention: The stream holds approximately 1 hour of the latest messages.

---

1. REST API

Endpoint

GET https://[mycloud].ardexa.com/api/v1/tables/{workgroupId}/{tableName}/latest
GET https://[mycloud].ardexa.com/api/v1/tables/{workgroupId}/-/latest   # Wildcard "-": all tables

Authentication

• Standard API Token with Bearer authentication

• Required scope: search:read

• Additional scope for audit_logs table or wildcard access: org:audit

Authorization: Bearer <api_token>

Query Parameters

Parameter	Type	Required	Description
offset	BigInt string	No	Resume from a specific offset (returned by previous call). Default: the start of the stream (which is about 1h ago)
timestamp	ISO string or Unix ms	No	Start from a specific timestamp (server-side append time)
limit	Integer	No	Max messages to return (default & max: 10,000)

Offset Priority: If both offset and timestamp are provided, offset takes precedence.

Response Format

Content-Type: application/cbor

{
  result: [
    {
      kind: "device",
      metadata: {
        device: "deviceID",
        table: "tableName",
        source: "sourceName",
        event_time: "2025-01-15T10:30:00.000Z",  // RFC 3339/ISO 8601
        source_time: "2025-01-15T10:29:59.500Z"  // RFC 3339/ISO 8601
      },
      data: {
        // Actual telemetry data collected by the agent
      }
    },
    // ... more messages
  ],
  offset: "123456789012345",  // BigInt as string - SAVE THIS
  error: "error message"      // Only present on error
}

Integration Notes

1. Offset Management: Customers must persist the offset value returned by each request. Use this offset in the next request to retrieve subsequent messages without gaps or duplicates.

2. Idle Timeout: The endpoint has a 1-second idle timeout. If no new messages arrive within 1 second, the response is returned with available data.

3. Initial Request Options:

   1. No params: Returns messages from the beginning of the stream (~1 hour ago)

   2. With timestamp: Returns messages from the specified time forward

   3. With offset: Resumes from exact position (recommended for ongoing polling)

---

2. WebSocket API

Connection

Connect to the standard WebSocket endpoint with authentication.

Events

Subscribe to Stream

Emit: stream:latest

{
  table: "tableName",    // Use "-" for wildcard (all tables)
  requestId: "UUID", // REQUIRED - client-generated identifier
  offset: "123456789",   // Optional: resume from offset. Default: the latest data (next offset to be written)
  timestamp: "2025-01-15T10:00:00Z",  // Optional: start from timestamp
  includeDeviceEvents: true, // Optional: include device online/offline data in the stream. Default: false
}

Receive Data

Listen: stream:latest:data

Telemetry data:

{
  result: {
    kind: "device",
    metadata: {
      device: "deviceID",
      table: "tableName",
      source: "sourceName",
      event_time: "2025-01-15T10:30:00.000Z",
      source_time: "2025-01-15T10:29:59.500Z"
    },
    data: {
      // Actual telemetry data
    }
  },
  offset: "123456789012345",
  requestId: "unique-id",
  streamId: "uuid"        // Server-assigned stream identifier
}

Device online/offline data:

{
  result: {
      kind: "agentState",
      metadata: {
        device: "deviceID",
        event_time: "2025-01-15T10:30:00.000Z"
      },
      online: true           // true: device goes online, false: device goes offline
    },
    offset: "123456789012345",
    requestId: "unique-id",
    streamId: "uuid"        // Server-assigned stream identifier
}

Stop Stream

Emit: stream:latest:stop

{
  streamId: "uuid"  // The streamId received in data events
}

Authentication

• Same scopes as REST API: search:read, plus org:audit for audit logs

• Authentication is handled via the WebSocket connection handshake

Integration Notes

1. requestId is mandatory: The server will reject requests without a requestId.

2. Default Behavior: Without offset or timestamp, the WebSocket starts from Offset.next() (newest messages only, no historical data).

3. Stream Lifecycle: Always call stream:latest:stop when done to free server resources.

---

3. Data Encoding

Both APIs return data encoded in CBOR (Concise Binary Object Representation).

Decoding Examples

JavaScript (Node.js):

import { decode } from 'cbor-x';

// REST API
const response = await fetch(url, { headers });
const buffer = await response.arrayBuffer();
const data = decode(new Uint8Array(buffer));

// WebSocket
socket.on('stream:latest:data', (buffer) => {
  const data = decode(buffer);
});

Python:

import cbor2

# REST API
response = requests.get(url, headers=headers)
data = cbor2.loads(response.content)

---

4. Integration Considerations

Stream Health & Availability

Concern	Details
Retention Window	~1 hour of messages. Data older than this is not available via stream
Message Ordering	Messages are ordered by offset. Use offset for reliable sequencing
Duplicates	Possible on reconnection if offset not properly managed

Recommended Practices

1. Persist Offsets Reliably

2. Store the last processed offset in durable storage (database, file)

3. Update offset after successfully processing each batch/message

4. On restart, resume from the stored offset

5. Handle Gaps Gracefully

6. If stored offset is older than 1 hour, it may be invalid

7. Implement fallback: start from timestamp or beginning of stream

8. Log and alert when gaps are detected

9. Error Handling

10. REST: Check for error field in response

11. WebSocket: Listen for error messages and handle reconnection

12. Implement exponential backoff for retries

13. Rate Limiting

14. REST: Respect the 10,000 message limit per request

15. Poll at reasonable intervals (recommended: 1-5 seconds)

16. WebSocket: Implement backpressure if processing can't keep up

17. Device Group Filtering

18. If the API token has device group restrictions, only data from permitted devices is returned

19. This filtering is automatic based on token permissions

Migration from REST Search API

Aspect	REST Search API	WebSocket Stream API
Latency	Minutes delay	Near real-time
Data Window	Historical archive	~1 hour rolling
Query Flexibility	Full search/aggregation	Sequential stream only
Use Case	Analytics, reporting	Live monitoring, alerting

Recommendation: Use Stream API for real-time needs; retain REST Search API integration for historical analysis and complex queries.

---

5. Quick Reference

REST API Checklist

• [ ] API token with search:read scope

• [ ] CBOR decoder in client

• [ ] Offset persistence mechanism

• [ ] Error handling

• [ ] Polling interval configured

WebSocket API Checklist

• [ ] Socket authentication configured

• [ ] UUID requestId generator

• [ ] CBOR decoder for binary messages

• [ ] stream:latest:stop called on disconnect

• [ ] Reconnection logic with offset resume