Overview

The purpose of this document is to elaborate the mechanism for client applications to subscribe for remote events and receive remote event notifications via WebSockets. The general structure of the remote events that are pushed through the WebSockets and the syntax for encoding and providing event subscription criteria in order to create event subscriptions on the server are specified below.
A fully functional WebSocket server is in use. It implements the RFC-6455 web socket specification and allows the use of subprotocols and bi-directional communication.

Subscription Mechanism

Subscribing for Remote Events and Receiving Event Notifications

The event subscription is created at the moment the client application opens a connection via WebSocket. Event filtering criteria needs to be encoded in the target URI. Any events generated on the server that match the subscription are pushed through the web socket as long as the connection is open. The format of the events conforms to the general event format defined in Remote Events Subscriptions and Pull Notifications via Long Polling. Event representations are encoded in JSON.

Accepted URI Formats for Event Subscriptions via WebSockets


URI

Description

ws://<host>:<port>/m2m/[<gatewayID>]/re/subscriptions?topics=<topic1>,<topic2>,...&filter=<filter>

Creates an event subscription with simple subscription criteria which consist of a single subscription criterion defined via query parameters. The subscription criterion specifies topics of interest <topic1> , <topic2>, ..., <topicN> and an optional LDAP filter <filter> which should be defined over the properties of the events with the specified topics. The subscription is created at the moment the connection is opened and removed when the connection is closed. Any events generated on the server that match the subscription are pushed through the web socket as long as the connection is open. If the subscription will be scoped to a single OSGi service gateway accessed remotely via the RM then the gateway identifier shall be provided explicitly in the URI.

ws://<host>:<port>/m2m/[<gatewayID>]/re/subscriptions/[{"topics":[<topic11>, <topic12>, ...], "filter":<fitler1>}, {"topics":[<topic21>, <topic22>, ...], "filter":<fitler2>}, ...]

Creates an event subscription with subscription criteria consisted of multiple subscription criterion entries defined within a path parameter. Each subscription criterion specifies topics of interest and a corresponding optional LDAP filter which should be defined over the properties of the relevant events. The subscription is created at the moment the connection is opened and removed when the connection is closed. Any events generated on the server that match the subscription are pushed through the web socket as long as the connection is open. If the subscription will be scoped to a single OSGi service gateway accessed remotely via the RM then the gateway identifier shall be provided explicitly in the URI.

Possible subscription topics:

Type of Topic

Value

TOPIC_ALL

com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/*

TOPIC_ITEM_REGISTERED

com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/REGISTERED

TOPIC_ITEM_UNREGISTERED

com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/UNREGISTERED

TOPIC_ITEM_PROPERTY_CHANGED

com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/PROPERTY_CHANGED

TOPIC_ITEM_OPERATION_EXECUTED

com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/OPERATION_EXECUTED

Example Web Socket subscription:

  1. TOPIC_ALL ws://<host>:<port>/m2m/re/subscriptions?topics=com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/*
  2. TOPIC_ITEM_REGISTERED ws://<host>:<port>/m2m/re/subscriptions?topics=com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/REGISTERED
  3. TOPIC_ITEMUNREGISTERED ws://<host>:<port>/m2m/re/subscriptions?topics=com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/UNREGISTERED
  4. TOPIC_ITEM_PROPERTY_CHANGED ws://<host>:<port>/m2m/re/subscriptions?topics=com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/PROPERTY_CHANGED
  5. TOPIC_ITEM_OPERATION_EXECUTED ws://<host>:<port>/m2m/re/subscriptions?topics=com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/OPERATION_EXECUTED

This example covers ALL Functional Item Events and their respective topics as received by the Remote Manager. If you want a subscription for a specific gateway you need to add [<gatewayID>] as shown in the Accepted URI Formats for Event Subscriptions via Web Sockets table. 

Examples for subscribing for remote events and receiving pull notifications via websockets are provided below.

The communication is one-directional and all the subscription functionality is done with the URI. The user receives everything via the websockets without sending anything further.

The following JavaScript code snippet demonstrates creating a websocket subscription with a specified topic in the URI.

var webSocketConfig = {
baseWebSocketURL: (window.location.protocol.toLowerCase().indexOf('s') > -1 ? 'wss' : 'ws') + '://' + window.location.host,
openWS: null
};
function logMessage(eo) {
console.log(JSON.parse(eo.data));
}
 
function openWebSocket() {
webSocketConfig.openWS = new WebSocket(webSocketConfig.baseWebSocketURL + '/m2m/re/subscriptions/?topic=com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/REGISTERED');
webSocketConfig.openWS.onmessage = logMessage;
}
.....

Deleting the Subscription

The following JavaScript code snippet demonstrates deletion of a websocket subscription via closing the websocket connection.

function closeWebSocket() {
webSocketConfig.openWS.close();
webSocketConfig.openWS = null;
 
}
 
....

Java Example

Below are described the three classes in the Java Example for the WebSocket Push functionality.
The websocket dependency for compilation in the Maven POM is:


<dependency>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-client</artifactId>
<version>9.3.0.M2</version>
</dependency>

Push WS Example Class

This class is responsible for the user-friendly testing of the Events Push functionality – you will be prompted for the following input on bundle startup:

  • host - ws://<RM IP>
  • port - <RM Websockets port>


When the events have been sent in the Event Admin, the sample client will receive a JSON notification.
Code for unregistration has also been included, but is not used by the example.
The URI for subscription is /m2m/re/subscriptions .
The topic of the test events is "testTopic".
The Web Console Event page can be used to send sample events to test the example.

package maintest.m2m.test.doc;
 
import java.io.IOException;
 
import org.osgi.framework.BundleContext;
 
/**
* Push Example.
*/
public class PushWsExample {
 
BundleContext bc;
 
/**
* The active web socket.
*/
private JettyWebSocket activeClient = null;
 
/**
* Constructor for the push example.
*
* @param bundleContext The bundle context.
* @param name The name of the command group.
* @param help The help message.
*/
public PushWsExample(BundleContext bundleContext) {
this.bc = bundleContext;
}
 
/**
* Closes the active clients.
*/
public void close() {
try {
activeClient.destroy();
} catch (IOException e) {
e.printStackTrace();
}
 
}
 
/**
* Subscription method for demonstration purposes.
*
* @param host The web socket server host address, such as ws://172.0.0.1
* @param port The port of the web socket server, such as 1505
*/
public void subscribe(String host, String port) {
int portNumber = Integer.parseInt(port, 10);
 
try {
startNewClient(host, portNumber);
} catch (Exception e) {
e.printStackTrace();
}
}
 
/**
* Starts a new client.
* @param host The host of the client, such as ws://172.22.172.149
* @param port The port of the client.
* @throws Exception General, all-purpose exception.
*/
private void startNewClient(String host, int port) throws Exception {
final JettyWebSocket socket = new JettyWebSocket(host, port, "/m2m/re/subscriptions");
String topic = "com/prosyst/mbs/services/m2m/rest/fim/ItemRemoteEvent/PROPERTY_CHANGED";
 
// "(&(objectClass=com.prosyst.mprm.simple.test.fi.Sensor)(propertyName=temperature)(propertyValueNew>=25))";
String filter = "%28%26%28objectClass%3Dcom.prosyst.mprm.simple.test.fi.Sensor%29%28propertyName%3Dtemperature%29%28propertyValueNew%3E%3D25%29%29";
socket.start(topic, filter);
 
activeClient = socket;
}
 
}

Activator Class

The following class acts as a BundleActivator for the example:

package demo.test.education.samplewsclient;
 
import java.io.BufferedReader;
import java.io.InputStreamReader;
 
import org.osgi.framework.BundleActivator;
import org.osgi.framework.BundleContext;
 
/**
* Activator for the REST Events Push Example.
*
*/
public class Activator implements BundleActivator {
 
/**
* The example implementation we will use.
*/
private PushWSExample demoImpl;
 
/**
* Starts the bundle.
* @param context The bundle context.
* @throws Exception General, all-purpose exception.
*/
public void start(BundleContext context) throws Exception {
demoImpl = new PushWSExample(context);
System.out.println("Enter host:");
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
String host = reader.readLine();
System.out.println("Enter port:");
String port = reader.readLine();
demoImpl.subscribe(host, port);
}
 
/**
* Stops the bundle.
* @param context The bundle context.
* @throws Exception General, all-purpose exception.
*/
public void stop(BundleContext context) throws Exception {
if (demoImpl != null) {
demoImpl.close();
}
}
 
}

Jetty Web Socket Class

A Jetty Web Socket implementation for use in the example:

package maintest.m2m.test.doc;
 
import java.io.IOException;
import java.net.URI;
 
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
import org.eclipse.jetty.websocket.api.annotations.WebSocket;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
 
/**
* Jetty web socket implementation class for the REST Jetty Events Push example.
*/
@WebSocket()
public class JettyWebSocket {
 
/**
* Timeout for socket closure.
*/
private static final int SOCKET_CLOSURE_TIMEOUT_IN_MILLISECONDS = 1000;
 
/**
* The session of the web socket.
*/
private Session session;
 
/**
* If set to true, there is a pending session for a connection.
*/
private boolean isPendingSession = true;
 
/**
* The client related to the Web Socket.
*/
WebSocketClient client;
 
/**
* The host name.
*/
private final String host;
 
/**
* The port name.
*/
private final int port;
 
/**
* The alias of the web socket connection.
*/
private final String alias;
 
/**
* Constructor for the JettyWebSocket class.
*
* @param host The host name.
* @param port The port number.
* @param alias The alias of the web socket connection.
*/
public JettyWebSocket(String host, int port, String alias) {
this.host = host;
this.port = port;
this.alias = alias;
}
 
/**
* Starts the web socket.
*/
public void start(String topic, String filter) throws Exception {
if (client != null) {
return;
}
 
if (topic == null) {
System.out.println("Invalid topic!");
return;
}
 
String eventsURI = "?topics=" + topic + (filter == null ? "" : "&=" + filter);
URI uri = new URI(host + ":" + port + alias + eventsURI);
client = new WebSocketClient();
client.start();
ClientUpgradeRequest request = new ClientUpgradeRequest();
client.connect(this, uri, request);
final JettyWebSocket currentSocket = this;
Thread clientThread = new Thread(new Runnable() {
@Override
public void run() {
try {
synchronized (currentSocket) {
while (currentSocket != null && currentSocket.isConnected()) {
currentSocket.wait(SOCKET_CLOSURE_TIMEOUT_IN_MILLISECONDS);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
currentSocket.destroy();
} catch (Exception e) {
e.printStackTrace();
}
}
}
});
 
clientThread.start();
Thread.sleep(3000);
}
 
/**
* Reacts to the closing of the web socket.
* @param statusCode The status code of the closing.
* @param reason The reason for the closing.
*/
@OnWebSocketClose
public void onClose(int statusCode, String reason) {
try {
destroy();
} catch (IOException e) {
e.printStackTrace();
}
}
 
/**
* Reacts to a successful connection to a web socket.
*
* @param session The newly created session.
*/
@OnWebSocketConnect
public void onConnect(Session session) {
this.session = session;
isPendingSession = false;
}
 
/**
* Reacts on a received message /via the web socket/.
*
* @param message The received message.
*/
@OnWebSocketMessage
public void onMessage(String message) {
System.out.println("Message received:" + message);
}
 
/**
* Checks if the client is connected.
*
* @return True if connected, false otherwise.
*/
public boolean isConnected() {
return !isPendingSession && session != null && session.isOpen();
}
 
/**
* Destroys a web socket client.
*
* @throws IOException Exception in case of input/output problems.
*/
public void destroy() throws IOException {
if (session != null) {
session.close();
session.disconnect();
session = null;
}
 
if (client != null) {
client.destroy();
client = null;
}
}
 
/**
* Sends a message via the web socket.
*
* @param message The message to send.
*
* @throws IOException Exception in case of input/output problems.
*/
public void sendMessage(String message) throws IOException {
if (session == null || session.getRemote() == null) {
return;
}
 
System.out.println("Sending message");
session.getRemote().sendString(message);
}
 
}