Stream Updates (Core Services)

Stream Updates is a way in PI Web API to stream incremental and most recent data updates for PIPoints/Attributes on streams and streamsets without opening a websocket. It uses markers to mark the specific event in a stream where the client got the last updates and uses those to get the updates since that point in the stream.

Markers

The marker is a way to mark where in the particular stream the client last got an update. It is used to find any new updates and detect if any updates have been missed by a client. Latest markers are returned both when the client registers for updates and retrieves the updates.

Below are some examples of StreamUpdates URLs:

• POST https://myserver/piwebapi/streams/{webId}/updates - Registering for updates
• GET https://myserver/piwebapi/streams/updates/{marker} - Retrieving updates using a marker
• POST https://myserver/piwebapi/streamsets/updates?{webIds} - Registering for updates for multiple streams
• GET https://myserver/piwebapi/streamsets/updates?{markers} - Retrieving updates from multiple streams

Usage

To use StreamUpdates, the client has to register an attribute or a point for updates by sending a POST request. If registration is successful, then the client can get updates by using the marker in the registration response and sending out the receive updates GET request. There is also an "Updates" link in the response and a link to the latest marker in the "Location" header of the response which can be used directly to get updates.

The response to the GET updates request will include a LatestMarker. This latest marker will now be the current position in the stream and the user can get new updates after this position by sending out GET request using this new marker. These requests can be chained to get incremental updates for registered resources. Stream Updates work in the same way for multiple resources using the streamsets endpoint.

Errors in Retrieval and Registration

When registering for updates, there are 3 possible statuses: 1. Succeded 2. AlreadyRegistered 3. Failed

In case of a Failed Status, clients should inspect the Exception property which will contain the registration errors. If there is an error retrieving updates for a stream, that stream will be unregistered for updates. In the event of an error during retrieval, the client should register for updates again and then request new data to replace the old data (for instance, by calling the streams/{webId}/recorded endpoint).

Metadata Changes

Stream Updates does not return any specific metadata/system-level changes to registered attributes or points; however, Stream Updates can notify you when any previously recieved data is no longer valid due to metadata/system-level changes. Some cases where this is possible include: - Changing the data reference for an attribute - Changing the unit of measure for an attribute - Deleting an attribute altogether

In these cases, Stream Updates will return an error stating that the previously returned data is no longer valid. The client can handle these errors like any other by registering for updates again and then requesting new data.

For AF Attributes, an error will be returned if any metadata changes are made to the attribute. Note that this does not include static value changes; static value changes will still be returned as normal updates.

For PI Points/Tags, __metadata changes will not return any errors__, unlike AF Attributes. To work around this limitation, it is recommended to use an AF Attribute with a PI Point data reference instead. Another option is to periodically poll for the point's metadata and track any changes clientside.

Other Remarks

• Clients should always use the most up-to-date marker to retrieve updates; previous markers may expire as events get cleaned up periodically.
• Using a Load Balancer with Stream Updates is only supported when the Load Balancer is configured with server affinity.

Sample Client

Below is a sample JavaScript client for StreamUpdates usage :

this.latestMarker = null; 
function DataQuery() {
    let piWebApi = myServer;           
    if (attribute != null) {

    //attribute is the selected attribute by the client
    var webId = attribute.webId;
    var xhr = new XMLHttpRequest();
    var url = "${piWebApi}/streams/updates";

    xhr.open("POST", url, true);

    xhr.onload = function() {
        //Set latest marker
        this.latestMarker = xhr.responseText.LatestMarker;
        }
    xhr.send(webId);
}

// Start up a thread to retrieve updates every 5 seconds
openedThread = setInterval(this.ReceiveUpdates(), seconds(5));
}

function ReceiveUpdates() {
    if (this.latestMarker == null) {
        return;
    }

    let piWebApi = myServer;
    let marker = this.LatestMarker;
    var webId = attribute.webId;
    var xhr = new XMLHttpRequest();
    var url = "${piWebApi}/streams/updates";

    xhr.open("GET", url, true);

    xhr.onload = function() {
        //Update Markers
        this.latestMarker = xhr.responseText.LatestMarker;
        //Display latest Events
        console.log(xhr.responseText.Events)                
    }
    xhr.send(this.latestMarker);        
}