Hot Path Subscription Example

The ABB Ability™ Platform provides a way to read alarms, events, platform events, and telemetry/variable data from the Platform in real time using the "Hot" path endpoint. This provides a faster, more performative way to request live data from the ABB Ability™ Platform, compared to the "Warm" or "Cold" path routes.

Inside the platform instance, device telemetry and platform events are routed through an Azure Service Bus Namespace via their respective service bus topics, which can be subscribed to for access to live data. This application uses the Subscriptions endpoint in the Data Access API to perform CRUD operations on a hot path subscription. A subscription can be created, renewed, and deleted via this endpoint.

We have created a .NET and Node.js examples of how to use Hot Path subscription functionality of Ability Platform.

In this tutorial, the applications function as follows:

1. The application reads a configuration file containing the Instance API endpoint URL, the type of subscription to create, the components of the subscription request body, etc.

2. The application uses information from the JSON response to the subscription request to create a service bus subscription client to receive messages on the Azure Service Bus topic.

3. The app listens for hot path data and writes this data to the console output.

4. The application runs for the life of the subscription request. When it completes, the application closes the client and deletes the subscription if it has not already expired.

Prerequisites

Follow the article with the technology of your liking.

The C# app was created using the .NET 5 framework. A basic console application will be created. Familiarity with the basics of .NET applications development is recommended. A basic knowledge of the ABB Ability™ Platform, and the Data Access API is assumed. A basic knowledge of Azure Service Bus topics and subscriptions is helpful since ABB Ability™ Platform uses Service Bus for Hot Path functionality.

The corresponding code for this tutorial can be found at this ABB Codebits repository.

The application can be run and compiled using Visual Studio or Visual Studio Code with the .NET Core SDK, or run from the command line.

The Node.js app uses axios to perform HTTP requests, and @azure/service-bus for communication with Azure Service Bus. A basic console application will be created. Familiarity with the basics of Node.js is recommended. A basic knowledge of the ABB Ability™ Platform, and the Data Access API is assumed. A basic knowledge of Azure Service Bus topics and subscriptions is helpful since ABB Ability™ Platform uses Service Bus for Hot Path functionality.

The corresponding code for this tutorial can be found at this ABB Codebits repository.

The application can be run using any code editor (i.e. Visual Studio Code) or CLI.

To make a call to the API management gateway for the subscription endpoint, the application will need to present a bearer token as the value of the Authorization header in the POST request. Information about authorization with Ability APIs can be found here.

In this example we're using the Background Application (OAuth 2.0 Client Credentials flow). Have a look at how the access token is fetched:

static async Task<string> GetAccessToken()
{
  using var request = new HttpRequestMessage(HttpMethod.Post, Config.AccessTokenUrl)
  {
    Content = new FormUrlEncodedContent(new[]
    {
      new KeyValuePair<string, string>("grant_type", "client_credentials"),
      new KeyValuePair<string, string>("client_id", Config.ClientId),
      new KeyValuePair<string, string>("client_secret", Config.ClientSecret),
      new KeyValuePair<string, string>("scope", $"https://{Config.B2CTenant}.onmicrosoft.com/{Config.SolutionNamespace}/instance")
    })
  };
  using var response = await _httpClient.SendAsync(request);
  var responseString = await response.Content.ReadAsStringAsync();

  return JsonSerializer.Deserialize<AccessToken>(responseString, JsonSerializerOptions).access_token;
}

async function generateAccessToken() {
  const params = new URLSearchParams()
  params.append('grant_type', 'client_credentials')
  params.append('client_id', process.env.CLIENT_ID)
  params.append('client_secret', process.env.CLIENT_SECRET)
  params.append('scope', `https://${process.env.B2C_TENANT}.onmicrosoft.com/${process.env.SOLUTION_NAMESPACE}/instance`)

  const config = {
    headers: {
      'Content-Type': 'application/x-www-form-urlencoded'
    }
  }

  return (await axios.post(process.env.ACCESS_TOKEN_URL, params, config))
    .data
    .access_token
}

You can also use another authorization flow, but in such case you need to provide the access token by yourself (more on that later).

Create the Hot Path Subscription

Example Configuration Files

appsettings.json file:

{
  "instanceApiUrl": "https://ADDRESS.abilityplatform.app/v1/", // Ability Platform Instance API URL
  "subscriptionType": "variables", // variables or alarms or events, or platformEvents
  "subscriptionTtl": "60", // Duration of the life of your subscription in minutes (Min: 5)
  "subscriptionFilter": "", // filter of the subscription, i.e.: objectId='9130c8ed-f2f5-4f55-b2f4-afc1d9113e2d'
  "sasTokenTtl": "60", // Duration of the life of your SAS token in minutes (Min: 5)
  "accessToken": "", // User generated access token. If it's not provided, the app will fetch access token using parameters below
  "accessTokenUrl": "https://ADDRESS.abilityplatform.abb/api/oauth2/token", // Client Credentials Token endpoint
  "clientId": "", // Client Id used to get the access token
  "clientSecret": "", // Client Secret used to get the access token
  "B2CTenant": "", // B2C Tenant name used by your Platform instance
  "solutionNamespace": "" // Namespace of the solution where the Background app resides
}

.env file:

INSTANCE_API_URL=https://ADDRESS.abilityplatform.app/v1/
SUBSCRIPTION_TYPE=variables (it could also be alarms, events, or platformEvents)
SUBSCRIPTION_TTL=60
SUBSCRIPTION_FILTER=Some filter, i.e.: objectId='9130c8ed-f2f5-4f55-b2f4-afc1d9113e2d'
SAS_TOKEN_TTL=60
ACCESS_TOKEN=A pre-generated access token that will be used to create a subscription. Alternatively, access token may be generated by this app. In such case, do not include this parameter, provide the ones below.
ACCESS_TOKEN_URL=An URL to get an access token for a background application. It is not needed if you have used a pregenerated access token in ACCESS_TOKEN
CLIENT_ID=Client ID of your Background App identity. It is not needed if you have used a pregenerated access token in ACCESS_TOKEN
CLIENT_SECRET=Client Secret of your Background App identity. It is not needed if you have used a pregenerated access token in ACCESS_TOKEN
B2C_TENANT=B2C Tenant name of your instance. It is not needed if you have used a pregenerated access token in ACCESS_TOKEN
SOLUTION_NAMESPACE=The namespace of your Ability Platform solution. It is not needed if you have used a pregenerated access token in ACCESS_TOKEN%  

The apps are able to read the token from the configuration file, or they can fetch it using a Background Application identity. If you decide to provide access token in the configuration, don't bother with settings of:

• access token URL
• client ID
• client secret
• B2C Tenant
• solution namespace

They will not be used.

Ability Sandbox

If you are an Ability Sandbox user, you have to provide a pre-generated access token. Sandbox users are not able to create Background Applications.

Example Subscription Filter

// Will filter for all data associated with specific variable of specified objectId
"objectId='<some object id>' AND variable='<some variable name>'"

Hot Path Subscription Response

After performing a POST to the Subscriptions/{type} endpoint, the application will parse the JSON response for data needed to create a subscription client object.

Response Body

{
"subscriptionId": "<Subscription ID GUID>", // Name of application created subscription
"sasToken": "<Example SAS Token>", // Used to authenticate against the Service Bus Namespace
"connectionUrl": "<URL>" // Will be parsed for fully qualified namespace and entity path of client object
}

Listen for Hot Path Data

The application will create a Azure Service Bus Subscription Client, and register a method to handle messages and errors from the client

Register a message handler to the Azure Service Bus Subscription Client object:

private static void Subscribe(SubscriptionResponse subscriptionInfo)
{
  var connectionString = GetConnectionString(subscriptionInfo);
  _subscriptionClient = new SubscriptionClient(connectionString, subscriptionInfo.SubscriptionId);
  var messageHandlerOptions = new MessageHandlerOptions(HandleMessageError)
  {
    MaxConcurrentCalls = 1,
    AutoComplete = false
  };

  _subscriptionClient.RegisterMessageHandler(HandleMessage, messageHandlerOptions);
}

Read messages and print them to the console:

private static async Task HandleMessage(Message message, CancellationToken token)
{
  Console.WriteLine($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}");
  await _subscriptionClient.CompleteAsync(message.SystemProperties.LockToken);
}

Listen for exceptions from Client: i.e. Loss of connectivity, timeouts, authentication failures:

private static Task HandleMessageError(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
{
  Console.WriteLine("Error!");
  Console.WriteLine($"{exceptionReceivedEventArgs.Exception}");
  Console.WriteLine($"{exceptionReceivedEventArgs.ExceptionReceivedContext}");
  return Task.CompletedTask;
}

Connect with Service Bus and subscribe to the Hot Path:

function subscribe(subscriptionDetails) {
  const serviceBusClient = new ServiceBusClient(getConnectionString(subscriptionDetails))
  const receiver = serviceBusClient.createReceiver(getTopicName(), subscriptionDetails.subscriptionId, null)

  receiver.subscribe({
    processMessage: async (message) => {
      console.log(`Received message: SequenceNumber:${message.sequenceNumber} Body:${JSON.stringify(message.body)}`);
      await receiver.completeMessage(message)
    },
    processError: async (args) => {
      console.log(
        `Error occurred with ${args.entityPath} within ${args.fullyQualifiedNamespace}: `,
        args.error
      );
    }
  });
}

Renew or Delete a Subscription

When the Service Bus SAS token and Service Bus Topic Subscription expire, the application will terminate. Before ending, the application will close the client and delete the subscription using the DELETE message request, based on the /{{type}/{subscriptionId} of the subscription.

The lifetime of the SAS token can be extended, using the PUT message request based on the /{{type}/{subscriptionId} of the subscription. The provided samples do not cover that.

WARNING

If the bearer token expires before the subscription delete is called, the DELETE will fail with 403 Forbidden error

Running the Application

The application can be run from Visual Studio, or from the command line from the root directory of the application using the commands below

dotnet build
dotnet run

You can run the app from CLI:

node index.js

Validate The Application is Listening for Data

WARNING

By default, if no filtering is provided to the client, the application will listen to all messages on the specified Service Bus topic of the subscription for the entire platform instance.

In a running platform instance, there should be a number of steady telemetry streams comprising of data from connected edge devices, running applications, platform events, etc. These streams can be subscribed to and used to validate the functionality of the Hot Path Subscription application. The application can be validated by using a variety of methods. Some examples are given below:

• An ABB Ability™ Edge on the platform instance, sending heartbeat telemetry to the Platform

• An ABB Ability™ Edge running a custom module sending telemetry data or alarms and events.

• An application using the publishers data reingestion endpoint

• A curl command simulating data being pushed into the platform, using the Publishers Reingestion Endpoint

1. Create Publisher Token

  curl -X POST \
  https://api-xxxxx.abilityplatform.abb/v1/publishers/<application id>/token \
  -H 'Authorization: Bearer  <token>' \
  -H 'Content-Type: application/json' \
  -H 'cache-control: no-cache' \
  -d '{
      "sasTokenTTL": 60
  }'

2. Publish Data using SaS Token and Connection Url provided by token request

  curl -X POST \
  <Connection Url> \
  -H 'Authorization: <SaS Token>' \
  -H 'Content-Type: application/json' \
  -H 'ability-messagetype: <Type of Data to Push, usually <'timeSeries'>' \
  -H 'cache-control: no-cache' \
  -d '[
      {
        "model": "<Model of Information Model Object associated with Data>",
        "value": "<Value to be pushed>",
        "objectId": "<Object Id of Information Model Object instance>",
        "timestamp": "<Timestamp using ISO 8601 Round-trip Format Specifier>",
        "variable": "<Variable Name>"
      }
  ]'