C#
This document explains, in detail, how to interact with the NeuroScale Platform API using the C# client client_basic.cs from our public GitHub repository as an example (you can follow the code side-by-side with this documentation).
Installation¶
Before you start, you need to make sure that you have the right packages installed. Open Visual Studio. In NuGet Console window, type the following commands to install the necessary packages:
- install-package msgpack.cli
- Install-Package M2Mqtt -Version 4.2.0
- Install-Package Newtonsoft.Json
In order to use JavaScriptSerializer class, you also need to add a reference to System.Web.Extensions.dll which resides here:
C:\Program Files (x86)\Reference Assemblies\Microsoft\Framework\.NETFramework\v4.5\System.Web.Extensions.dll
To be able to use HttpClient class, you also need to add a reference to System.Net.Http.dll.
The first thing you need when connecting to NeuroScale is the URL at which the API lives. Besides the default endpoint (used in the below snippet), you may have received one or more endpoints for testing or for customer-specific features, which you can use here. The below default corresponds to the "production" platform. There are also staging and testing platforms to which you may be granted access if you are using cutting-edge pipelines.
The other piece of information you need is your personal access token. Access tokens are credentials specific to a customer and should be treated with the same secrecy as passwords for security reasons. They should never be committed to repositories, and should also not be entered verbatim as command-line arguments to prevent them from leaking into your shell history file. One way to manage this key is to read it from a file that only authorized users have access to, or to hold it in a system environment variable. Access tokens are usually strings of hexadecimal digits of the form 01234567-89ab-cdef-0123-456789abcdef.
Note¶
If you do not have an access token yet, check online documentation on how to generate one yourself using your username and password combination.
Instantiating a Pipeline¶
The API is designed according to REST principles, meaning that interaction takes place by performing HTTP requests with standard verbs (POST, GET, PATCH, DELETE) against certain URLs. The data that is being transmitted in the requests/responses represent objects encoded in JSON. All interactions need to be authorized by including an additional HTTP Authorization header whose value should be Bearer <your-access-token>:
// next we will look up the pipeline of interest by name from the list
// of available pipelines
HttpClient client = new HttpClient();
client.BaseAddress = new Uri(api_url);
// construct the HTTP header that will be used for authorization
client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", access_token);
// Add an Accept header for JSON format.
client.DefaultRequestHeaders.Accept.Add(
new MediaTypeWithQualityHeaderValue(@"application/json"));
Now we are ready to connect to the NeuroScale Platform API. NeuroScale provides a number of processing pipelines, which we can launch via the API and then stream data to/from a running pipeline instance. In this example, we use the "Echo" pipeline, which returns the upstream data without performing any processing (it does check your input for correct formatting, though):
The pipeline refers to a description of the processing pipeline, and not an actual running instance of it. The API then allows you to spawn an instance of a pipeline of your choice. It should be noted that spawning an instance is a costly operation on the backend (and also can take several dozen seconds), and to minimize churn it is best to reuse your running instances when running the same pipeline, or when you are testing and debugging your software. Once you have started an instance, you will get its unique identifier back, which is a string starting with "in_". You can then choose to destroy the instance after you are done with it, or alternatively you can reuse and reconfigure it if required. The example code allows you to set the name of an existing instance to connect to here (if empty, the example code will spawn a new instance):
Before we can instantiate the pipeline, we need to look up its global unique identifier (a string that starts with "pl_" followed by a hexadecimal string) from the list of available pipelines. One can obtain a list of the available pipelines by issuing an HTTP GET request to the \
// next we will look up the pipeline of interest by name from the list
// of available pipelines
HttpClient client = new HttpClient();
client.BaseAddress = new Uri(api_url);
// construct the HTTP header that will be used for authorization
client.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", access_token);
// Add an Accept header for JSON format.
client.DefaultRequestHeaders.Accept.Add(
new MediaTypeWithQualityHeaderValue(@"application/json"));
// Get data response.
HttpResponseMessage response = client.GetAsync(@"v1/pipelines").Result;
// Blocking call!
Next we should check the status code to ensure that the call went through; if either the API URL or authorization credentials were invalid, this call would fail. The status code corresponding to success is 200. Upon success, we can extract the JSON-encoded response body, which contains the list of pipelines in the "data" field of the resulting object:
string pipeline_id = string.Empty;
if (response.IsSuccessStatusCode)
{
// Parse the response body. Blocking!
string jsonContent = response.Content.ReadAsStringAsync().Result;
var jsSer = new JavaScriptSerializer();
var body = jsSer.Deserialize<Dictionary<string, dynamic>>(jsonContent);
var pipelines = body["data"];
The result is a list of pipelines with the human-readable names, descriptions, and unique resource identifiers, which we iterate over next to find the pipeline with the desired name:
// now go through all pipelines and pick the one that has a matching name
foreach (var p in pipelines)
{
if (p["name"] == pipeline_name)
{
pipeline_id = p["id"];
}
}
}
else
{
Console.WriteLine("{0} ({1}) \nERROR: Could not query available pipelines ", (int)response.StatusCode, response.ReasonPhrase);
}
Also, before we can launch this pipeline we need to describe the data that we want to stream up; this metadata is part of the launch configuration for the pipeline instance. First and foremost, a pipeline instance can accept one or more independent "data streams". Each stream is a multi-channel time series with a fixed or irregular sampling rate, and the channel values may be numbers (numeric time series) or strings (e.g. event markers or other data structures). Each of the streams has a name (no duplicates allowed), a content-type (for example, EEG, eye tracking, video, etc.) and a list of channel labels (also no duplicates allowed). In this example, we will send two streams; an 8-channel EEG time series and a stream of sporadic event markers. If you have no preference for your channel labels, it is customary to name them ch1 to chN:
var channel_labels = new List<string>() { "ch1", "ch2", "ch3", "ch4", "ch5", "ch6", "ch7", "ch8" };
// the sampling rate that we're giving here is the "nominal" sampling rate, but
// since all our data that we stream up is going to be time-stamped, the cloud
// system will be able to figure out the actual ("effective") sampling rate,
// which might be e.g., 99.8 Hz
int sampling_rate = 100;
The content-type here is named "modality" (as in measurement modality). While you can in principle use any string here, it is best to check first if there is a standard name for the modality that you are using because there may be nodes both now and in the future that can treat your data more intelligently if the modality is known. For naming we follow the conventions of the XDF multi-modal file format. Also, note that modality strings are case-sensitive.
Next we build the configuration data structure that will go into our JSON-encoded HTTP request from the bottom up, starting with the previously defined properties of the EEG stream object. In addition to the type, sampling rate and list of channel labels, each stream also has an arbitrary name (which must be unique among the streams that you send up if multiple):
var listOfChannels =
from d in channel_labels
select new Dictionary<string, string>() { { "label", d } };
var eeg_stream = new Dictionary<string, object>()
{
{"name", "myeeg"},
{"type", modality},
{"sampling_rate", sampling_rate},
{"channels", listOfChannels}
};
Next we define a marker stream. A marker stream is not required for many simple use cases where all that is being streamed up is a continuous time series. However, in cases where there are particular events happening at various times (e.g. button pressed, image presented, etc.) and when data needs to be analyzed in correspondence with these events, you need to have a marker stream. To identify a stream as holding markers, its type must be set to "Markers" for the pipeline instance to recognize it. In general, a sampling rate of 0 indicates that the data stream is irregularly sampled; besides markers, this also allows you to stream up irregularly sampled time series. A marker stream usually has only a single channel, and the channel label is irrelevant:
var marker_stream = new Dictionary<string, object>()
{
{"name","mymarkers"},
{"type", "Markers"},
{"sampling_rate", 0},
{"channels", new List<object> ()
{
new Dictionary<string, string> ()
{
{"label", "mylabel"}
}
}
}
};
The list of streams we transmit is, unsurprisingly, just a list of stream descriptors. If you (and the pipeline that you use) do not need a marker stream, you can simply omit it from this list:
Now we have all information at hand to write out the full metadata that we need to launch our pipeline instance. The metadata object has a list of input node descriptors and a list of output node descriptors (since pipelines can also have multiple output nodes). Since we expect that the pipeline just passes our inputs through without changes to either the number of streams and/or the number of channels, we can just reuse our input node descriptor for the output as well:
var metadata = new Dictionary<string, object>
{
{"nodes", new Dictionary<string, object> ()
{
{"in", new List<object>() {node_decl}},
{"out", new List<object>() {node_decl}}
}
}
};
With the metadata at hand, we can now form the HTTP request body that we use to bring up a pipeline instance. We need our pipeline ID as well as the metadata to configure our instance. Also, we may choose between either JSON or MessagePack encoding for our messages:
json_fallback = false; // true is JSON-encoded
// assemble the final launch options for the pipeline; for further options have
// a look at the API docs (for instance, you can toggle some server-side
// features)
var parameters = new Dictionary<string, object>
{
{"pipeline", pipeline_id},
{"metadata", metadata},
{"encoding", json_fallback ? "json": "msgpack"}
};
For initial tests and debugging it is fine to use JSON, but for production use you will want to use the much more efficient (faster and smaller) MessagePack encoding, especially if you stream data at moderately high rates (more than 10 channels x 100 Hz).
The request URL to launch a new instance is \
string jsonDumps = JsonConvert.SerializeObject(parameters);
var content = new StringContent(jsonDumps, Encoding.UTF8, @"application/json");
HttpResponseMessage postOrPatchResult;
if (String.IsNullOrWhiteSpace(instance_id))
{
// to launch our processing pipeline now we need to create a new "instance"
// of the desired pipeline (a running piece of code)
// IMPORTANT: note that, once the instance has been created, you are
// consuming resources in the cloud until you destroy the instance again
postOrPatchResult = client.PostAsync(@"v1/instances", content).Result;
// Blocking call!
}
The Content-Type header is implicit in the above call since we used json= to define the body. If the request succeeded you are now consuming resources in the cloud that you will need to release at some point, therefore please read on!
Note that the above request is used to create a new instance if we do not already have one that we want to reuse. If we do however want to recycle an existing instance, we need a slightly different request. For this, the HTTP verb is PATCH, and the URL is \
else
{
// we already know the id of a running instance and we'd like to reconfigure
// it with new parameters -- this is usually much faster
postOrPatchResult = client.PatchAsync(@"/v1/instances/" + instance_id, content).Result;
}
The above code does not work without defining the extension method “PatchAsync()” as follows (you can create it in the same namespace):
public static class HttpClientExtensions
{
public async static Task<HttpResponseMessage> PatchAsync(this HttpClient client, string requestUri, HttpContent content)
{
var method = new HttpMethod("PATCH");
var request = new HttpRequestMessage(method, requestUri)
{
Content = content
};
return await client.SendAsync(request);
}
}
Reconfiguring a running instance can be orders of magnitude faster than destroying the old instance and launching a new one (depending on the pipeline used to launch the instance). To complete the instance creation, we will want to check the status code of the HTTP response, for instance in case our URL or body was malformed in some way. Generally, a response code in the 200 range will indicate success, but note that the response code signaling success is 201 if you created a new instance, and 200 if you recycled an existing one:
Important!¶
You cannot reconfigure an instance to use a different pipeline (any attempt to do so will result in a HTTP 400 Bad Request error). You can only reset an existing instance or update an existing instance by changing its runtime parameters.
if (postOrPatchResult.StatusCode == System.Net.HttpStatusCode.Created ||
postOrPatchResult.StatusCode == System.Net.HttpStatusCode.OK)
{...
If instance creation was unsuccessful there is no need to clean up; otherwise you need to look up the instance ID from the response body, and use that later to destroy the instance again. To prevent accidental leakage of resources, instances will destroy themselves by default after running continuously without having received data for a few hours (this default can be changed in the instance creation HTTP request body). The instance ID of interest is simply the "id" field of the response body:
// Parse the response body.
string jsonContent = postOrPatchResult.Content.ReadAsStringAsync().Result;
var jsSer = new JavaScriptSerializer();
var postOrPatchBody = jsSer.Deserialize<Dictionary<string, dynamic>>(jsonContent);
instance_id = postOrPatchBody["id"];
You may find it instructive to inspect the remaining data that you get in the response body, which includes various other information about the instance and its input/output endpoints.
Waiting Until the Instance is Ready¶
The instance will not be immediately available to process data, and therefore you will want to wait until it has fully launched. This can be done by periodically polling the state of the instance until that state has changed to running (initially it will be configuring).
The HTTP request to check the state is a GET with URL \
Console.WriteLine("instance {0} has been requested successfully", instance_id);
// wait until the instance has launched; this is generally useful since
// any data you stream up while the instance is not yet running is lost
Console.WriteLine("waiting for instance to come up...");
var last_state = string.Empty;
while (last_state != "running")
{
HttpResponseMessage comeUpResponse = client.GetAsync(@"/v1/instances/" + instance_id).Result; // Blocking call!
jsonContent = comeUpResponse.Content.ReadAsStringAsync().Result;
jsSer = new JavaScriptSerializer();
var comeUpResponseBody = jsSer.Deserialize<Dictionary<string, dynamic>>(jsonContent);
var state = comeUpResponseBody["state"];
if (state != last_state)
{
Console.WriteLine(state + "...");
last_state = state;
Thread.Sleep(1000);
}
Obtaining the MQTT Input/Output Endpoints¶
At this point, you have a running instance, but you do not yet know the concrete endpoints over which to send/receive data to communicate with the instance. These endpoints can be looked up from the response body that corresponded to the \
read_endpoint = get_endpoint(postOrPatchBody, "read");
write_endpoint = get_endpoint(postOrPatchBody, "write");
static Uri get_endpoint(Dictionary<string, dynamic> body, string mode = "read")
{
var urls = new List<string>();
foreach (var e in body["endpoints"]["data"])
{
if (e["mode"] == mode)
urls.Add(e["url"]);
}
Uri uri = new Uri(urls.FirstOrDefault<string>());
return uri;
}
Setting up Communication With the Instance¶
Now that we have the endpoints, we can establish communication with the instance. Data transmission is done via MQTT, so we first create an MQTT client (the userdata is just for our own use in our callback function implementations). Since we have two endpoints, we need to create two clients, one per endpoint:
MqttSslProtocols sslprotocol = MqttSslProtocols.None;
System.Security.Cryptography.X509Certificates.X509Certificate caCert = new System.Security.Cryptography.X509Certificates.X509Certificate();
mqttClientReader = new MqttClient(read_endpoint.Host, read_endpoint.Port,
false, caCert, sslprotocol);
mqttClientWriter = new MqttClient(write_endpoint.Host, write_endpoint.Port,
false, caCert, sslprotocol);
We will also need a handler that interprets the messages that we receive from the pipeline instance. Messages typically carry chunks of time series data, both in the upstream (raw data) and in the downstream (processed data):
Lastly, we can increase the robustness of our client by making it automatically reconnect to the MQTT broker if it ever gets disconnected. The disconnection can happen, for example, during debugging when the program is paused for too long; in those cases there may be a backlog of messages accumulating in the broker which can cause it to disconnect the offending client:
private static void onConnectionClosed(object sender, EventArgs e)
{
Console.WriteLine("Connection Closed for id = " + e.ToString());
MqttClient readerOrWriter = sender as MqttClient;
if (sender != null)
{
byte code = mqttClientReader.Connect(Guid.NewGuid().ToString());
if (code == 0)
Console.WriteLine("Reconnected successfully");
else
Console.WriteLine("Failed to reconnected");
}
}
Most MQTT implementations also support other handlers, such as onSubscribed(), onPublish (), onConnectionClosed(), and onUnsubscribed() for diagnostics. Having defined the handlers, we can complete the MQTT setup:
mqttClientReader.MqttMsgPublishReceived += onReceived;
mqttClientReader.MqttMsgPublished += onPublish;
mqttClientReader.MqttMsgSubscribed += onSubscribed;
mqttClientReader.MqttMsgUnsubscribed += onUnsubscribed;
mqttClientWriter.MqttMsgPublishReceived += onReceived;
mqttClientWriter.MqttMsgPublished += onPublish;
mqttClientWriter.MqttMsgSubscribed += onSubscribed;
mqttClientWriter.MqttMsgUnsubscribed += onUnsubscribed;
mqttClientWriter.ConnectionClosed += onConnectionClosed;
The next step is to actually connect the reader and the writer to their respective endpoints:
byte readerCode = mqttClientReader.Connect(Guid.NewGuid().ToString());
if (readerCode == 0)
{
Console.WriteLine("Reader connected successfully");
}
else
{
Console.WriteLine("Reader failed to reconnected");
return;
}
byte writerCode = mqttClientWriter.Connect(Guid.NewGuid().ToString());
if (writerCode == 0)
{
Console.WriteLine("Writer connected successfully");
}
else
{
Console.WriteLine("Writer failed to connected");
return;
}
Using the Paho MQTT client, the actual work is implemented in callback handler functions, which we go over next. Once our reader has connected to the host, we need to subscribe to a "topic", which is the MQTT way of specifying what messages shall be received. The topic is part of the URL of the read endpoint (specifically, the path of the URL) and begins with a '/' character. However, the topic namespace is hierarchical (like directories), and the read endpoint gives only the super-topic under which the pipeline instance generates all its data. Specifically, for every output node of the pipeline instance there will be a subtopic of that name (typically this will be a single subtopic named "default"). So if the read endpoint was, for instance, mqtt://myhost.com:1883/testing then the default node would publish its data under the topic /testing/default. If at some point we chose to listen to other nodes of a pipeline instance, we would have to subscribe to those subtopics as well. However, to make things simpler we can just subscribe to all subtopics, using the '#' (multi-level wildcard) subtopic as done below. We can then filter the received messages upon receipt according to their topic:
readerTopicPath = read_endpoint.AbsolutePath + @"/#";
ushort readerMsgId = mqttClientReader.Subscribe(new string[] { readerTopicPath }, new byte[] { MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE });
This concludes the connection establishment.
Streaming Data Up¶
Once all is set up, we are ready to stream data to the cloud. Since we have already declared the metadata of our streams, all we need to do is periodically send up chunks of the actual time series data which we might be getting from some sensor hardware (such as an EEG headset). The following example will skip the hardware aspect and stream up randomly generated data (there is a separate section dedicated to alternatively getting the data from actual hardware).
First off, the following steps will be performed in a main loop:
Next we generate a chunk of random "samples". Each sample is a multi-channel measurement at a particular point in time. For efficiency, you will usually want to transmit multiple samples at a time (unless the data source has a very low or irregular sampling rate), here 10. The generated chunk is a list of samples, each of which is a list of values (one per channel):
int chunk_samples = 10;
// 10 packets of 8 data points (we have 8 channels).
var samples = new List<List<double>>() { };
for (int i = 0; i < chunk_samples; i++)
{
var octet = new List<double>();
for (int j = 0; j < channel_labels.Count; j++)
{
octet.Add(new Random().NextDouble());
}
samples.Add(octet);
}
In NeuroScale, each transmitted sample has an associated timestamp, which indicates the time at which the underlying data was originally measured. Since we are generating samples, we also need to timestamp them. These timestamps should be floating-point numbers (double-precision) in seconds. If you do not transmit multiple streams that shall be synchronized, it does not matter what clock you read them off of and how accurate they are, so you can use the system clock with no repercussion. Generally, if you send a chunk, you need a timestamp for each sample, and a simple way to do that is to use the current clock reading for the last sample in the chunk, and successively smaller timestamps (by 1/sampling_rate) for the preceding samples, like we do it here:
var now = DateTime.Now.Second;
var timestamps = (from t in Enumerable.Range(1, chunk_samples).Reverse()
select now - (double)t / sampling_rate).ToList(); // Added ToList() to prevent
// msgpack error because appraratly msgpack is unable to serialze IEnumerable<T> type.
If you do need to synchronize multiple streams with each other, you should make sure that the timestamps for each stream are read off the sample clock. Also, if precision better than \~50ms is needed, you need to consider that by the time you get the sample from the hardware, it will already be a few milliseconds old (having travelled through the device, USB, various buffers, the driver stack, and the OS). If you know the average delay of your device, you can subtract this number from your timestamps to achieve accurate synchronization. For some hardware vendors and/or data acquisition systems (such as the Lab Streaming Layer), you may be able to inquire about the device latency if needed, or alternatively you can stimulate a signal in both of your devices, record the data, and then manually determine the relative delay between the two (which is all that matters).
When you attempt to achieve millisecond-level synchronization between devices, you may be worried about jitter in your timestamp measurements — it is important to note that NeuroScale will internally remove the jitter using a Kalman filter, so you do not (and should not) have to worry about it. Attempting to filter the jitter by hand will usually make it (much) worse.
The array of samples and timestamps is all we need to construct the chunk object for transmission. The chunk has its samples in a field named "samples", the timestamps in a field named "timestamps", and a name that needs to match the one that was declared in the metadata for that stream in the field "name". This object will later be encoded using either JSON or MessagePack:
var eeg_chunk = new Dictionary<string, object>()
{
{"name", "myeeg"},
{"samples", samples},
{"timestamps", timestamps}
};
For completeness we now also generate a dummy chunk containing some event markers. This chunk also has an array of samples, each of which represents a single event marker, and is stored as a list of one string (it is technically a 1-channel sample with string values instead of numeric values). The string defines the type of event (and/or its metadata if so desired) and the naming should match the way in which the pipeline expects these events to be formatted (e.g. a pipeline might require that button presses are indicated by a marker named "button"). The timestamps are simply the times at which the events happened, again read off the same clock as the signal of course. In this example, we are simply using three different time points that each lies in the past by a fraction of a second. In a real application, these events will have been generated elsewhere in the system and will have been time-stamped already at the time when they occurred, and then would be transported to the part of the program that is responsible for streaming markers up to the cloud. Also, while it is possible to have multiple marker streams, most pipelines will expect that there is only one marker stream which has all markers in it:
var markers = new List<List<string>>()
{
new List<string>() {"mrk1"},
new List<string>() {"mrk2"},
new List<string>() {"mrk3"}
};
var marker_stamps = new List<double>()
{
now - 0.025,
now - 0.01,
now - 0.05
};
var marker_chunk = new Dictionary<string, object>()
{
{"name", "mymarkers"},
{"samples", markers},
{"timestamps", marker_stamps}
};
Having defined all the data, we can define the object that shall be serialized into a message. This is simply a dict/struct with the field "streams" holding a list of chunks (one per stream):
var msg = new Dictionary<string, List<Dictionary<string, object>>>()
{
{"streams", new List<Dictionary<string, object>> ()
{
eeg_chunk,
marker_chunk
}
}
};
It is not necessary that a stream be present in every message (e.g. if no event happened, the marker chunk might be omitted). Also, it is not strictly necessary that the chunks in a single message overlap in time (e.g. the markers could be older than the EEG). It is important to realize that, as long as all the data is correctly time-stamped, any sort of transmission delays of one or more streams do not change the output values of the pipeline instance (within accuracy limitations), just their delivery time; however, it is best to not unnecessarily delay the transmission of data, and to consult the documentation of the specific pipeline that you use to see if there are any limitations on how much streams may be delayed with respect to each other at transmission time (for instance, one pipeline may have to delay its output until data from a straggler stream has arrived or another pipeline may discard delayed information in order to make a delivery deadline).
Once the message has been defined, we need to encode it either into JSON or MessagePack (the encoding used needs to match the one that was declared in the pipeline instance's launch parameters):
// depending on the pipeline configuration, we need to encode either
// via JSON or msgpack
byte[] encodedMsg;
if (json_fallback)
{
// JSON serialization
var jsonized = JsonConvert.SerializeObject(msg); // -debug TODO: get it back
encodedMsg = Encoding.UTF8.GetBytes(jsonized);
}
else
{
// msgpack serialization
var serializer = SerializationContext.Default.GetSerializer<Dictionary<string,
List<Dictionary<string, object>>>>();
using (var byteStream = new MemoryStream())
{
// Pack obj to stream.
Console.WriteLine("msgpacking...");
serializer.Pack(byteStream, msg);
encodedMsg = byteStream.ToArray();
}
}
Lastly, we can send off the message bytes by publishing it via MQTT. Same as for the read endpoint, the write endpoint contains the topic on which data shall be published as part of the URL (beginning with a '/'). Any messages that are published directly under the super-topic for the instance will go into the first input node that was declared when the pipeline was instantiated. If a pipeline has multiple input nodes, then messages can be addressed to different nodes by publishing them on a subtopic named the same as the respective node (e.g. "default" for the default node):
mqttClientWriter.Publish(write_endpoint.AbsolutePath, encodedMsg, MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE, false);
Lastly, we end the main loop by sleeping for the interval between any two successive chunks, which ischunk_samples/sampling_rate:
Streaming Data Down¶
Once the pipeline instance is producing output, the data will be received by the MQTT reader client, and will arrive at the onReceived() handler function introduced earlier. For pipelines with multiple outputs this function should first check that the topic matches the one of interest (unless that is already taken care of by subscribing to only the desired output node's topic):
static void onReceived(object sender, MqttMsgPublishEventArgs e)
{
var topic = e.Topic.Split('/').LastOrDefault();
if (topic != "default")
Console.WriteLine("received message on non-default topic ({0}); ignoring.", e.Topic);
Next, the payload of the message needs to be parsed. The message format is the same as for sending data, and is encoded using JSON or MessagePack, according to what was declared when the pipeline was first instantiated. Once decoded the list of chunks (one per output stream) can be obtained from the "streams" field of the decoded object. Just like a single node can take in multiple streams simultaneously, an output node can also produce multiple output streams. In the case of the Echo pipeline, the output streams are the same as the input streams, and, in fact, the messages should have the same content. Then, we go over the list of stream chunks and print the contents. As in the upstream, each chunk has a name, an array of multi-channel samples (each an array of values), and an array of timestamps (one per sample):
// the payload must first be decoded using either JSON or msgpack
// (depending on pipeline settings)
dynamic data;
if (json_fallback)
{
var payload = Encoding.UTF8.GetString(e.Message);
data = JsonConvert.DeserializeObject<Dictionary<string, dynamic>>(payload);
var streams = data["streams"];
if (streams.Count != 0)
Console.WriteLine("received message on topic {0}", topic);
else
Console.WriteLine("received empty message on topic {0}", topic);
foreach (var stream in streams)
{
var name = stream["name"];
var samples = stream["samples"];
var stamps = stream["timestamps"];
Console.WriteLine(" streams {0}: {1} samples", name, stamps.Count);
}
}
else // if MsgPack
{
var serializer = SerializationContext.Default.GetSerializer<Dictionary<string, List<Dictionary<string, MessagePackObject>>>>();
using (var byteStream = new MemoryStream(e.Message))
{
Console.WriteLine("msg unpacking...");
data = serializer.Unpack(byteStream);
List<Dictionary<string, MessagePackObject>> streams = data["streams"];
if (streams.Count != 0)
Console.WriteLine("received message on topic {0}", topic);
else
Console.WriteLine("received empty message on topic {0}", topic);
foreach (var s in streams)
{
var name = s["name"];
var samples = s["samples"].AsList();
var stamps = s["timestamps"].AsList();
Console.WriteLine("stream {0}: {1} samples", name, stamps.Count);
}
}
}
Instance Teardown¶
The example client performs a few additional steps prior to the teardown to offer the user a prompt to destroy or retain the instance for later reuse. A similar prompt should probably be implemented in custom clients that are used during research and development. It might be beneficial to more intelligently carry over the instance ID from one program run to the next as done here (for simplicity).
Tearing down an instance amounts to emitting a HTTP DELETE request with the URL set to \
When performing a deletion, the status code should be checked to ensure that the instance was actually deleted. The status code should be 204 for success: