trimblecloud-events

Events Client library is a lightweight module for Producing and Consuming events through a websocket connection. It is capable of:

• Sending and receiving events through a websocket connection at higher TPS.
• Retrying if the connection is lost or interrupted and ensuring events are not getting lost.
• Sending statistics (total events produced and consumed, time taken to publish…etc See more here) to Datadog and other metrics systems supported by Micrometer.
• Supporting Authentication with TID V4.
• Pause polling based on response queue size

Getting Started

Installation

Apache Maven

<dependency>
    <groupId>trimblecloud</groupId>
    <artifactId>trimblecloud-events</artifactId>
    <version>2.2.0</version>
</dependency>

Gradle

implementation group: 'trimblecloud', name: 'trimblecloud-events', version: '2.2.0'

This library is available in Trimble Artifactory.

Prerequisites

For an application to access the Events server:

1. The application must be registered in the Cloud Console.
2. You must authorize the application to publish to the event topic.
3. Use the Client Credential Grant Type to get token.

Compatibility

Compatible with Java 11 and above.

Publisher

A publisher sends events to Events server through a websocket connection. To publish events, publishers must need a:

• NameSpace configured
• An event Schema configured
• An Event with the event schema grouped under a namespace

Publishing events

Follow the below steps to publish events,

1. Configure the properties for the Publisher client

Below properties are necessary for publishing events.

Publisher Client Configuration

Key	Description	Value	Mandatory/Optional
ClientConfig.PUBLISH_SERVICE_ENDPOINT	Provide publisher service web socket url	wss://TBU:1000	Mandatory
ClientConfig.CONNECTION_TIMEOUT_MS	Specify web socket connection timeout	5000	Mandatory
ClientConfig.TID_OAUTH_URL	Provide TID authorization url	Staging : https://stage.id.trimblecloud.com/oauth/token Production : https://id.trimble.com/oauth/token	Mandatory
ClientConfig.CLIENT_ID	Provide API client Id	<CLIENT_ID>
ClientConfig.CLIENT_SECRETS	Provide API client secrets	<CLIENT_SECRET>	Mandatory
ClientConfig.TID_SCOPE	Set the value of TID Scope for the application. Applicable for clients that require TID scope for authentication	<TID_SCOPE>	Mandatory
ClientConfig.CONNECTION_RETRIES	Maximum retry count	int	Mandatory

Below code snippet shows how to create a Properties object to store the above configuration properties.

Properties publisherProperties = new Properties();
publisherProperties.setProperty(ClientConfig.PUBLISH_SERVICE_ENDPOINT, "wss://<<publisher_endpoint>>");
publisherProperties.setProperty(ClientConfig.CONNECTION_TIMEOUT_MS, "5000");
publisherProperties.setProperty(ClientConfig.TID_OAUTH_URL, "<TID_OAUTH_URL>");
publisherProperties.setProperty(ClientConfig.CLIENT_ID, "<CLIENT_ID>");
publisherProperties.setProperty(ClientConfig.CLIENT_SECRETS, "<CLIENT_SECRET>");
publisherProperties.setProperty(ClientConfig.TID_SCOPE, "<TID_SCOPE>");
publisherProperties.setProperty(CONNECTION_RETRIES, "30"); // Number of connection retries in case of network failure

2. Create Websocket Publisher factory

Here we create a WebSocket Publisher, which is used to establish a connection and publish events. PublisherFactory object is created with the below parameters.

Publisher Factory Parameters

Parameter	Type	Description
publisher type	String	Currently supported type ‘PublisherType.WEBSOCKET’.
publisher properties	java.util.Properties	Provide configuration properties required to create new Events websocket connection.
registry	io.micrometer.core.instrument.MeterRegistry	If metrics needed then provide micrometer supported registry for example DatadogMeterRegistry InfluxMeterRegistry etc. else null.

PublisherFactory publishFactory = new PublisherFactory(PublisherType.WEBSOCKET, publisherProperties, null);

The meter registry parameter is optional and can be used to provide a micrometer-supported registry, such as DatadogMeterRegistry or InfluxMeterRegistry. This allows for monitoring and metrics collection related to the publisher’s performance.

3. Create a Publisher instance

Publisher instance can be created in two ways

3.1. Client handles batch processing

This approach allows the client to explicitly handle the processing of events in batches. A Publisher object is created using the createEventsSimplePublisher method of the publishFactory object. This method takes the below callback function as a parameter.

Simple Publisher Parameters

Parameter	Type	Description
responseCallback	BiConsumer<BatchAck, PublisherException>	Lambda expression which acts as a websocket listener that receives response and any exceptions from the Events server. BatchAck object contains the acknowledged batchId and type of acknowledgement. If the PublisherException object is not null, then there is an exception for the request that has been sent to the server.

BatchAck

Parameter	Type	Description	Value
type	String	Type of message	Acknowledgement
batchId	String	Batch id received from the server	String

Below code shows how to create a publisher that allows client to handles event batching explicitly.

Publisher publisher = publishFactory.createEventsSimplePublisher((message, exception) -> {
    if(exception != null) {
      System.out.println(exception.getRequestDetail().getRequestId()); //Get ID of the failed request
      System.out.println(exception.getMessage()); //Get exception message
    } else {
      System.out.println(message.getBatchId()); //Get Id of the Batch
      System.out.println(message.getType()); //Get Type of the Acknowledgement
    }
});

3.2. Inbuilt batch processing

This approach allows the SDK to handle the batching of events internally, increasing network throughput. To enable internal batching, the createEventsPublisher method is invoked on the publishFactory object. This method creates a Publisher instance that is capable of handling batch processing. In order to enable batching, the ClientConfig.ENABLE_BATCH configuration property needs to be set to true. Below are the configuration parameters related to inbuilt batching processing,

Key	Description	Value
ClientConfig.ENABLE_BATCH	To enable internal batching. By default it is false. If it is set as False, batching and batch retry has to be handled by the client.	boolean
ClientConfig.BATCH_SIZE	Maximum size of each batch of events that are sent to the server. Once this size is reached, the batch is sent to the server.	1000
ClientConfig.BATCH_WAIT_MS	Maximum wait time in millis post which the batch is sent to the server.	1000
ClientConfig.BATCH_EXPIRY_MS	Time in millis post which the batch expires if not sent to the server. Once the batch expires it throws an exception and sends back all the events in the batch to the registered callback method.Ensure you are handling/resending the expired events.	5000
ClientConfig.BATCH_RETRIES	Total number of times a batch can be retried if the acknowledgement is not received from the server. When maximum retries are reached, the client throws a PublisherException and sends back the undelivered events. Ensure you are handling /resending the undelivered events.	10
ClientConfig.ACK_WAIT_MS	Specify maximum wait time to receive ACK from the server. After which,the client resends the event to the server.	10000

createEventsPublisher takes the below callback function as a parameter.

Publisher Parameters

Parameter	Type	Description
responseCallback	BiConsumer<PublisherRequest,PublisherException>	Lambda expression which acts as a websocket listener that receives response and any exceptions from the Events server. PublisherRequest object contains the request details and PublisherException object contains any exceptions that occurred during the request. If the PublisherException object is not null, then there is an exception for the request that has been sent to the server.

Below code shows how to create a publisher that handles batch processing internally.

Publisher publisher = publishFactory.createEventsPublisher((message, exception) -> {
    if(exception != null) {
       System.out.println(exception.getRequestDetail().getRequestId()); //Get ID of the failed request
    } else {
         System.out.println(message.getPayload().toString());
    }
});

4. Populate PublisherRequest object

To define the details of an event that will be published using the Events SDK, we can populate a PublisherRequest object with the below properties.

PublisherRequest

Parameter	Type	Description	Example
Namespace	String	Namespace name. It should follow the pattern of “com.trimble.division.servicename”.	com.trimble.tcp.events
Event	String	Name of the Event. It should contain only letters and numbers. No special characters are allowed.	UserDelete
EventVersion	String	Version of the event. It should follow the pattern v[0-9].[0-9].	v1.0, v2.0
payload	String	Publish payload object.	"{\"id\":\"1214\",\"specversion\":\"1.0\",\"source\":\"\\id\\/<apiid>\"," +"\"type\":\"delete.v1\",\"time\":\"{{isoTimestamp}}\",\"subject\":\"\\/id\\/UserDelete\"," +"\"data\":{\"uuid\":\"121212\",\"application\":\"02cf0a0c-19db-45eb-8b19-90f93c2703ce\"," +"\"subject\":\"02cf0a0c-19db-45eb-8b19-90f93c2703ce1\"}}"
additional Fields	Map<String, String>	Specify Additional fields
EntityId	String

Below code shows how to populate the PublisherRequest object.

PublisherRequest publisherRequest = new PublisherRequest();
publisherRequest.setNamespace("com.trimble.events.qa");
publisherRequest.setEvent("ws.qa.event");
publisherRequest.setEventVersion("v1.0");
publisherRequest.setEntityId("ab");
publisherRequest.setPayload("{\n" +
"        \"specversion\": \"1.0\",\n" +
"        \"source\": \"test\",\n" +
"        \"type\": \"test\",\n" +
"        \"time\": \"{{timestamp}}\",\n" +
"        \"subject\": \"test\",\n" +
"        \"data\": {\n" +
"            \"uuid\": \"test\"\n" +
"        }\n" +
"    }");

5. Add the above PublisherRequest to EventBatch object

To create a batch of events that can be published using the Events SDK, we need to add the PublisherRequest object to an EventBatch object.

In the beloww code, an EventBatch is created and set with a unique identifier. A List of PublisherRequest objects is created to hold the events that will be included in the batch. The publisherRequest object is added to the requestList. Finally, the setEvents method is called on the batch object, passing in the requestList. This associates the list of events with the batch.

EventBatch batch = new EventBatch();
batch.setId(UUID.randomUUID().toString());
List<PublisherRequest> requestList = new ArrayList<>();
requestList.add(publisherRequest);
batch.setEvents(requestList);

Events Batch

Parameter	Type	Description	Example	Mandatory /Optional
id	String	Batch Id.Batch id cannot be empty.	******************	Mandatory
events	list	List of Events. Can contain more than one event of type PublisherRequest. Cannot be empty.	List requestList = new ArrayList(); requestList.add(publisherRequest);	Mandatory

6. Publish the EventBatch object

Finally we have to publish the events in the batch to the Events server. The publish method is called on the publisher object, passing in the batch object as a parameter.

try {
      publisher.publish(batch);
} catch (PublisherException exception) {
      exception.printStackTrace();
}

Publisher Connection Status Methods

The below methods are used to check publisher’s connection status.

Parameter	Type	Description
isActive()	boolean	To check if the client is active (connection has been successfully established)
getStatus()	Status	To check the state of the client. Status.RUNNING -Current websocket connection is open. Status.STOPPED - Current websocket connection is closed or not active. Status.CONNECTING - Current websocket connection is in the process of establishing the connection

Exceptions

Parameter	Description
PublisherException	Publisher Exception is thrown in the following cases : Error while connecting to server, Connection disconnected, Authentication failure, Exceeded maximum Retry connection, Batch expired, Batch, ACK, Event Request cannot be sent, Header missing (Contact Events team for such Exception), Batch with empty events is passed in the send method.

Publisher Connection Close

Once the events are published, you have to close the websocket connection by calling close method.

publisher.close()

After the close is called,

• The ENS client will not accept any new events. If you try to send, it throws ConnectionClosedException.
• The undelivered events/not acknowledged events will be returned as a list.
• It is recommended for the Client application to stop sending messages and wait for ack with timeout before Close, if required.

Consumer

A Consumer consumes events from Events server through a websocket connection. It’s recommended that you review the schema of an event to decide whether an event is relevant to you. For an application to consume from Events server:

• Review the Events that you want to Consume.
• The application must be registered in the - Trimble Cloud Console.
• Submit the following Onboarding request to get access to the namespace and the event that you wish to access.

Consuming events

Follow the below steps to consume events,

1. Configure properties for consumer

First set the required properties for configuring a consumer in the Events SDK. Below properties are necessary for establishing a connection to the consumer endpoint and authenticating with the TID OAuth service.

Consumer Client Configuration

Key	Description	Value	Default Value	Mandatory/Optional
ClientConfig.CONSUMER_SERVICE_ENDPOINT	Provide consumer service web socket url	wss://TBU:1000		Mandatory
ClientConfig.CONNECTION_TIMEOUT_MS	Specify web socket connection timeout	5000		Mandatory
ClientConfig.TID_OAUTH_URL	Provide TID authorization url	Staging : https://stage.id.trimblecloud.com/oauth/token Production : https://id.trimble.com/oauth/token		Mandatory
ClientConfig.CLIENT_ID	Provide API client Id	<CLIENT_ID>		Mandatory
ClientConfig.CLIENT_SECRETS	Provide API client secrets	<CLIENT_SECRET>		Mandatory
ClientConfig.TID_SCOPE	Set the value of TID Scope for the application. Applicable for clients that require TID scope for authentication	<TID_SCOPE>		Mandatory
ClientConfig.CONNECTION_RETRIES	Maximum retry count	int		Mandatory
ClientConfig.RESPONSE_QUEUE_SIZE	size of the response queue. Each response object may have multiple objects inside it (Optional)	int	100	optional
ClientConfig.PAUSE_ON_RES_QUEUE_LIMIT	Ignore polling when response queue size reaches the limit (Optional)	boolean	false	optional

A properties object is created to hold the configuration properties.

Properties consumerProperties = new Properties();
consumerProperties.setProperty(ClientConfig.CONSUMER_SERVICE_ENDPOINT, "wss://<CONSUMER_ENDPOINT>");
consumerProperties.setProperty(ClientConfig.CONNECTION_TIMEOUT_MS, "5000");
consumerProperties.setProperty(ClientConfig.TID_OAUTH_URL,"<TID_OAUTH_URL>");
consumerProperties.setProperty(ClientConfig.CLIENT_ID, "<CLIENT_ID>");
consumerProperties.setProperty(ClientConfig.CLIENT_SECRETS, "<CLIENT_SECRETS>");
consumerProperties.setProperty(CONNECTION_RETRIES, "10");

2. Create ConsumerFactory instance

Next we have to create an instance of the ConsumerFactory that is responsible for creating a consumer that can receive events from the Events server using a WebSocket connection with the below parameters.

Consumer Factory Parameters

Parameter	Type	Description
consumer type	String	Currently supported type ‘ConsumerType.WEBSOCKET’.
consumer properties	java.util.Properties	Provide configuration properties required to create new Events websocket connection.
registry	io.micrometer.core.instrument.MeterRegistry	Provide micrometer supported registry for example DatadogMeterRegistry InfluxMeterRegistry.

Below code shows how to create a consumer factory instance.

ConsumerFactory consumerFactory = new ConsumerFactory(ConsumerType.WEBSOCKET, consumerProperties, null);

The meter registry parameter is optional and can be used to provide a micrometer-supported registry, such as DatadogMeterRegistry or InfluxMeterRegistry. This allows for monitoring and metrics collection related to the consumer’s performance.

3. Create Consumer instance

Here we have to create a Consumer instance using the ConsumerFactory and define a callback function to handle the response or exception. .The createConsumer method is called on the consumerFactory object, passing in the below callback as a parameter.

Consumer Callback

Parameter	Type	Description
responseCallback	BiConsumer<Response, ConsumerException>	Lambda expression which acts as a websocket listener that consumes event and any exceptions from the Events server. Response.getResponseMessage() contains the poll response from the server. ConsumerException object is not null, then there is an exception for the request that has been sent to the server.

Below code show hoe to create a consumer.

Consumer consumer = null;
consumer = consumerFactory.createConsumer((response, exception) -> {
            //Logic to handle response or exception from the server

});

Exceptions

Parameter	Description
ConsumerException	ConsumerException are thrown in the following cases : Error while connecting to server, Connection disconnected, Authentication failure, Exceeded maximum Retry connection, On receiving error messages from server on consume

4. Populate ConsumerRequest object

Next we populate a ConsumerRequest and specify the details of the event that the consumer wants to receive from the Events server.

ConsumerRequest consumerRequest = new ConsumerRequest();
consumerRequest.setNamespace("com.trimble.wse.java20211213103300");
consumerRequest.setEvent("Ee120211213103300");
consumerRequest.setEventVersion("v1.1");

Below are the paramters that can set to consumer request.

Consumer Request

Parameter	Type	Description	Example
namespace	String	Namespace name. It should follow the pattern of “com.trimble.division.servicename”.	com.trimble.tcp.events
event	String	Name of the Event. It should contain only letters and numbers. No special characters are allowed.	UserDelete
eventVersion	String	Version of the event. It should follow the pattern v[0-9].[0-9].	v1.0, v2.0
visibilityTimeout	long	Provide the duration the server should wait to receive acknowledgment for the message before resending the message. If the value is not provided, the default value is set to 30000 ms. Min value - 3000 ms, Max value - 3600000 ms	30000
subject	String

5. Call poll method with the above ConsumerRequest object as input parameter

Next we need to poll for events using the poll method of the Consumer class to receive events from the Events server. The code snippet shows an example of how to continuously poll for events using a ConsumerRequest object as the input parameter.

try
   while(true){
      sleep();
      consumer.poll(consumerRequest);
   }
  } catch (ConsumerException exception) {
   exception.printStackTrace();
}

Consumer Connection Status Methods

The below methods are used to check the consumer connection status.

Parameter	Type	Description	Return Value
isActive()	boolean	To check if the client is active (connection has been successfully established)	true/ false
getStatus()	Status	To check the state of the client.	Status.RUNNING -Current websocket connection is open. Status.STOPPED - Current websocket connection is closed or not active. Status.CONNECTING - Current websocket connection is in the process of establishing the connection

Long Poll

Long polling is a technique used for low volume events where the client makes a request to the server and waits for a response, keeping the connection open for a specified duration. Long poll is applicable for low volume events.

In Events SDK, the allowed poll duration for a long polling request is 60 seconds only. Any other value will be rejected by the server. Requests with additionalFields map with the key long.poll.duration and value as 60000 will be considered as a long poll request.

Only one long poll request is allowed for a particular Event for 60 seconds till the response is received for the same. Others will be rejected with the exception "Long poll is in progress for the event : ${eventName}".

Once the response has been received for a long poll request for a particular Event, we will be able to send a new long poll request for the same Event.

If there is already a long poll request from another client instance is in progress for a particular Event at the server side and if we send another long poll request for the same Event then we will get the exception, "Already long poll is in progress" from the server.

This code snippet provides an example of how to perform a long poll request using the Events SDK.

ConsumerRequest consumerRequest = new ConsumerRequest();
consumerRequest.setNamespace("com.trimble.wse.java20211213103300");
consumerRequest.setEvent("Ee120211213103300");
consumerRequest.setEventVersion("v1.1");
Map<String, String> additionalFields = new HashMap<>();
additionalFields.put(AdditionalField.LONG_POLL_DURATION, "60000");
consumerRequest.setAdditionalFields(additionalFields);
  try {
       consumer.poll(consumerRequest);
      } catch (Exception exception) {
        exception.printStackTrace();
     }

Explicit Acknowledgement of Messages

By using Client Acknowledgment, developers have more control over the acknowledgement process and can manually acknowledge messages or events based on their specific requirements. For polling, if we need to disable the auto acknowledgement of the Events client and manually send acknowledgement, we need to set the ClientConfig.CLIENT_ACKNOWLEDGE property to “true” and call acknowledge() method. The message can be acknowledged as a whole or each event in the message can be acknowledged.

Below code acknowledges the message as a whole.

private static void responseHandler(Response response, Exception exception) {
        if (exception != null) {
            System.out.println("Exception - " + exception.getMessage());
        } else {
            System.out.println("Response received " + response.getResponseMessages());
            System.out.println("Sending Acknowledgment");
            consumer.acknowledge(response.getAcknowledgmentId());
        }
    }

Message level Acknowledgment

Here each event within the message is acknowledged separately.

private static void responseHandler(Response response, Exception exception) {
        if (exception != null) {
            System.out.println("Exception - " + exception.getMessage());
        } else {
            System.out.println("Response received " + response.getResponseMessages());
            System.out.println("Sending Acknowledgment");
            consumer.acknowledge(response.getAcknowledgmentId());
            response.getResponseMessages().forEach(kafkaResponse ->
                consumer.acknowledge(kafkaResponse.getAcknowledgmentId()));
        }
}

Acknowledge method throws the following exception

• Connection Already Closed
• Connecting (initial connection or retry in progress)
• Any other client Exception on sending ack
• When tried to send acknowledgement with ClientConfig.CLIENT_ACKNOWLEDGE set to “false”`.

Acknowledgment not received within 30 seconds considered as failure events will be resent again in the subsequent polls

Events Ordering

To maintain event ordering, the publisher will publish message with entityId, which will be sent in the entityId filed along with the messages. For more information on orderings refer here.

private static void responseHandler(Response response, Exception exception) {
        if (exception != null) {
            System.out.println("Exception - " + exception.getMessage());
        } else {
            System.out.println("Response received " + response.getResponseMessages());
            message.getResponseMessages().forEach(kafkaResponse ->
                System.out.println(kafkaResponse.getEntityId());
        }
}

Consumer Connection Close

Once the events are consumed, you have to close the websocket connection by calling close method.

consumer.close()

After the close is called, The ENS client will not accept any poll request. If you try to send, it throws ConsumerException.

Connection Retry

• Provide the ClientConfig.CONNECTION_RETRIES in the configuration.
• If the connection could not be opened, or if abnormal closure occurred due to transient / non-transient error, the client retries internally for 15 times and resends the undelivered events.
• When maximum retry counts (ClientConfig.CONNECTION_RETRIES + 15 internal retries) are reached, the client throws a websocket exception and sends back the undelivered events. Ensure you are handling/resending the undelivered events.
• The Client uses an exponential backoff mechanism, when trying to reconnect . Internally the exponential retry wait time is set to 5 seconds.

NOTE: Exponential backoff algorithm means if the first reconnect attempt fails, subsequent reconnect attempts are delayed by increasingly longer amounts of time. For example, retry.wait.time is configured as 5sec, it may retry the connection after 5 seconds, 10 seconds, 15 seconds, and so on.

Monitor Application

Metrics helps to monitor the consumer and publisher. By enabling metrics and configuring a monitoring application, developers can gain visibility into the consumption of events and track any errors that occur. This allows for better monitoring and troubleshooting of the event processing system.

To improve the code, it would be beneficial to provide additional information about how to enable and configure Micrometer’s MeterRegistry object for metrics tracking in the Events SDK. Including examples or use cases demonstrating the practical usage of these metrics and how they can be integrated with a monitoring application would enhance the understanding of this code snippet. Metrics can be enabled by passing Micrometer’s MeterRegistry object while creating PublisherFactory/ConsumerFactory.

Below are the metrics that will be pushed to the monitoring application configured.

Metric	Description	Publisher/Consumer
no.of.events.produced	Number of events produced at a particular interval.	Publisher
no.of.events.consumed	Number of events consumed at a particular interval.	Publisher
total.events.produced	Total number of events produced for a particular period.	Publisher
published.time	Time taken to publish an event.	Publisher
published.queued.time	Time spent by the batch in publisher queue	Publisher
no.of.disconnects	Number of disconnects happened at that moment.	Publisher
no.of.errors	Number of errors occurred at that moment.	Publisher
batch.size	Size of the batch given in the configuration.	Publisher
batches.resent	Number of batches which are not acknowledged, are sent back to the server.	Publisher
no.of.events.consumed	Number of events consumed at a particular interval.	Consumer
no.of.errors	Number of errors occurred at that moment.	Consumer

Support

Send email to cloudplatform_support@trimble.com