Creating a composite stream

A composite stream generates new Sensor Updates with SUs from other streams as input. In contrast with a simple stream that stores SUs  coming from the API, it needs a “current-value” property for the channels to calculate the new values, and optionally can make use of the properties “pre-filter” and “post-filter” to condition its the triggering.

servIoTicy composition

Composition pipeline

A Service Object streams can be the simple representations of external SU producers that stores all their outputs. servIoTicy also has the ability to host Service Objects with streams that are the SU producers themselves. Such streams are subscribed to the outputs of other streams (composite or not), and uses those input SUs as operands in a operation where the result is the new output SU. Conceptually, the pipeline of a composite stream might look like this:

Where three SUs  owned by stream A, stream B and stream C are sent to servIoTicy through the API and are stored. A composite stream is subscribed to the streams, and so it receives their SUs as input. It generates a new SU, stores it and becomes sent to further composite streams if any.

This composition is intended to be scalable. As such, it follows three key design principles:

• Event-driven: A new SU calculation is triggered in a stream when it receives an SU.
• Lock-free: A stream that needs of several different SUs to generate a new one will not lock until all of them are received. It makes use of the one SU received, and queries the last SUs from the other needed streams.
• Stream-oriented: The input of a stream is the immediate output of another.

Taking these principles into account, the previous pipeline would look more like this:

A stream B SU is sent to the platform and gets stored. Composite stream is subscribed to the stream B output, so it receives SU 2 and triggers the generation of a new SU. The generation of SU 4 also requires of SU 1 and SU 3, so composite stream queries them to streams A and C. A single event (receiving a SU) generates a single output in composite stream.

Data consistency

In order to have consistency in the stored data of the SO streams, only newer SUs are accepted by a composite stream. This means that in order to be processed, the lastUpdate field must be greater than the one contained in the last generated SU. If the SU is no newer, then it will be discarded.

In this figure a SU with a lastUpdate timestamp 3 has been received. The composite stream makes use of it and the ones from the streams A and C with timestamps 2 and 1. The previously generated SU by the composite stream has a timestamp 4, so it is newer than all the received SUs and the computation will be aborted.

As a side effect this avoids the execution of cycles which could keep infinite executions running in the platform.

SO with composite streams example

The following listing is an example of a SO with its two streams generating SUs from other streams output.

{
    "groups": {
        "group1": {
            "soIds": [
                "1393322724658c4f69d27627f496bbb6fadf64bb65f4c",
                "13933230199264843a282ca8c471e962b4c61c53dadf5"
            ],
            "stream": "location"
        },
        "group2": {
            "soIds": [
                "139332303660140b55563e1b34751a9496ecffb0c64a5",
                "139332304556888d068cef3e645fd803bb4d2a60a15ea"
            ],
            "stream": "location"
        }
    },
    "streams": {
        "proximity": {
            "pre-filter": "{$group1.}!=null && {$group2.}!=null",
            "channels": {
                "p": {
                    "current-value": "Math.sqrt(Math.pow({$group1.channels.latitude.current-value} - {$group2.channels.latitude.current-value}, 2) + Math.pow({$group1.channels.longitude.current-value} - {$group2.channels.longitude.current-value}, 2))",
                    "type": "number"
                }
            },
            "post-filter": "{$proximity.} == null || {$proximity.channels.p.current-value} != {$result.channels.p.current-value}"
        },
        "near": {
            "channels": {
                "n": {
                    "current-value": "{$proximity.channels.p.current-value} <= 0.0001",
                    "type": "boolean"
                }
            },
            "post-filter": "{$near.} == null || {$near.channels.n.current-value} != {$result.channels.n.current-value}"
        }
    }
}

In one stream it measures the distance between two location points, and in the other it takes the previous stream SUs to determine if the two location points are near or not. It introduces concepts such as groups and filters  that will be explored in following sections.

Calculate a new SU

To write a stream that instead of simply storing SUs from received from the exterior generates new SUs, at least one channel needs to have a “current-value” property.

"channels": {
                "example": {
                    "current-value": "true",
                    "type": "boolean"
                }
}

“current-value” is a string that must contain the right side of a JavaScript regular assignment. The value of this assignment must be able to be cast to the type in the “type” property. The accepted types for a composite stream are:

• number
• boolean
• string

Which are the javascript primary types.

In the example “type” is boolean, and so the expression in “current-value” is a boolean assignment. This expression can be more complex using operations or even some JavaScript objects as Math, String or Array.

"channels": {
                "example": {
                    "current-value": "(4 > 2 || 3 != 6/2) && true",
                    "type": "boolean"
                }
}

This would result always in “true”. It does not make much sense to write a complex expression that will generate always the same SU, though. To make dynamic assignment expressions input Sensor Updates are needed.

Input Sensor Updates

We need to define and identify the origins of Sensor Updates so they can be used to make proper calculations of new SUs.

groups

groups is a property at SO level in which groups of streams can be defined. Each group is an array of SO ids and a stream id. So, all the SOs must have a stream with the same name, and hopefully with the same form. From the example:

"groups": {
   "group1": {
       "soIds": [
            "1393322724658c4f69d27627f496bbb6fadf64bb65f4c",
            "13933230199264843a282ca8c471e962b4c61c53dadf5"
        ],
        "stream": "location"
    },
    "group2": {
        "soIds": [
            "139332303660140b55563e1b34751a9496ecffb0c64a5",
            "139332304556888d068cef3e645fd803bb4d2a60a15ea"
        ],
        "stream": "location"
    }
}

Where group1 identifies the origin of any output SUs from the stream location from the SOs  1393322724658c4f69d27627f496bbb6fadf64bb65f4c and 13933230199264843a282ca8c471e962b4c61c53dadf5. The same for group2 but from the SOs 139332303660140b55563e1b34751a9496ecffb0c64a5 and 139332304556888d068cef3e645fd803bb4d2a60a15ea.

Groups work in best effort basis. This means that a group will get the SUs from the existing stated SO ids that have the stated stream id. If any of the SO ids does not exist or does not have the stream id, it will be simply ignored.

Streams

An stream is defined by a simple SO itself, so we can use this local definition and reference it to obtain SUs with the stream name. If we have a stream called proximity, like in the example, we can get its SUs from another or the same stream in the SO by using its name.

The same behaviour can be achieved with groups, but the use of the stream names as identifiers of SU sources is more convenient.

result

The output SU from the stream can also be accessed once created, using the result id. This can only be done in the property post-filter (see Filters section).

Input SU navigation

Now we know how to identify the SUs we are interested in, but we need a way to navigate on them and get the information we need from the JSON document. Remember that a SU has the following shape:

{
    "channels": {
        "latitude": {
            "current-value": 41.3879758,
            "unit": "degrees"
        },
        "longitude": {
            "current-value": 2.1150167,
            "unit": "degrees"
        }
    },
    "lastUpdate": 1392984030
}

To navigate and extract data from such a document JSONPath is used. JSONPath is a JSON navigation language, similar to XPath for XML (See http://goessner.net/articles/JsonPath/ for JSONPath specifics). To access the longitude current-value property using the previous document, in JSONPath it would look like:

$.channels.longitude.current-value

This would return 2.1150167.

 JSONPath works in a context in which there is only one document to navigate, in the case of our example this is the SU. But in the case of a composite stream, there are several origins with their own identifiers. In order to know which document is being navigated, the origin identifier from the previous section has to be inserted between the first ‘$’ and ‘.’ characters.

Then consider that the example SU comes from the group2 of the previous section. It should be accessed like this:

 $group2.channels.longitude.current-value

In the same way, to get the current-value from the only channel of the stream proximity, it should be accessed like this:

$proximity.channels.p.current-value

Going back to the current-value field of a composite stream, the JSONPaths must be written in the JavaScript assignment between the ‘{‘ characters ‘}’. For example, the stream near that takes the SUs from the proximity stream, which has the distance between two points, checks if the points are near or not:

"near": {
    "channels": {
        "n": {
            "current-value": "{$proximity.channels.p.current-value} <= 0.0001",
            "type": "boolean"
        }
    }
}

If the JSONPath does not navigate correctly through the data because the path does not exist, it will return null. This may be due to the specific field does not exist in the SU, or because the origin has not SUs yet. It can be used and controlled in the JavaScript code.

The different possible origin identifiers may collide, for example having a group called like a stream, or having a stream called result. For this cases, the identifiers work in a cascade fashion:

result → stream → group

First the result identifier is defined, then the streams overwriting the previous identifiers, and then the groups which will overwrite all the previous. So, if there is a stream called result, $result.channels.p.current-value will always access the result stream. And if there is a stream and a group both of them called proximity,  $proximity.channels.p.current-value will always access the group.

Filters

Not always is interesting to generate a new SU from some input. It may in the inputs and the outputs. For this reason a composite stream may make use of two additional properties: pre-filter and post-filter.

pre-filter

It is a String and must contain a JavaScript conditional expression which can make use of JSONPaths on the origins. If the condition expression is false then the SU calculation is discarded, the current-value will not be processed.

If this field does not appear in the stream, its value is “true” by default.

"proximity": {
    "pre-filter": "{$group1.}!=null && {$group2.}!=null",
    "channels": {
        "p": {
            "current-value": "Math.sqrt(
                                 Math.pow({$group1.channels.latitude.current-value} - {$group2.channels.latitude.current-value}, 2) + 
                                 Math.pow({$group1.channels.longitude.current-value} - {$group2.channels.longitude.current-value}, 2)
                             )",
            "type": "number"
        }
    }
}

This code is a stream with a channel that calculates the distance between two points received from two groups. If the SO with this stream is deployed when the origins has no data yet, the pre-filter it contains could be useful. The pre-filter in this stream considers this possibility and if there is not all the needed SUs, the calculation will not occur.

post-filter

Works basically in the same way as pre-filter, but with two main differences.

• Instead of being evaluated before the current-value, it is evaluated after. So it can not prevent the calculation of the new SU, but can prevent its storing and emitting stages.
• It makes use of the result document id to use the result SU in the filtering process.

So returning to the initial example:

"proximity": {
    "pre-filter": "{$group1.}!=null && {$group2.}!=null",
    "channels": {
        "p": {
            "current-value": "Math.sqrt(
                                 Math.pow({$group1.channels.latitude.current-value} - {$group2.channels.latitude.current-value}, 2) + 
                                 Math.pow({$group1.channels.longitude.current-value} - {$group2.channels.longitude.current-value}, 2)
                             )",
            "type": "number"
        }
    },
    "post-filter": "{$proximity.} == null || {$proximity.channels.p.current-value} != {$result.channels.p.current-value}"
}

This post-filter avoids creating an SU with the same value as the previously generated one. More literally, the SU will be only stored and emitted if there is not a previous SU, or if the previous SU has the same value.