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
• 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.
• Supporting Authentication with TID system

Installation

This package is available on Trimble Articatory PyPI registry. Run the following command to install the package:

pip install TrimbleCloud.Events

Prerequisites

For an application to access the Events server:

• The Application must be registered in the API cloud.
• Provide the Client Credential in the client library configuration.

NOTE: Client Credentials will not be transferred to Events Servers. Library internally uses the credentials to get the Authorization token from TID and set it in the Websocket connection Header to authenticate with Events Server.

Client Creation

Authentication

Events client library supports authentication with TID system. It uses Client Credentials grant type to get the access token from TID.

To connect to the Events Service, the STG/PROD TID client credentials (client_id, client_secret, auth_url) must be provided in the configuration. Ensure that valid credentials are provided, if not an exception will be thrown.

Client credentials will be used only to authenticate with TID, these credentials are not transferred to the Events server.

Client configuration

You can create a client by passing the following parameters to the client constructor.

import TrimbleCloud.Events as events

client = events.client(
            endpoint="<<publisher_server_endpoint>> || <<consumer_server_endpoint>>",
            client_id="<<client_id>>",
            client_secret="<<client_secret>>",
            tid_scope="<<tid_scope>>",
            auth_url="<<tid_oauth_url>>",
            enable_batch=False,
            call_back=custom_publisher_callback,
            workflow=Workflow.PUBLISHER || Workflow.CONSUMER,
            retry_count=30,
            enable_metrics=False
        )

Please refer this for more information about client configurations.

Parameter	Description	Required	Default Value	Example	Workflow
endpoint	Events server endpoint	Yes	None	wss://endpoint:443	Publisher/Consumer
client_id	Client ID of the application	Yes	None	****	Publisher/Consumer
client_secret	Client Secret of the application	Yes	None	****	Publisher/Consumer
tid_scope	Scope of the application	Yes	None	application_name	Publisher/Consumer
auth_url	TID OAuth URL	Yes	None	https://id.trimble.com/oauth/token	Publisher/Consumer
enable_batch	Enable batching of events. To enable internal batching. Note: If it is set as False, batching and batch retry has to be handled by the client	No	False	False	Publisher
call_back	Callback function to handle the events.	No	def custom_callback(response, exception)	Publisher/Consumer
workflow	Workflow of the client.	Yes	None	Workflow.PUBLISHER	Publisher/Consumer
retry_count	Number of times to retry the connection if it is lost or interrupted.	No	5	5	Publisher/Consumer
enable_metrics	Enable metrics to Datadog.	No	False	False	Publisher/Consumer
client_acknowledge	When enabled, auto acknowledgment of the poll is disabled and the client has to call the acknowledge method on the Response object received in the callback.	No	False	False	Consumer

To know connection status

To know the connection status, you can use the following methods.

>>> status = client.active() # Returns True if the connection is established, else False
>>> print(status)

To check state of the client connection

To check the state of the client, you can use the following methods.

>>> state = client.connection_status() # Status.RUNNING - connection is established and active. Status.CONNECTING - Connection establishment in progress or retrying connection on anomalies. Status.STOPPED - connection closed either due to close call or connection retry limit reached

>>> print(state)

Close the connection

To close the connection, you can use the following methods.

>>> client.close() # Closes the connection

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.

Connection retry

Events client library supports connection retry. It will retry to connect to the Events server if the connection is lost or interrupted.

• Provide the retry_count in the configuration. Default value is 200.
• 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 (retry_count + 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.

Enabling metrics

Metrics can be sent to datadog by enabling the enable_mertics property to true while creating the client.

ens_client = ens.client(
           endpoint="endpoint",
           client_id="client_id",
           client_secret="client_secret",
           auth_url="auth_url",
           enable_batch=False,
           call_back=custom_callback,
           workflow=Workflow.PUBLISHER,
           retry_count=1,
           enable_metrics=True,
           datadog_api_key="datadog_api_key",
           datadog_application_key="datadog_application_key",
           datadog_metrics_tags={
               "env": "python_test",
               "instance": (socket.gethostbyname(socket.gethostname()))
           },
           metric_recorder="data_dog"
       )

Parameter	Description	Required	Default Value	Example
enable_metrics	Enable metrics to Datadog.	No	False	True
datadog_api_key	Datadog API key	Required if enable_metrics is True	None	****
datadog_application_key	Datadog Application key	Required if enable_metrics is True	None	****
datadog_metrics_tags	Datadog metrics tags	Required if enable_metrics is True	None	{ “env”: “python_test”, “instance”: (socket.gethostbyname(socket.gethostname())) }
metric_recorder	Provide metric recorder type.Currently supports DATA_DOG_RECORDER	Required if enable_metrics is True	None	data_dog

The Below Metrics are sent to DataDog.

Metric	Description
no.of.events.produced	Total number of events produced
no.of.events.consumed	Total number of events consumed
rate.produced	Number of events produced for a particular period
no.of.disconnects	Number of disconnects happened at any moment
no.of.errors	Number of errors occurred at any moment

Publisher

A Publisher sends events through a websocket connection to Events Servers. To Publish events, Publishers must need a:

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

For more information about Namespaces, Event Schemas and Events, please refer to the following documentation. https://docs.trimblecloud.com/events/content/Concepts/publishers/

Publishing batch of events

Follow the below steps to configure Events client and publish events

1. Import events module

>>> import TrimbleCloud.Events as events

2. Configure event client with below client configurations

client = events.client(
            endpoint="<<publisher_server_endpoint>>",
            client_id="<<client_id>>",
            client_secret="<<client_secret>>",
            auth_url="<<tid_oauth_url>>",
            enable_batch=False,
            call_back=custom_publisher_callback,
            workflow=Workflow.PUBLISHER,
            retry_count=30,
            enable_metrics=False
        )

Please see here for more information about publisher client configurations.

3. Call send method to publish batch of events.

payload = json.dumps({
               "id": str(count),
               "specversion": "1.0",
               "source": "/tid/<apiid>",
               "type": "delete.v1",
               "time": "{{isoTimestamp}}",
               "subject": "/tid/UserDelete",
               "data": {
                   "uuid": "121212",
                   "event": "123",
                   "namespace": "namespace",
                   "message": "message 1 from produce"
               }
           })

# Configure the Publisher Event Request.
publish_event = PublishEvent(
    namespace="com.trimble.tcp.sample",
    event="SampleEvent",
    entity_id="ab",
    event_version="v1.0",
    payload=payload
)

events_batch= EventsBatch(id="id")

events_batch.events.append(publish_event)

# Check for client status before send.
if client.active:
    client.send(events_batch)
else:
    print("Client status: " + str(client.status) + ",Batch not published: " + event_batch.to_json())

Publishes events to the Events server. It takes EventsBatch as an argument.

4. Acknowledgement will be received in the callback function.

def custom_publisher_callback(response, exception):
    if exception is not None:
        print(exception)
    else:
        print("Batch ID: " + response.batch_id)
        print("Batch type: " + response.type)
        print("Batch published: " + response.to_json())

The response object contains the following properties and method.

• batch_id - Batch id received from the server
• type - Type of the response. It can be “Acknowledgement” or “Error
• to_json() - Returns the response object as JSON string.

Please note send throws following exceptions.

• Connection Already Closed
• Connecting (initial connection or retry in progress)
• Any other client Exception on sending batch

Please see here for more information about Exceptions.

Recommendation

• Before calling send method check if the client is active.
• Check the status to know the appropriate state of the client, if the client is not active.

EventsBatch

EventsBatch is a collection of events. It has the following properties.

Property	Description	Type	Example	Required
id	Unique identifier for the batch.	String	12345678-1234-1234-1234-123456789012	Yes
events	List of events. Can contain more than one event of type PublishEvent. Cannot be empty.	list	[Event1, Event2]	Yes

PublishEvent

PublishEvent is an event that is published to the Events server. It has the following properties.

Property	Description	Type	Example	Required
namespace	Namespace of the event.	String	com.trimble.tcp.sample	Yes
event	Name of the event.	String	sampleEvent	Yes
event_version	Version of the event.	String	1.0	Yes
payload	Publisher Payload .It should be JSON encoded, and should match the Avro Schema of the respective Event.	Object	{ “key”: “value” }	Yes
entity_id	Unique identifier for the event.	String	12345678-1234-1234-1234-123456789012	No

Events Ordering

Events Servers take the best efforts to deliver the events in the order in which it is received. Clients should have logic within messages to manage strict ordering. To send sequential events use entity_id. For example, user profile id to maintain order in edits on user information. For more information about ordering, reach out to the Events team.

Comprehensive Publisher Example Code

import json
import logging
import socket
import threading
import time
import uuid
from collections import OrderedDict
# importing metrics, publish event and workflow from the ens module.
import TrimbleCloud.Events as ens
from TrimbleCloud.Events.Models.EventsBatch import EventsBatch
from TrimbleCloud.Events.Models.PublishEvent import PublishEvent
from TrimbleCloud.Events.Models.Workflow import Workflow


format_string = "%(asctime)s %(name)s [%(levelname)s] %(message)s"

formatter = logging.Formatter(format_string)

root_handler = logging.StreamHandler()
root_handler.setFormatter(formatter)

root_logger = logging.getLogger()
root_logger.setLevel(logging.INFO)
root_logger.addHandler(root_handler)

logging.getLogger('ens').setLevel(logging.INFO)
logging.getLogger('datadog').setLevel(logging.CRITICAL)

logger = logging.getLogger(__name__)
logging.getLogger(__name__).setLevel(logging.INFO)

# Dictionary to store Batch ID and Thread Event mapping.
ack_wait_dict = OrderedDict()

# Callback function to send response and exception from Events server.
def custom_callback(response, exception):
   if exception is not None:
       logger.error(exception)
       pass
   else:
       e = ack_wait_dict.pop(response.batch_id)
       e.set() # The event for the particular batchId is set True once the ack is received
       pass



# Main function to Publish Events.
def main():
   request_ts = time.monotonic()
   rps = 100
   send_timeout = 10  # time to wait for ack

   try:
       logger.info('Starting producer')

       # Client configuration
       ens_client = ens.client(
           endpoint="endpoint",
           client_id="client_id",
           client_secret="client_secret",
           auth_url="auth_url",
           enable_batch=False,
           call_back=custom_callback,
           workflow=Workflow.PUBLISHER,
           retry_count=1,
           enable_metrics=True,
           datadog_api_key="datadog_api_key",
           datadog_application_key="datadog_application_key",
           datadog_metrics_tags={
               "env": "python_test",
               "instance": (socket.gethostbyname(socket.gethostname()))
           },
           metric_recorder="data_dog"
       )
       # Initial Batch creation.The batch can contain one or more events. If the batch contains only one event then batch id and event id have to be the same.
       event_batch = EventsBatch(id=str(uuid.uuid4().urn)[9:])
       count = 0

       while True:
           count += 1

           # Event payload.
           payload = json.dumps({
               "id": str(count),
               "specversion": "1.0",
               "source": "/tid/<apiid>",
               "type": "delete.v1",
               "time": "{{isoTimestamp}}",
               "subject": "/tid/UserDelete",
               "data": {
                   "uuid": "121212",
                   "event": "123",
                   "namespace": "namespace",
                   "message": "message 1 from produce"
               }
           })

           # Configure the Publisher Event Request.
           publish_event = PublishEvent(
                namespace="com.trimble.tcp.sample",
                event="SampleEvent",
                entity_id="ab",
                event_version="v1.1",
                payload=payload
           )

           # Only 1 event is attached to each Batch
           event_batch.events.append(publish_event)

           # Check client status before send
           if ens_client.active:

               """
                   An event is created for each batch it is stored in ack_wait_dict           (OrderedDict) as (batchId, event)
                   A new thread is created for each send call and it wait until the ack is received or the timeout is reached
               """
               e = threading.Event()
               ack_wait_dict[event_batch.id] = e
               threading.Thread(name=event_batch.id,
                                target=send_wait,
                                args=(e, send_timeout, event_batch, ens_client.send)).start()

           else:
               logger.debug(
                   "Client status: " + str(ens_client.status) + ", Batch not published: " + event_batch.to_json())

           # Refresh batch after send
           event_batch = EventsBatch(id=str(uuid.uuid4().urn)[9:])

           request_ts += (1.0 / rps)
           now = time.monotonic()
           if now < request_ts:
               time.sleep(request_ts - now)
               pass

       # Close connection returns unpublished events, if any
       events = ens_client.close()
       logger.info(json.dumps(events, default=obj_dict))

   except Exception as error:
       logger.exception("Fatal error in test-producer-new")


def obj_dict(obj):
   return obj.to_dict()


def send_wait(e, timeout, event_batch, send):
   """
       e: event
       timeout: time to wait for acknowledgement
       event_batch: the batch that has to sent
       send: client send method
   """
   logger.debug("Sending batch id : " + str(event_batch.id))
   send(event_batch)
   logger.debug("Waiting on batch id : " + str(event_batch.id))
   event_is_set = e.wait(timeout)
   if event_is_set:
       # When event_is_set is True, batch is sent successfully and ack has been received.
       logger.debug("Released on batch id : " + str(event_batch.id))
   else:
       # When event_is_set is false, ack has not been received within timeout.
       # Batch retry should be done here
       logger.info("Timeout on batch id : " + str(event_batch.id))


if __name__ == '__main__':
   main()

Consumer

A Consumer consumes events from the 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.

Consuming events

Follow the below steps to create Consumer and consume events,

1. Import events module

>>> import TrimbleCloud.Events as events

2. Configure event consumer client with below client configurations

client = events.client(
            endpoint="<<consumer_server_endpoint>>",
            client_id="<<client_id>>",
            client_secret="<<client_secret>>",
            tid_scope="<<tid_scope>>",
            auth_url="<<tid_oauth_url>>",
            client_acknowledge=False,
            call_back=custom_consumer_callback,
            workflow=Workflow.CONSUMER,
            retry_count=30,
            enable_metrics=False
        )

Please see here for more information about consumer client configurations.

3. Poll for events

    consume_event = ConsumeEvent(
                        namespace="com.trimble.tcp.sample",
                        event="SampleEvent",
                        event_version="v1.0"
                    )
    client.poll(consume_event)

Please refer here for ConsumeEvent properties.

4. Response will be received in the callback function.

def custom_consumer_callback(response, exception):
    if exception is not None:
        print(exception)
    else:
        print("Consumed event is: " + response.to_json())

ConsumeEvent

The Consumer event is asynchronous. Response will be sent via call back.

Property	Description	Type	Example	Required
namespace	Namespace of the event.	String	com.trimble.tcp.sample	Yes
event	Name of the event.	String	sampleEvent	Yes
event_version	Version of the event.	String	1.0	Yes
visibility_timeout	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	Integer	3000	No

Explicit consumer response acknowledgement

By default, the consumer response is acknowledged automatically. If you want to explicitly acknowledge the response, you can set the client_acknowledge property to True while creating the client.

def custom_callback(response, exception):
    if exception is not None:
        print(exception.to_json())
    elif ens_client.active:
        print("Message received")
        ens_client.acknowledge(response.acknowledgement_id)
     else:
        print("Client status: " + str(ens_client.status) + ",acknowledgment not sent: " + response.to_json())

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 client_acknowledment disabled

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

Long Polling

Long polling is a mechanism where the server holds the request open until new data is available. It reduces the number of requests and improves efficiency.

Consumes events from the ENS server. It takes consumer request (see here) as an argument. Applicable for low-volume events. Allowed poll duration for a long polling request is 60 seconds only. Any other value will be rejected by the server. Requests with additional_fields with the key “long.poll.duartion” and value as 60000 will be considered as a long poll request. 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.

consume_event = ConsumeEvent(
                namespace="com.trimble.tcp.sample",
                event="SampleEvent",
                event_version="v1.0",
                additional_fields={
                   "long.poll.duration": "60000"
                }
            )
ens_client.poll(consume_event)

Exception is sent in callback when

• Connection Already Closed
• Connecting (initial connection or retry in progress)

Poll method throws the following exceptions

• When long poll for the event is already in progress
• Any other client Exception during poll

Comprehensive Consumer Example Code

import logging
# importing metrics, consume events and workflow from the ens module.
import TrimbleCloud.Events.Metrics as ens
from ens.Metrics.Metric import MetricRecorderFactory
from ens.Models.ConsumeEvent import ConsumeEvent
from ens.Models.Workflow import Workflow

format_string = "%(asctime)s %(name)s [%(levelname)s] %(message)s"
formatter = logging.Formatter(format_string)

root_handler = logging.StreamHandler()
root_handler.setFormatter(formatter)

root_logger = logging.getLogger()
root_logger.setLevel(logging.INFO)
root_logger.addHandler(root_handler)

logging.getLogger('ens').setLevel(logging.INFO)
logging.getLogger('datadog').setLevel(logging.CRITICAL)

logger = logging.getLogger(__name__)

# Callback function to send response and Exception from Events server
def custom_callback(response, exception):
    if exception is not None:
        logger.error(exception.to_json())
    else:
        logger.info("Message received")
        logger.debug(response.to_json())

# Function to test the Consuming Events.
def main():

    try:
        logger.info('Starting consumer')

        # Client configuration
        consumer_client = ens.client(
            endpoint="endpoint",
            client_id="client_id",
            client_secret="client_secret",
            auth_url="auth_url",
            call_back=custom_callback,
            workflow=Workflow.CONSUMER,
            retry_count=1,
            enable_metrics=False
        )

        count = 0

        while True:
            count += 1

            # Configure the Consumer Event Request.
            consume_event = ConsumeEvent(
                namespace="com.trimble.tcp.sample",
                event="SampleEvent",
                event_version="v1.1",
            )

            # Poll using Consume Event.
            consumer_client .poll(consume_event)
            logger.debug('Poll request sent')
  # Close the websocket connection.
  ens_client.close()

    except Exception as error:
        logging.exception("Fatal error in test-consumer")

# Driver program to test main function
if __name__ == '__main__':
    main()
    pass

Exceptions

Capture the following exceptions in the callback and re-raise it in the main thread loop. Since the callback is run on a separate thread and does not impact the main thread.

Exception	Description
ConnectionClosedException	Connection is already closed. ConnectionClosedException is thrown in following cases, 1. If publish is called, after connection close 2. If the send method is called when connection is closed or during connection retry.
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) if the Publish_message method is called when enable_batch is set False during client creation. Batch with empty events is passed in the send method.
ValueError	ValueError is thrown in the following cases. Mandatory parameter missing Parameter type mismatch Parameter value not within limits
Exception	Any other exception is thrown in the following cases. Error occurred during TID oauth token request TID Token generation failed Unsupported recorder for Datadog
WebsocketException	Websocket Exception is thrown if Connection retry fails.
ConsumerException	Consumer Exception is 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

Support

If you have any questions or issues, please contact the Trimble Cloud Platform team at cloudplatform_support@trimble.com