Module `connection`

This documentation contains proprietary and confidential information of Cloud Software Group, Inc. Use, disclosure, or reproduction is prohibited without the prior express written permission of Cloud Software Group, Inc.

Classes

class Eftl

Programs use Eftl Class to connect to an eFTL server.

Initialize the Eftl instance.

Class variables

var CONNECTED
var CONNECTING
var DISCONNECTED
var DISCONNECTING
var RECONNECTING

Methods

async def connect(self, url, **kwargs)

Connect to an eFTL server.

This call returns immediately; connecting continues asynchronously. When the connection is ready to use, the eFTL library calls your on_connect method, passing a EftlConnection object that you can use to publish and subscribe.

When a pipe-separated list of URLs is specified this call will attempt a connection to each in turn, in a random order, until one is connected. A program that uses more than one server channel must connect separately to each channel.

:param: url: The call connects to the eFTL server at this URL. This can be a single URL, or a pipe ('|') separated list of URLs. URLs can be in either of these forms

    ws://host:port/channel
    wss://host:port/channel

Optionally, the URLs can contain the username, password,
and/or client identifier

    ws://username:password@host:port/channel?clientId=<identifier>
    wss://username:password@host:port/channel?clientId=<identifier

:param: kwargs auto_reconnect_attempts : int, optional Maximum number of reconnect attempts. The default is 256 attempts. auto_reconnect_max_delay : optional Maximum reconnect delay in milliseconds. The default is 30 seconds. max_pending_acks : optional Maximum number of unacknowledged messages allowed for the client. user : optional Login credentials to use if not found in the url. password : optional Login credentials to use if not found in the url. client_id : optional User-specified client identifier. handshake_timeout : optional Seconds to wait for websocket handshake to complete. login_timeout : optional Seconds to halt waiting for a login message reply. If a reply is not received in time, raise an EftlClientError. polling_interval : optional Seconds to wait between each message reply check. trust_all : true/false (optional) trust_store : certificate path (optional) loop : event loop provided by user (optional) Callbacks : on_connect(connection): on_disconnect(connection, loop, code, reason): on_reconnect(connection): on_error(connection, code, reason): on_state_change(connection, state): :return: The EftlConnection object that can used to publish and subscribe Raises

ValueError: If any provided urls lack a host name or the correct scheme.
EftlClientError: If the connection is not established in time.

def get_version(self)

Return the eFTL Python library version.

Returns: str: The eFTL Python library version

class EftlAlreadyConnected (*args, **kwargs)

Raise when connect method is called but connection is already connected.

Ancestors

builtins.Exception
builtins.BaseException

class EftlClientError (*args, **kwargs)

Raise when establishing a connection to the server fails.

Ancestors

builtins.Exception
builtins.BaseException

class EftlConnection

A websocket wrapper exchanging eftl protocol messages with a server.

Different functions implement each of the types of protocol messages (e.g. publish, subscribe, disconnect). The messages are sent asynchronously, but by default these functions halt waiting for a reponse. The user may set a timeout for this waiting period if they desire.

Users also may specify callbacks to trigger when messages get delivered, fail to get delivered, etc.

Initialize the connection object, which is unconnected.

Methods

async def acknowledge(self, message)

Acknowledge this message.

Messages consumed from subscriptions with a client acknowledgment mode must be explicitly acknowledged. The eFTL server will stop delivering messages to the client once the server's configured maximum number of unacknowledged messages is reached.

:param message: The message being acknowledged :return:

async def acknowledge_all(self, message)

Acknowledge all messages up to and including this message.

:param message: The message being acknowledged1 :return:

async def close_all_subscriptions(self)

Close all the subscriptions on this connection.

For durable subscriptions, this call will cause the persistence service to stop delivering messages while leaving the durable subscriptions to continue accumulating messages. Any unacknowledged messages will be made available for redelivery.

Programs receive subscription identifiers through their on_subscribe method.

async def close_subscription(self, subscription_id)

Close subscription for specified subscription identifier.

Programs receive subscription identifiers through their on_subscribe method.

async def create_kv_map(self, name)

Return a new EftlKVMap associated with this connection.

:param name: name of the KVMap :return: The KVmap object

def create_message(self)

Create an EftlMessage that can be used to publish or send request/replies.

:return: The EftlMessage object

async def disconnect(self)

Disconnect from the eFTL server.

Programs may disconnect to free server resources.

This call returns immediately; disconnecting continues asynchronously.

When the connection has closed, the eFTL library calls your on_disconnect callback.

:return:

def get_clientId(self)

Get the client identifier for this connection.

:return: The client's identifier.

def is_connected(self)

Determine whether this connection to the eFTL server is open or closed.

:return: True if this connection is open, False otherwise

async def publish(self, message, **kwargs)

Publish a one-to-many message to all subscribing clients.

This call returns immediately; publishing continues asynchronously. When the publish completes successfully, the eFTL library calls your

:param message: Publish this message. :param kwargs: callbacks: on_publish on_error :return:

ValueError If the message size exceeds the maximum. TypeError If the message is not of type EftlMessage ConnectionError If the connection is closed.

async def reconnect(self)

Reopen a closed connection.

You may call this method within your on_disconnect method.

This call returns immediately; connecting continues asynchronously. When the connection is ready to use, the eFTL library calls your on_reconnect callback

Reconnecting automatically re-activates all subscriptions on the connection. The eFTL library invokes your on_subscribe callback for each successful resubscription.

:return:

async def remove_kv_map(self, name)

Remove a EftlKVMap object.

:param name: the name of the map :return:

async def send_reply(self, requestMessage, replyMessage, **kwargs)

Send a reply message in response to a request message.

This call returns immediately. When the send completes successfully the eFTL library calls your {@link CompletionListener#onCompletion} callback.

:param requestMessage: The reply message to send :param replyMessage: The request msg :param kwargs: callbacks :return:

async def send_request(self, message, timeout, **kwargs)

Publish a request message.

This call returns immediately. When the reply is received the eFTL library calls your {@link RequestListener#onReply} callback.

:param message: The request message to publish. :param timeout: timeout seconds to wait for reply :param kwargs: callbacks: :return:

async def subscribe(self, **kwargs)

Send a message to register a matcher-based subscription.

Parameters

timeout : optional: Number of seconds to halt waiting on acknowledgement from the server (default set in EftlConnection.connect()).
matcher : str, optional: JSON content matcher to subscribe to.
type : {"standard", "shared", "last-value"}, optional: Durable type.
key : optional: The last-value index key, if type is "last-value".
durable : str, optional: Name to give the durable.
ack_mode : client, None. Default is auto

Returns

sub_id: The subcription identifier of the new subscription.

async def unsubscribe(self, sub_id)

Unsubscribe from messages on a subscription.

For durable subscriptions, this call will cause the persistence service to remove the durable subscription, along with any persisted messages.

Programs receive subscription identifiers through their methods.

:param sub_id: Subscription identifier of the subscription to delete. timeout` :return:

async def unsubscribe_all(self)

Unsubscribe from all subscriptions.

Parameters

timeout : optional: Number of seconds to halt waiting on acknowledgement from the server (default set in EftlConnection.connect()).

class EftlKVMap (connection, name)

A key-value map object represents a program's connection to an FTL map.

Programs use key-value map objects to set, get, and remove key-value pairs in an FTL map.

Methods

async def get(self, key, **kwargs): Get the value of a key from the map, or null if the key is not set.

:param key: Get the value for this key :param kwargs: Callbacks: on_success: on_error: :return: The value as a EftlMessage
async def remove(self, key, **kwargs): Remove a key-value pair from the map.

This call returns immediately; removing continues asynchronously. When the remove completes successfully, the eFTL library calls your on_success callback.

:param key: Revove the value for this key. :param kwargs: on_success on_error :return:
async def set(self, key, value, **kwargs): Set a key-value pair in the map, overwriting any existing value.

This call returns immediately; setting continues asynchronously. When the set completes successfully, the eFTL library calls your on_success callback.

:param key: Set the value for this key. :param value: Set this value for the key. :param kwargs: on_success on_error :return:

class EftlMessage (obj=None)

Programs use message objects to publish messages on an eFTL connection or receive messages for a given subscription.

Message objects contain typed fields that map names to values.

Initialize the EftlMessage instance properties.

Methods

def clear_all_fields(self)

Remove all the fields in this message.

def clear_field(self, field_name)

Remove the given field from this message.

:param field_name:

def get_datetime(self, field_name)

Get the value of a long field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type datetime

def get_datetime_array(self, field_name)

Get the value of a long field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type datetime array

def get_delivery_count(self)

Message's delivery count assigned by the persistence service.

:return: The message delivery count.

def get_double(self, field_name)

Get the value of a long field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type double

def get_double_array(self, field_name)

Get the value of a long field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type double array

def get_field_names(self)

Return the list of field names.

:return: The field names of this message as a list object.

def get_field_type(self, field_name)

Return the type of value of this field.

:param field_name: The name of the field :return: type of value of this field.

Possible return values : 
        'int' represents Integer
        'str' represents String
        'float' represents Float
        'datetime.datetime' represents datetime.datetime 
        'message' represents EftlMessage object type
        'opaque' represents opaque type (bytes)
        'unknown'

def get_long(self, field_name)

Get the value of a long field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type long

def get_long_array(self, field_name)

Get the value of a long array field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type long array

def get_message(self, field_name)

Get the value of a long field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type message

def get_message_array(self, field_name)

Get the value of a message array field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type message array (list of messages)

def get_opaque(self, field_name)

Get the value of opaque field from a message.

:param field_name: The name of the field :return: The value of the field if the field is present

def get_store_message_id(self)

Message's unique store identifier assigned by the persistence service.

:return: A monotonically increasing long value that represents message's unique store identifier

def get_string(self, field_name)

Get the value of a string field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type string

def get_string_array(self, field_name)

Get the value of a string array field from a message.

:param field_name: The name of the field

:return: The value of the field if the field is present and has type string array

def is_field_set(self, field_name)

Return True if the field is set, False otherwise.

:param field_name: The name of the field :return: True if the field is set, False otherwise

def set_datetime(self, field_name, value)

Set a datetime field in a message.