Streaming Events

An example of how to use the REST API to stream sensor events in real-time.

Overview

There are many applications where having a continuous stream of data is preferred over polling chunks periodically, like plotting life graphs or triggering certain alarms. In this example, we will see how the REST API can be used to set up a stream with some best-practices and a simple retry-policy.

Preliminaries

  • Data Connectors If you want to forward your data in a server-to-server integration, consider using Data Connectors for a simpler and more reliable service with an added at-least-once guarantee.

  • Basic Auth For simplicity, we here use Basic Auth for authentication. We recommend replacing this with an OAuth2 flow for integrations more complex than local experimentation.

  • Service Account Credentials You need to have created and know the credentials of a Service Account. Any role will suffice.

Stream API

To stream events from one or several devices, the following endpoint will be used.

https://api.disruptive-technologies.com/v2/projects/{project}/devices:stream

For full details on the endpoint, see the REST API Reference.

Parameters

The following parameters are useful to know about for streaming events.

  • device_ids A list of device identifiers to limit the query to a set of specific devices.

  • device_types Filter streaming events to one or several device types only.

  • event_types Filter streaming events to one or several types of events only.

  • label_filters

    Filter streaming events to one or more labels. You can either specify just a label key (labelKey) or a key-value pair (labelKey=labelValue).

Response Format

The :stream endpoints support two types of response formats, text/event-stream and application/json. They are set using the Accept header, and defaults to application/json.

headers = { "Accept": "application/json" } # default
headers = { "Accept": "text/event-stream" } # alternative

The different headers are used in different cases.

  • text/event-stream: A Server-Sent Events specific format. Used by any Server-Sent Events libraries.

  • application/json: Returns a JSON object for each event, separated by line-breaks. Easy to parse in any high-level language, and used in the following example.

Stream Best Practices

The following practices should be considered and are present in the following example code.

  • Dropped Connections The connection to a stream can be lost at any moment due to a number of factors that break the connection and should be handled by implementing some kind of retry-policy.

  • Query Parameters Use the various query parameters to reduce the number of events that need to be processed.

  • Authorization Header Use the Authorization header if possible. Otherwise, such as when using the EventSource library in JavaScript, the token query parameter can be used.

  • Pinging An optional pingInterval query parameter may be used to make sure the client is still able to receive messages from the server. If no ping messages have been received in the specified interval, the client should reconnect to the stream.

Example Code

The provided example code can be summarized by the following points.

  • Sends a GET request to the REST API to initialize an event stream.

  • Keep the TCP connection open while receiving events.

  • If the connection is lost or it's been too long between pings, retry N times before giving up.

Environment Setup

If you wish to run the code locally, make sure you have a working runtime environment.

Python 3.9
Python API (Beta)
Node.js 14
Python 3.9

The following packages are required by the example code and must be installed.

pip install requests==2.25.1
Python API (Beta)

The latest version of our Python API can be installed through pip.

pip install --upgrade disruptive
Node.js 14

The following modules are required by the example code and must be installed.

npm install [email protected]
npm install [email protected]
npm install [email protected]

Add environment variables for authentication details.

export DT_SERVICE_ACCOUNT_KEY_ID=<YOUR_SERVICE_ACCOUNT_KEY_ID> # [string]
export DT_SERVICE_ACCOUNT_SECRET=<YOUR_SERVICE_ACCOUNT_SECRET> # [string]
export DT_SERVICE_ACCOUNT_EMAIL=<YOUR_SERVICE_ACCOUNT_EMAIL> # [string]
export DT_PROJECT_ID=<YOUR_PROJECT_ID> # [string]

Source

The following code snippet implements streaming in a few languages.

Python 3.9
Python API (Beta)
Node.js 14
Python 3.9
import os
import time
import json
import requests # pip install requests==2.25.1
# Service Account credentials.
SERVICE_ACCOUNT_KEY_ID = os.environ.get('DT_SERVICE_ACCOUNT_KEY_ID')
SERVICE_ACCOUNT_SECRET = os.environ.get('DT_SERVICE_ACCOUNT_SECRET')
# Construct API URL.
PROJECT_ID = os.environ.get('DT_PROJECT_ID')
BASE_URL = 'https://api.d21s.com/v2'
DEVICES_STREAM_URL = '{}/projects/{}/devices:stream'.format(
BASE_URL,
PROJECT_ID
)
MAX_CONNECTION_RETRIES = 5
PING_INTERVAL = 10
PING_JITTER = 2
if __name__ == '__main__':
# Set up a simple catch-all retry policy.
nth_retry = 0
while nth_retry <= MAX_CONNECTION_RETRIES:
try:
print('Streaming... Press CTRL+C to terminate.')
# Set up a stream connection.
# Connection will timeout and reconnect if no single event
# is received in an interval of PING_INTERVAL + PING_JITTER.
stream = requests.get(
url=DEVICES_STREAM_URL,
auth=(SERVICE_ACCOUNT_KEY_ID, SERVICE_ACCOUNT_SECRET),
stream=True,
timeout=PING_INTERVAL + PING_JITTER,
params={
'event_types': ['temperature'],
'ping_interval': '{}s'.format(PING_INTERVAL),
},
)
# Iterate through the events as they come in (one per line).
for line in stream.iter_lines():
# Decode the response payload and break on error.
payload = json.loads(line.decode('ascii'))
if 'result' not in payload:
print(payload)
break
event = payload['result']['event']
# Skip ping events.
if event['eventType'] == 'ping':
continue
# Print temperature events.
if event['eventType'] == 'temperature':
temperature = event['data']['temperature']
print('\t- {:<5}°C from {} at {}'.format(
temperature['value'],
event['targetName'].split('/')[-1],
temperature['updateTime']
))
# Reset retry counter.
nth_retry = 0
except KeyboardInterrupt:
break
except requests.exceptions.ConnectionError:
print('TIMEOUT. Reconnecting...')
nth_retry += 1
except Exception:
# Print the error and try again up to MAX_CONNECTION_RETRIES.
if nth_retry < MAX_CONNECTION_RETRIES:
print('Connection lost. Retry #{}'.format(nth_retry+1))
# Exponential backoff in sleep time.
time.sleep(2**nth_retry)
nth_retry += 1
else:
break
Python API (Beta)
import os
import disruptive as dt
# Environment variables for authentication credentials and target.
KEY_ID = os.environ.get('DT_SERVICE_ACCOUNT_KEY_ID')
SECRET = os.environ.get('DT_SERVICE_ACCOUNT_SECRET')
EMAIL = os.environ.get('DT_SERVICE_ACCOUNT_EMAIL')
PROJECT_ID = os.environ.get('DT_PROJECT_ID')
if __name__ == '__main__':
# Authenticate the package using Service Account credentials.
dt.default_auth = dt.Auth.service_account(KEY_ID, SECRET, EMAIL)
# Enable logging to see what's happening under the hood.
dt.log_level = 'debug'
# Initialize a stream generator for all temperature events in a project.
for e in dt.Stream.event_stream(PROJECT_ID, event_types=['temperature']):
# Print event information.
print('- {:<5}°C from {} at {}'.format(
e.data.celsius,
e.device_id,
e.data.timestamp,
))
Node.js 14
const EventSource = require("eventsource") // npm install [email protected]
const jwt = require('jsonwebtoken') // npm install [email protected]
const axios = require('axios').default // npm install [email protected]
// Service Account credentials
const serviceAccountKeyID = process.env.DT_SERVICE_ACCOUNT_KEY_ID
const serviceAccountSecret = process.env.DT_SERVICE_ACCOUNT_SECRET
const serviceAccountEmail = process.env.DT_SERVICE_ACCOUNT_EMAIL
// Construct API URL
const projectID = process.env.DT_PROJECT_ID
const apiBase = 'https://api.d21s.com/v2/'
const devicesStreamUrl = apiBase + `projects/${projectID}/devices:stream`
// Constants
const maxConnectionRetries = 3 // Max retries without any received messages
const pingInterval = 10 // Expected interval between pings in seconds
const pingJitter = 2000 // Expected ping jitter in milliseconds
async function main() {
let retryCount = 0
let stream
setupStream()
// Sets up a timer that will restart the stream if there has passed too much
// time between ping events. This timer is reset every time we receive a ping.
const pingTimer = setTimeout(() => {
console.log("Too long between pings. Reconnecting...")
setupStream()
}, pingInterval * 1000 + pingJitter)
async function setupStream() {
// If we've retried too many times without getting any messages, exit
if (retryCount >= maxConnectionRetries) {
console.log("Retried too many times. Exiting")
process.exit(1)
}
retryCount += 1
// Since EventSource does not support setting HTTP headers, we need to use
// OAuth for authentication. The received access token will be set as a
// `token` query parameter when setting up the stream in the next step.
const accessToken = await getAccessToken(
serviceAccountKeyID,
serviceAccountEmail,
serviceAccountSecret,
)
// Add query parameters to the URL
let url = devicesStreamUrl
url += `?eventTypes=temperature` // Filters temperature events
url += `&pingInterval=${pingInterval}s` // Specifies ping interval
url += `&token=${accessToken}` // Access token for authentication
// Close the existing stream if we have one
if (stream) {
stream.close()
}
// Set up a new stream with callback functions for
// messages and errors
console.log('Streaming... Press CTRL+C to exit.')
stream = new EventSource(url)
stream.onmessage = handleStreamMessage
stream.onerror = handleStreamError
}
function handleStreamError(err) {
console.error("Got error from stream:")
console.error(err)
console.log("Reconnecting...")
setupStream()
}
function handleStreamMessage(message) {
// Parse the payload as JSON
const data = JSON.parse(message.data)
if (!data || !data.result) {
return
}
// Reset the retry counter now that we got an event
retryCount = 0
// Parse the event object
const event = data.result.event
if (event.eventType === "ping") {
// We got a ping event. Reset the ping timer
pingTimer.refresh()
} else if (event.eventType === "temperature") {
// We got a temperature event
handleTemperatureEvent(event)
}
}
// Prints out the values of new incoming temperature events
function handleTemperatureEvent(event) {
const temp = event.data.temperature
const deviceID = event.targetName.split("/")[3]
console.log(`Got ${temp.value}°C from ${deviceID} at ${temp.updateTime}`)
}
}
main().catch((err) => {console.log(err)})
// Fetches an access token. See the following guide for documentation:
// https://developer.disruptive-technologies.com/docs/authentication/oauth2
async function getAccessToken(keyID, email, secret) {
// Construct the JWT header
let jwtHeaders = {
'alg': 'HS256',
'kid': keyID,
}
// Construct the JWT payload
let jwtPayload = {
'iat': Math.floor(Date.now() / 1000), // current unixtime
'exp': Math.floor(Date.now() / 1000) + 3600, // expiration unixtime
'aud': 'https://identity.disruptive-technologies.com/oauth2/token',
'iss': email,
}
// Sign and encode JWT with the secret
const jwtEncoded = jwt.sign(
jwtPayload,
secret,
{
header: jwtHeaders,
algorithm: 'HS256',
},
)
// Prepare POST request data
const requestObject = {
'assertion': jwtEncoded,
'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
}
// Converts the requestObject to a Form URL-Encoded string
const requestData = Object.keys(requestObject).map(function(key) {
return encodeURIComponent(key)+'='+encodeURIComponent(requestObject[key])
}).join('&')
// Exchange JWT for access token
const accessTokenResponse = await axios({
method: 'POST',
url: 'https://identity.disruptive-technologies.com/oauth2/token',
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
data: requestData,
}).catch(function (error) {
if (error.response) {
console.log(error.response.data)
}
throw error
})
// Return the access token in the request
return accessTokenResponse.data.access_token
}

Expected Output

For each new event in the stream, a line will be printed to stdout.

Streaming... Press CTRL+C to terminate.
- 21.85°C from bjei75vbluqg00dltkig at 2021-02-15T13:27:58.536000Z
- 22.25°C from bjejgi7bluqg00dlu0g0 at 2021-02-15T13:28:16.911000Z
- 21.75°C from bjei55fbluqg00dltjvg at 2021-02-15T13:28:33.965000Z
- 23.2 °C from bfuj75ho5b7g0093bbk0 at 2021-02-15T13:28:57.709000Z
- 21.5 °C from bjehnue7gpvg00cjo000 at 2021-02-15T13:29:06.409000Z
- 21 °C from bjei8rgpismg008hqdu0 at 2021-02-15T13:29:19.768000Z