Protocol interface

Each aspect of the mKTL protocol is split into two different components: the client-facing half, and a server-facing half. The interfaces described here are generally not exposed to direct usage, they support the end-user functionality exposed in the Item class.

Message handling is divided into similar phases, and the various methods involved have consistent names to indicate their role in the handling chain.

1.	The layer closest to the transport is defined in incoming or outgoing methods, depending on the direction of the data flow. These are the methods that directly receive or send messages to the transport layer, and provide final interpretation of Message instances. This layer is sufficient for all publish/subscribe handling, and for outbound requests.
2.	Responding to an incoming request starts with a handling layer; the methods in this layer are visible in the request.Server class. req_incoming() reconstructs a Request instance; that instance is passed to req_handler() for dispatch to an appropriate handling endpoint, such as req_get() or req_set(). The Request instance stops here, as well as its Payload, if any.
3.	The last step in the handling chain is a method to perform the request. Methods like perform_get() and perform_set() have a narrower scope, and are intended to be application-specific, implementing core application logic as needed for those items, and do not require awareness of any message handling concepts.

The above steps are not all contained to the protocol interface; the first two layers largely are, though the req_handler() method is subclassed in the Daemon interface code, and endpoints like req_get() or or perform_get() are defined as part of the mktl.Item class. The intent of isolating the application-specific awareness to the last layer is that an application developer can apply their customizations there, without needing to be aware of the other steps in the handling chain.

Payload class

class mktl.protocol.message.Payload(**kwargs)

This is a lightweight class to properly encapsulate a Python-native value for later inclusion in a Message instance. All attributes of this class, except for the bulk attribute, or any attrbiutes set to None, will be added to as a JSON dictionary via encapsulate(), which is called when a Message instance finalizes itself before generating its final on-the-wire representation.

No interpretation of the Payload contents is performed, any interpretation (such as converting a numpy array to a form suitable for enapsulation) must occur before the Payload is instantiated.

Variables:

• bulk – A bulk data value, in bytes, or None

• omit – A set of fields to omit from encapsulation

add_origin()

Add fields to this payload to provide information describing the origin of this message. The primary use case is for debugging or logging, as opposed to uniquely identifying the sender.

encapsulate()

Add all non-omitted local attributes to a dictionary, and return the JSON encoding of that dictionary. For example, if the .value and .time attributes of a Payload are assigned, the caller will receive a value like:

b'{"value": 12, "time": 1761100609234571}'

Message classes

class mktl.protocol.message.Message(type, target=None, payload=None, id=None, flags=None)

The Message provides a very thin encapsulation of what it means to be a message in an mKTL context. This class will be used to represent mKTL messages that do not result in a response.

The fields are largely in order of how they are represented on the wire: the message type, the key/store target for the request, the payload of the message (a Payload instance), and an identification number unique to this correspondence. The identification number is the one field that is out-of-order compared to the multipart sequence on the wire; this is because some message types (publish messages, in particular) do not have an identification number, and it is automatically generated for request messages. Rather than force the caller to pass an explicit None, the id is left as the last field, so that the arguments for all Message instances can have a similar structure.

The type of a base Message instance can only be ‘ACK’ or ‘REP’.

Variables:

• payload – The Payload, if any, for the message.

• valid_types – A set of valid strings for the message type.

_finalize()

Take the contents of this Message, interpet them as bytes, and prepare the tuple that will be used for the multipart transmission on the wire.

log(logger=None, level=10)

Write a debug message describing the contents of this message. The intent is that other sections of code would use this, as needed, to log the contents of a message, rather than bake in direct awareness of all the message and payload fields that might be of interest for debugging.

classmethod reconstruct(parts)

Reconstruct a Message instance from the specified sequence of message parts. This is effectively the inverse of the Message._finalize() method.

class mktl.protocol.message.Broadcast(type, target=None, payload=None, id=None, flags=None)

A Broadcast is a minor variant of a Message, with a change to format the multipart tuple in a PUB/SUB specific fashion.

The type of a Broadcast instance can only be ‘PUB’.

_finalize()

Take the contents of this Message, interpet them as bytes, and prepare the tuple that will be used for the multipart transmission on the wire.

classmethod reconstruct(parts)

Reconstruct a Broadcast instance from the specified sequence of message parts. This is effectively the inverse of the Broadcast._finalize() method.

class mktl.protocol.message.Request(type, target=None, payload=None, id=None, flags=None)

A Request has a little extra functionality, focusing on local caching of response values and signaling that a request is complete. This is the class that will be used on the client side when a server is expected to provide a response, such as returning a requested value, or signaling that a set operation is complete.

The type of a Request instance can only be ‘GET’ or ‘SET’.

Variables:

response – The response (as a Message) to this request

poll()

Return True if the request is complete, otherwise return False.

wait(timeout=60)

Block until the request has been handled. This is a wrapper to a threading.Event instance; if the event has occurred it will return True, otherwise it returns False after the requested timeout. If the timeout argument is None it will block indefinitely.

wait_ack(timeout)

Block until the request has been acknowledged. This is a wrapper to a threading.Event instance; if the event has occurred it will return True, otherwise it returns False after the requested timeout. If the timeout argument is None it will block indefinitely.

wait_sent(timeout)

Block until the request has been sent. This is a wrapper to a threading.Event instance; if the event has occurred it will return True, otherwise it returns False after the requested timeout. If the timeout argument is None it will block indefinitely.

Request client

mktl.protocol.request.client(address, port)

Factory function for a Client instance. Use of this method is encouraged to streamline re-use of established connections.

mktl.protocol.request.send(address, port, request)

Use client() to connect to the specified address and port, and send the specified Request instance. This method blocks until the completion of the request.

class mktl.protocol.request.Client(address, port)

Issue requests via a ZeroMQ DEALER socket and receive responses. Maintains a persistent connection to a single server; the address and port number must be specified.

send(request)

A request is a fully populated Request instance, which normalizes the arguments that will be sent via this method as a multi-part message. The request instance will also be used for notification of any/all responses from the remote end; this method will block while waiting for the ACK request, but will never block waiting for the full response; the caller is free to decide whether to block or wait for the full response, using the methods in the Request instance.

Request server

class mktl.protocol.request.Server(hostname=None, port=None, avoid={})

Receive requests via a ZeroMQ ROUTER socket, and respond to them. The default behavior is to listen for incoming requests on our locally known fully qualified domain name, on the first available automatically assigned port. The avoid set enumerates port numbers that should not be automatically assigned; this is ignored if a fixed port is specified.

The hostname and port variables associated with a Server instance are key pieces of the provenance for an mKTL daemon.

Variables:

• hostname – The hostname on which this server can be contacted.

• port – The port on which this server is listening for connections.

req_ack(request)

Acknowledge the incoming request. The client is expecting an immediate ACK for all request types, including errors; this is how a client knows whether a daemon is online to respond to its request.

req_handler(request)

The default request handler is for debug purposes only, and is effectively a no-op. mktl.Daemon leverages a custom subclass of Server that properly handles specific types of requests, since it needs to be aware of the actual structure of what’s happening in the daemon code.

req_incoming(parts)

All inbound requests are filtered through this method. It will parse the request as JSON into a Python dictionary, and hand it off to req_handler() for further processing. Error handling is managed here; if req_handler() raises an exception it will be packaged up and returned to the client as an error.

req_handler() is expected to call req_ack() to acknowledge the incoming request; if req_handler() is returning a simple payload it will be packged into a REP response. No response will be issued if req_handler() returns None.

respond(request, payload, error)

Issue a response for the specified request. This is broken out as a separate method to allow for background processing of requests.

send(response)

Queue a response to be sent back to the original requestor.

Publish client

mktl.protocol.publish.client(address, port)

Factory function for a Client instance. Use of this method is encouraged to streamline re-use of established connections.

class mktl.protocol.publish.Client(address, port)

Establish a ZeroMQ SUB connection to a ZeroMQ PUB socket and receive broadcasts.

propagate(message)

Invoke any/all callbacks registered via register() for a newly arrived message.

register(callback, topic)

Register a callback that will be invoked every time a new broadcast message arrives. The topic is case-sensitive and must be an exact match. Any callbacks registered in this fashion should be as lightweight as possible, as there is a single thread processing all arriving broadcast messages.

subscribe() will be invoked for any/all topics registered with a callback, it does not need to be called separately.

subscribe(topic)

ZeroMQ subscriptions are based on a topic. Filtering of messages happens on the server side, depending on what a client is subscribed to. A client can subscribe to all messages by providing the empty string as the topic.

unregister(callback, topic)

The inverse of register(), removing a callback from the list of registered callbacks. No errors are raised if the callback is not registered.

Refer to register() for a description of the arguments.

Publish server

class mktl.protocol.publish.Server(port=None, avoid={})

Send broadcasts via a ZeroMQ PUB socket. The default behavior is to set up a listener on all available network interfaces on the first available automatically assigned port. The avoid set enumerates port numbers that should not be automatically assigned; this is ignored if a fixed port is specified.

The port variables associated with a Server instance is a key pieces of the provenance for an mKTL daemon.

Variables:

port – The port on which this server is listening for connections.

publish(broadcast)

A broadcast is a Broadcast instance to send to any/all registered subscribers.