Stream services


A stream service is a software service that can be configured to handle requests continuously. The most common use of stream services within the ClearBlade Platform or Edge is to listen to MQTT topics and handle MQTT message requests from devices/users.

When to use stream services?

  • When users invoke a code service more than once a minute through a timer or trigger event

  • When users expect a real-time response since they reduce the time to invoke the code service

  • When users choose to perform parallel processing of incoming data, i.e., distribution of the incoming loads across multiple instances of the same stream service


Set execution timeout to never.

Template for stream services

Parallel execution using shared topic

Stream services are very efficient and can be designed to perform parallel processing. In this template, users can have ten stream service instances.

The template code below is written so that every stream service instance acts as a unique client to the shared MQTT group topic. Shared subscriptions allow for uniform load distribution across all the stream service instances. Refer to Shared subscriptions to understand how shared topics work.

function SharedTopicService(req, resp) { ClearBlade.init({ request: req }); var messaging = ClearBlade.Messaging(); var sharedTopic = "$share/Group1/device/+"; var deviceCollection = ClearBlade.Collection({ collectionName: "some_collection" }); // DEBUG MESSAGE messaging.publish("success", "Service Started"); messaging.subscribe(sharedTopic, function(err, errMsg) { if (err) { // DEBUG MESSAGE messaging.publish("error", "Subscribe failed: " + errMsg); resp.error(); } // DEBUG MESSAGE messaging.publish("success", "Subscribed to Shared Topic"); // Once successfully subscribed WaitLoop(); }); function WaitLoop() { // DEBUG MESSAGE messaging.publish("success", "Starting the Loop"); while (true) { messaging.waitForMessage([sharedTopic], function(err, msg, topic) { if (err) { // DEBUG MESSAGE messaging.publish("error", "Failed to wait for message: " + err + " " + msg + " " + topic); resp.error("Failed to wait for message: " + err + " " + msg + " " + topic); } else { // any action addCollectionRow(msg); } }); } } function addCollectionRow(msg) { try { var parseMsg = JSON.parse(msg); } catch (e) { // DEBUG MESSAGE messaging.publish("error","Problem with parsing: " + e); resp.error("Problem with parsing: " + e); } var data = { column_1_name: "column_1_data", column_2_name: "column_2_data" }; //debugging deviceCollection.create(data, function(err, result) { if (err) { // DEBUG MESSAGE messaging.publish("error", "failed to create: " + result); resp.error("create failed: " + result); } else { //no op } }); } }