Introduction

Stream Data Store (SDS) Smart Module is a scalable stream data storage service. SDS has infinite data streams to which you can store any kind of data that is generated in your IoT devices. If you create an SDS Smart Module, you can have multiple Streams in it. Atomic data in an SDS Stream is called a Data Packet, whose data structure consists of 1) byte sequence and 2) packet index within a stream. IoT devices can send their data to SDS Smart Module with the MQTT protocol, and your business application can retrieve uploaded data via a REST API by specifying the range of data you are interested in.

Setting up Stream Data Store Smart Module

ONE | In the Developer Console, navigate to the SMART MODULE LIST page and select +NEW. Click +Add on the section for Stream Data Store.

TWO | Fill out the "New Stream Data Store Module" form.

Data structure of Stream Data Store

When you create a Stream Data Store smart module in your project, your smart module will maintain structures like in the following diagram.

In a Stream Data Store smart module, you can create an arbitrary number of Streams (the bands in orange color).

In the Stream, there are Data Packets (yellow colored circles) which are entities of your applications' data. Each Data Packet is a tuple of 1) packet index and 2) actual data entity which is essentially an arbitrary byte sequence. Alongside with the Data Packets, you can have additional information, the so-called Metadata Packet (green colored callouts) in the Stream. Metadata Packet is represented by a JSON object which contains additional information about the corresponding Data Packet.

The group of Packets (Data Packets and Metadata Packets) enclosed by the curly braces in the diagram is called a Session. When your Device sends data to the Stream, you need to open and close the Stream, as you would do when working with the Unix filesystem. The session is automatically created when the stream is opened and closed. That means, the first Data Packet in a Session is the first Data Packet the Device has sent after it has opened the Stream, and the last Data Packet in the Session is the one that the Device has sent just before it has closed the Stream.

You can get access to those data structures via the REST API. Please refer to the API Documentation for the details.

Collecting Data

To send data to SDS, you need to use MQTT (please see How to use MQTT with MODE first if you are not familiar). The basic workflow consists of following three steps:

Just like writing to a file on a Unix filesystem, your IoT device needs to open an SDS Stream so that you can write Data Packets on it. Once your IoT device has opened the Stream, the other Devices cannot open it for writing until the device who opened that Stream would close it.

For step 1 and 3, opening/closing the Stream, you need to work with Bulk Data Request / Response MQTT topics, which are the special topics for enabling request-response communications.

For step 2, sending Data Packets, you would work with Bulk Data MQTT topic, which is a special topic for sending binary data.

In the following subsections, we will explain the details of the workflow.

Opening SDS Stream

The Bulk Data Request topic is used to open an SDS Stream.

The topic is in the format:

/devices/DEVICE_ID/bulkData/BULK_DATA_LABEL/request

The payload structure for the Request JSON object which consists of the following three data fields.

The entire request JSON to be published to the Bulk Data Request topic (e.g. /devices/1234/bulkData/mySDS/request) looks like the following example:

{
  "requestID": "24020E73-68E1-40AC-A8F1-763D77591ED7"
  "bulkDataLabel": "mySDS",
  "payload": {
    "requestType": "open",
    "streamName": "device1234:log-stream",
    "openFlag": 0
  }
}

The corresponding response is to be published back to the Bulk Data Response topic that the Device should subscribe to.

The topic is in the format:

/devices/DEVICE_ID/bulkData/BULK_DATA_LABEL/response

The response payload is also a JSON object which consists of following two data fields:

The status field should take one of the following values:

The entire Response JSON to be published to the Bulk Data Response topic (e.g. /devices/1234/bulkData/mySDS/response) that corresponds to the above Request JSON example will look like below.

{
  "requestID": "24020E73-68E1-40AC-A8F1-763D77591ED7",
  "status": "OK",
  "payload": {
    "status": "OK",
    "packetIndex": 0
  }
}

In this example Response, you can know the stream "device1234:log-stream" is created for the first time because the payload.packetIndex field in the Response JSON is set to 0. The next time the Stream is opened, the value of payload.packetIndex field in the Response for the stream open Request will be different — the value will be the packet index of the last Data Packet sent plus one.

Sending SDS Data Packets

Once a device has succeeded to open the stream, it can start sending actual data packets.

A Data Packet should be represented by a sequence of MessagePack format data consisting of following three factors:

Data Packet representation in MessagePack uses an extension type that must be "0". A custom encoder is likely needed to encode a Data Packet structure into MessagePack.

The following code snippet is an example code written in Go (with github.com/vmihailenco/msgpack) for defining a Data Packet structure which can be encoded as MessagePack.

type DataPacket struct {
    StreamName  string
    PacketIndex uint32
    Data        []byte
}

var _ msgpack.Marshaler  = DataPacket{}

func init() {
    msgpack.RegisterExt(int8(0), DataPacket{})
}

func (d DataPacket) MarshalMsgpack() ([]byte, error) {
    var buf bytes.Buffer
    enc := msgpack.NewEncoder(&buf)

    if err := enc.EncodeString(d.StreamName); err != nil {
        return nil, err
    }
    if err := enc.EncodeUint32(d.PacketIndex); err != nil {
        return nil, err
    }
    if err := enc.EncodeBytes(d.Data); err != nil {
        return nil, err
    }

    return buf.Bytes(), nil
}

The way to create data packets really depends on your application. You can send a single JSON object that represents your Device status at a certain point of time in a data packet. Or you can send unstructured log text on your device. You can also send a series of video data across multiple data packets.

But the following are the general rules of working with SDS.

The Data Packets need to be sent on the Bulk Data publish topic of MODE MQTT. In each MQTT Message for Bulk Data to be published, you can send multiple data packets.

Bulk Data payload is an arbitrary byte sequence, so you can send the sequence of the Data Packet byte representation described above.

Adding Metadata Packets when you send Data Packets

While you are sending Data Packets you can also send Metadata Packet(s).

The structure of Metadata Packet to be sent over the MQTT connection is similar to the Data Packet structure. The Metadata Packet is a tuple of the following three factors that are encoded by MessagePack:

Metadata Packet representation in MessagePack uses an extension type that must be "1". Again, a custom encoder is likely needed to encode a Metadata Packet structure into MessagePack.

The following code snippet is an example code written in Go (with github.com/vmihailenco/msgpack) for defining a Metadata Packet structure which can be encoded as MessagePack.

type MetadataPacket struct {
    StreamName  string
    PacketIndex uint32
    Content     map[string]interface{}
}

var _ msgpack.Marshaler = MetadataPacket{}

func init() {
    msgpack.RegisterExt(int8(1), MetadataPacket{})
}

func (m MetadataPacket) MarshalMsgpack() ([]byte, error) {
    var buf bytes.Buffer
    enc := msgpack.NewEncoder(&buf)

    if err := enc.EncodeString(m.StreamName); err != nil {
        return nil, err
    }
    if err := enc.EncodeUint32(uint32(m.PacketIndex)); err != nil {
        return nil, err
    }
    jsonBytes, err := json.Marshal(m.Content)
    if err != nil {
        return nil, err
    }
    if err := enc.EncodeString(string(jsonBytes)); err != nil {
        return nil, err
    }

    return buf.Bytes(), nil
}

Metadata Packets and Data Packets can be mixed up in the same payload for an individual MQTT Message for Bulk Data topic.

Closing SDS Stream

After all necessary Data Packets have been sent , it's time to close the Stream. Just like opening a Stream, closing a Stream uses the Bulk Data Request topic of MODE MQTT.

The payload structure for the request JSON object consists of the following three data fields:

As mentioned in the previous section, the packetIndex field should be the one that is attached to the last Data Packet sent.

The entire request JSON to be published to the Bulk Data Request topic (e.g. /devices/1234/bulkData/mySDS/request) which corresponds to the open Stream Request looks like the following:

{
  "requestID": "1B93967B-5DF2-4098-94EF-409DAC6866B9"
  "bulkDataLabel": "mySDS",
  "payload": {
    "requestType": "close",
    "streamName": "device1234:log-stream",
    "lastPacketIndex": 10
  }
}

The corresponding response is published to the Bulk Data Response topic that the device should subscribe to. The response payload is also a JSON object which consists of just one data field:

The status filed should take one of following values:

The entire Response JSON to be sent to the Bulk Data Response topic (e.g. /devices/1234/bulkData/mySDS/response) that corresponds to the above Request JSON example looks like below.

{
  "requestID": "1B93967B-5DF2-4098-94EF-409DAC6866B9",
  "status": "OK",
  "payload": {
    "status": "OK",
  }
}

Retrieving data

You can retrieve the stored Data Packets via the MODE REST API.

When you can download the Data Packets, you should make an HTTP GET request to the Data Packets endpoint. Please note the following:

The following code snippet is an example code written in Go (with github.com/vmihailenco/msgpack) for retrieving Data Packets using the REST API.

type Packet struct {
    StreamName  string
    PacketIndex uint32
    Data        []byte
}

var _ msgpack.CustomDecoder = &Packet{}

func (p *Packet) DecodeMsgpack(dec *msgpack.Decoder) error {
    return dec.DecodeMulti(&p.StreamName, &p.PacketIndex, &p.Data)
}

func RetrieveDataPacket(
    ctx context.Context,
    homeID uint64,
    smartModuleID string,
    streamName string,
    start int,
    stop int,
    apiKey string,
) ([]Packet, error) {
    reqURL := url.URL{
        Scheme: "https",
        Host:   "api.tinkermode.com",
    }
    reqURL.Path = fmt.Sprintf("/homes/%d/smartModules/%s/streams/%s/data", homeID, smartModuleID, streamName)
    params := url.Values{}
    params.Add("start", strconv.Itoa(start))
    params.Add("stop", strconv.Itoa(stop))
    reqURL.RawQuery = params.Encode()

    req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
    if err != nil {
        return nil, fmt.Errorf("failed to create request: %w", err)
    }
    req.Header.Set("Authorization", fmt.Sprintf("ModeCloud %s", apiKey))

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, fmt.Errorf("failed to get data from API: %w", err)
    }
    defer func() {
        _ = resp.Body.Close()
    }()

    // Go's http client automatically follows the redirect and handle chunked response
    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("unexpected response status: %d %s", resp.StatusCode, resp.Status)
    }

    dec := msgpack.NewDecoder(resp.Body)
    var ps []Packet
    for {
        var p Packet
        if err := dec.Decode(&p); err != nil {
            if !errors.Is(err, io.EOF) {
                return nil, fmt.Errorf("failed to decode: %w", err)
            }
            return ps, nil
        }
        ps = append(ps, p)
    }
}

For more details, please refer to the full REST API Documentation for SDS.