AWS Integration

Xively can be integrated into Amazon Web Services (AWS) environments to allow you to gain the benefits of Xively with the services available in AWS.

Xively’s Amazon Kinesis Bridge enables you to forward MQTT messages to a Kinesis stream in your AWS account. This allows you to obtain a real-time stream of messages from your device population and process that stream using AWS services.

You can selectively control which messages are forwarded to the Kinesis stream, based on the Xively device channels (MQTT topics) you designate for delivery into the stream. This allows you to have streams flowing to different consumers without requiring all device messaging to flow through your AWS account.

Why might I want to use this?

  • You are building a system in AWS which needs to receive messages from Xively devices
  • You want to integrate data from your IOT devices with AWS services such as Machine Learning
  • You want to use AWS services (such as push notifications) to send alerts to your customers based on messages received from their devices

How do I connect the bridge?

In your Xively account in CPM, administrators can specify one Kinesis stream to attach to your Xively account.

Once the the Kinesis stream is attached, Xively begin to deliver messages into the stream.

The steps below describe the process to attach a Kinesis stream to your Xively account.

Step 1: Create/choose a Kinesis Stream

If you haven't already, set up a Kinesis stream in your AWS account.

When you create the stream, an initial shard count of one is usually sufficient. You can increase the shard count at a later time without changing the bridge configuration.

After creating the stream, copy the ARN for your Kinesis stream someplace where you can paste it in the following steps.

You can find the ARN for your Kinesis stream as follows:

The stream ARN has the following format
arn:aws:kinesis:<region>:<account-id>:stream/<stream-name>.

Step 2: Create an AWS policy for Xively

Create a policy to allow PutRecords calls on your Kinesis stream. This policy allows the minimal access required to call PutRecords on the stream.

  • Go to the IAM console: https://console.aws.amazon.com/iam.
  • In the navigation column on the left, choose Policies.
  • At the top of the page, choose Create Policy.
  • In the Visual Editor, select Choose a Service.
  • Select Kinesis.
  • Click on Select Actions.
  • Expand the Write options.
  • Select PutRecords.
  • Scroll down to the next section, Resources, and click on the section.
  • The Resources area will expand, click on Add ARN.
  • Paste your full Kinesis stream ARN into the field, replacing any existing contents.
  • You should see the Region, Account, Stream Name fields populate.
  • Click Add.
  • Click the Review Policy button at the lower right.
  • Name the policy in accordance with any naming conventions you have. As an example, "KinesisBridgeDev".
  • Click the Create Policy button at the lower right.
  • The policy is created.

The Policy Summary should appear as follows:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": "kinesis:PutRecords",
            "Resource": "arn:aws:kinesis:xx-region-1:999999999999:stream/your-stream-name"
        }
    ]
}

Detailed information about IAM policies is available at
http://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html.

Step 3: Create an AWS role for Xively

Create a role to allow Xively to access your Kinesis stream using the policy created in the previous step.

To complete this step, you need your Xively account ID. You can obtain your Xively account ID by logging into CPM and clicking on your username at the top right of the CPM screen.

  • Go to the IAM console: https://console.aws.amazon.com/iam.
  • In the navigation column on the left, choose Roles.
  • Click the Create Role button at the lower right.
  • Select the "Another AWS Account" tile for type of trusted entity
  • Specify 462322024086 for the Account ID. This is the Xively AWS account ID which will make PutRecords calls to your stream.
  • Select the option to require an external ID.
  • Specify your Xively account ID as the External ID.
  • Click the Next: Permissions button at the lower right.
  • Search for the policy you created in the previous step, and click the checkbox for the policy.
  • Click the Next: Review button at the lower right.
  • Name the role in accordance with any naming conventions you have. As an example, "XivelyKinesisDev".
  • Click the Create Role button at the lower right.
  • The role is created.

After creating the role, copy the ARN for the role someplace where you can paste it in the following steps.

You can find the ARN for the role as follows:

  • Search for the role name.
  • Click on the name of the role.
  • You should see the Role Summary screen, at the top will be the Role ARN.

To verify the role configuration, in the role summary, under the Permissions tab, you should see the policy name you created in the previous step. Under the Trusted Relationships tab, you should see two tables - a Trusted Entities table listing the Xively AWS account ID and a Conditions table listing a "StringEquals" condition for "sts:externalId" with a value equal to your Xively account ID.

Detailed information about IAM roles is available at (http://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user.html).

Step 4: Tell Xively where to forward your messages

For this step, you will need:

  • the ARN of the Kinesis stream you created in step 1.
  • the ARN of the role you created in step 3.

To configure the bridge in Xively:

  • Go to CPM.
  • Under "Integrations", find the Amazon Kinesis Bridge Integrations and click "Set up".
  • Enter the ARN of your Kinesis stream from step 1.
  • Enter the ARN of the Role from step 3.
  • Click "Connect" (or "Save" if you are editing it).

Step 5: Specify which Xively channels should forward to your Kinesis stream

In CPM (and in the Xively API), Kinesis forwarding is controlled by a channel setting in the device template.

There are three possible Kinesis settings for a channel:

  • Do not send
  • Best effort
  • Guaranteed

The default is Do not send, which means the channel is not forwarded to the stream.

Selection of Best effort and Guaranteed depends on the QOS supported by your application for publishers on the channel:

If your channel supports only QOS0 publishers, select Guaranteed.

If you channel supports QOS1publishers, you can control PUBACK behavior.

  • Best effort: the Xively message broker sends a PUBACK to publishing clients immediately upon receiving a message, before attempting to forward that message to the Kinesis stream.
  • Guaranteed: the Xively message broker sends a PUBACK after a successful write (PutRecords) to the Kinesis stream.

In most circumstances, Guaranteed should be specified.

Step 6: Consume and process Kinesis messages

Xively delivers messages into the stream in a base64-encoded binary format.

If you are using node.js to process the messages, Xively provides a node.js library for parsing the messages into a JSON structure. The library is available at: https://github.com/xively/xively-kinesis-bridge-client.

The library is used by instantiating the parser and calling the parseData method.

// from command line

npm install xively-kinesis-bridge


// javascript

const xivelyKinesisBridge = require('xively-kinesis-bridge');

const parser = new xivelyKinesisBridge.KinesisBridgeEnvelopeParser();

// Assume event.kinesis.data contains the base64-encoded message from the stream
const envelope = parser.parseData(event.kinesis.data);


// typescript

import {KinesisBridgeEnvelopeParser} from 'xively-kinesis-bridge';

const base64Data = 'some base64 encoded string';
const parser = new KinesisBridgeEnvelopeParser();

const envelope : KinesisBridgeEnvelope = parser.parseData(base64Data);

The following example shows the format of the JSON returned by the parser. The payload of the MQTT message is located in contentBody. The MQTT payload may be any payload allowed by MQTT, including binary data.


{
    "headerVersion": 1,
    "timeUUID": "d7f00bfd-1a05-e811-b1e0-02df7d478174",
    "sourceNameLength": 36,
    "sourceName": "edbef238-17b7-4ede-93b8-ea1e4ea9e0f3",
    "sourcePropertiesLength": 0,
    "sourceProperties": {},
    "targetNameLength": 91,
    "targetName": "xi/blue/v1/e78b108a-8b5f-425e-8e95-08f57eef8bb5/d/6c2ec725-6aca-4038-8c11-ebfd165a8e4c/demo",
    "targetPropertiesLength": 311,
    "targetProperties": {
        "SchemaVersion": 1,
        "EntityType": "topic",
        "Owner": {
            "Id": "edbef238-17b7-4ede-93b8-ea1e4ea9e0f3",
            "Properties": {
                "SchemaVersion": 1,
                "EntityType": "device",
                "AccountId": "e78b108a-8b5f-425e-8e95-08f57eef8bb5",
                "TemplateId": "5fd86b0a-e0e1-40eb-a278-613df38483df",
                "OrganizationIds": [
                    "e24bf37c-353a-4c48-8fe6-46b6bd739e30"
                ]
            }
        }
    },
    "contentLength": 17,
    "contentBody": {
        "type": "Buffer",
        "data": [
            123,
            34,
            104,
            101,
            108,
            108,
            111,
            34,
            58,
            34,
            119,
            111,
            114,
            108,
            100,
            34,
            125
        ]
    }
}

Accessing publisher and group information

For convenience, Xively includes the following information in the kinesis payload:

  • Who published the message
  • The channel the message was published to
  • The Xively group of the target device

The publisher is located in the sourceName field. If published using Xively device MQTT credentials, the sourceName is the Xively device ID. If published using Xively user MQTT credentials, the sourceName is the Xively Blueprint ID for the user. If published using a Xively user JWT, the sourceName will be the IDM user ID.

The channel the message was published to is located in the targetName field. This is the full MQTT topic name in the Xively standard format, which includes the device ID of the target device.

The Xively group of the target device is located in targetProperties.Owner.Properties.OrganizationIds[0]. This field indicates which Xively group the device is currently associated (paired) with.

Writing your own parser for the Kinesis stream

If you are using a language other than node.js for your application, you need to write your own parser. The following table describes the format of the binary message.

Size (in bytes)
Type
Description

1

byte

Header version

16

TimeUUID

Timestamp: GUID v1 (Cassandra TimeUUID)

2

word

Source name length

<Source name length>

string

Source name: UTF8-encoded string (Xively id of the entity that published the message)

2

word

Source properties length

<Source properties length>

JSON

Source properties

2

word

Target name length

<Target name length>

string

Target name: UTF8-encoded string

2

word

Target properties length

<Target properties length>

JSON

Target properties

4

dword

Content length

<Content length>

blob

Content body (MQTT payload)

For example, given the following base64-encoded data from the stream:

AdfwC/0aBegRseAC331HgXQkAGVkYmVmMjM4LTE3YjctNGVkZS05M2I4LWVhMWU0ZWE5ZTBmMwAAWwB4aS9ibHVlL3YxL2U3OGIxMDhhLThiNWYtNDI1ZS04ZTk1LTA4ZjU3ZWVmOGJiNS9kLzZjMmVjNzI1LTZhY2EtNDAzOC04YzExLWViZmQxNjVhOGU0Yy9kZW1vNwF7IlNjaGVtYVZlcnNpb24iOjEsIkVudGl0eVR5cGUiOiJ0b3BpYyIsIk93bmVyIjp7IklkIjoiZWRiZWYyMzgtMTdiNy00ZWRlLTkzYjgtZWExZTRlYTllMGYzIiwiUHJvcGVydGllcyI6eyJTY2hlbWFWZXJzaW9uIjoxLCJFbnRpdHlUeXBlIjoiZGV2aWNlIiwiQWNjb3VudElkIjoiZTc4YjEwOGEtOGI1Zi00MjVlLThlOTUtMDhmNTdlZWY4YmI1IiwiVGVtcGxhdGVJZCI6IjVmZDg2YjBhLWUwZTEtNDBlYi1hMjc4LTYxM2RmMzg0ODNkZiIsIk9yZ2FuaXphdGlvbklkcyI6WyJlMjRiZjM3Yy0zNTNhLTRjNDgtOGZlNi00NmI2YmQ3MzllMzAiXX19fREAAAB7ImhlbGxvIjoid29ybGQifQ==

The decoded base64 displayed with hexdump -C is:

00000000  01 d7 f0 0b fd 1a 05 e8  11 b1 e0 02 df 7d 47 81  |.............}G.|
00000010  74 24 00 65 64 62 65 66  32 33 38 2d 31 37 62 37  |t$.edbef238-17b7|
00000020  2d 34 65 64 65 2d 39 33  62 38 2d 65 61 31 65 34  |-4ede-93b8-ea1e4|
00000030  65 61 39 65 30 66 33 00  00 5b 00 78 69 2f 62 6c  |ea9e0f3..[.xi/bl|
00000040  75 65 2f 76 31 2f 65 37  38 62 31 30 38 61 2d 38  |ue/v1/e78b108a-8|
00000050  62 35 66 2d 34 32 35 65  2d 38 65 39 35 2d 30 38  |b5f-425e-8e95-08|
00000060  66 35 37 65 65 66 38 62  62 35 2f 64 2f 36 63 32  |f57eef8bb5/d/6c2|
00000070  65 63 37 32 35 2d 36 61  63 61 2d 34 30 33 38 2d  |ec725-6aca-4038-|
00000080  38 63 31 31 2d 65 62 66  64 31 36 35 61 38 65 34  |8c11-ebfd165a8e4|
00000090  63 2f 64 65 6d 6f 37 01  7b 22 53 63 68 65 6d 61  |c/demo7.{"Schema|
000000a0  56 65 72 73 69 6f 6e 22  3a 31 2c 22 45 6e 74 69  |Version":1,"Enti|
000000b0  74 79 54 79 70 65 22 3a  22 74 6f 70 69 63 22 2c  |tyType":"topic",|
000000c0  22 4f 77 6e 65 72 22 3a  7b 22 49 64 22 3a 22 65  |"Owner":{"Id":"e|
000000d0  64 62 65 66 32 33 38 2d  31 37 62 37 2d 34 65 64  |dbef238-17b7-4ed|
000000e0  65 2d 39 33 62 38 2d 65  61 31 65 34 65 61 39 65  |e-93b8-ea1e4ea9e|
000000f0  30 66 33 22 2c 22 50 72  6f 70 65 72 74 69 65 73  |0f3","Properties|
00000100  22 3a 7b 22 53 63 68 65  6d 61 56 65 72 73 69 6f  |":{"SchemaVersio|
00000110  6e 22 3a 31 2c 22 45 6e  74 69 74 79 54 79 70 65  |n":1,"EntityType|
00000120  22 3a 22 64 65 76 69 63  65 22 2c 22 41 63 63 6f  |":"device","Acco|
00000130  75 6e 74 49 64 22 3a 22  65 37 38 62 31 30 38 61  |untId":"e78b108a|
00000140  2d 38 62 35 66 2d 34 32  35 65 2d 38 65 39 35 2d  |-8b5f-425e-8e95-|
00000150  30 38 66 35 37 65 65 66  38 62 62 35 22 2c 22 54  |08f57eef8bb5","T|
00000160  65 6d 70 6c 61 74 65 49  64 22 3a 22 35 66 64 38  |emplateId":"5fd8|
00000170  36 62 30 61 2d 65 30 65  31 2d 34 30 65 62 2d 61  |6b0a-e0e1-40eb-a|
00000180  32 37 38 2d 36 31 33 64  66 33 38 34 38 33 64 66  |278-613df38483df|
00000190  22 2c 22 4f 72 67 61 6e  69 7a 61 74 69 6f 6e 49  |","OrganizationI|
000001a0  64 73 22 3a 5b 22 65 32  34 62 66 33 37 63 2d 33  |ds":["e24bf37c-3|
000001b0  35 33 61 2d 34 63 34 38  2d 38 66 65 36 2d 34 36  |53a-4c48-8fe6-46|
000001c0  62 36 62 64 37 33 39 65  33 30 22 5d 7d 7d 7d 11  |b6bd739e30"]}}}.|
000001d0  00 00 00 7b 22 68 65 6c  6c 6f 22 3a 22 77 6f 72  |...{"hello":"wor|
000001e0  6c 64 22 7d                                       |ld"}|