habitat.message_server: the Server and its Sinks

The message server pushes each received Message to one more Sinks.

class habitat.message_server.Server(program)

The Server is the main message server class.

This is the class in which the magic happens. A Server manages the loading, unloading and reloading of ‘Sinks’, and pushes messages to each and every sink when message() is called.

program: a habitat.main.Program object

start()

Starts up the server.

load(new_sink)

Loads the specified sink

new_sink: can be a class, or a string, e.g., "myprogram.sinks.my_sink", where myprogram.sinks is a module and my_sink is a class inside that module

find_sink(sink)

Locates a currently loaded sink in the Server

sink: either the class, or the name of the class, of the sink to locate.

unload(sink)

Opposite of load(): Removes sink from the Server.

sink: either the class, or the name of a class that has been loaded by a call to load()

reload(sink)

Uses habitat.utils.dynamicloader.load() (with force_reload=True) to reload a sink

This function removes the old sink and adds a new object created from the result of the class reloading.

sink: either the class, or the name of a class that has been loaded by a call to load()

push_message(message)

Pushes a message to all sinks loaded in the server

This method will return instantly, having added the message object to an internal queue for processing by a thread.

message: a habitat.message_server.Message object

flush()

Blocks until the internal queue of messages has been processed

shutdown()

Shuts down the Server

This function calls Sink.shutdown() on every Sink currently loaded, and then empties the list of loaded sink

class habitat.message_server.Sink(server)

Sink is the parent class for all sinks.

To write a sink, have your sink class either inherit from SimpleSink or ThreadedSink.

All sinks have the following self-explanatory methods, all of which update the internal set of message-types which the sink wishes to receive. These functions are for use by the sink class.

• Sink.add_type
• Sink.add_types
• Sink.remove_type
• Sink.remove_types
• Sink.set_types
• Sink.clear_types

They also will have the following functions, which are used internally by the message server

• Sink.push_message()
• Sink.flush()
• Sink.shutdown()

A sink must define these functions:

• setup(): called once; the sink must call some of the self.*type* functions in order to set up the set of types that the sink would like to receive
• message(message): called whenever a message is received for the sink to process

Sinks are automatically initialised by the habitat.message_server.Server that is asked to load them.

server: the Server object that this Sink is now receiving messages from

push_message(message)

Called by the server in order to pass a message to the Sink.

This method is typically implemented by SimpleSink or ThreadedSink. Filtering based on Sink.types is done by this method.

flush()

Ensures that all current calls to push_message() finish

After calling flush(), provided that no calls to push_message() are made in the meantime, no threads will be executing in this Sink‘s push_message() method. The Server has a lock to prevent messages from being pushed while flushing.

shutdown()

Shuts down the Sink

This method flushes the sink, and then cleans up anything that was initalised in the Sink‘s __init__ method (for example, in ThreadedSink it joins the thread).

class habitat.message_server.SimpleSink(server)

A class for light weight, basic sinks to inherit

A sink that has SimpleSink as its parent class must have a message(message) method that conforms to some very strict criteria.

It must:

• be non blocking
• be thread safe (however you can’t use mutexes, these block)
• tolerate multiple calls to message by multiple threads, simultaneously

If the sink wishes to place messages “back into” the server the it must tolerate recursion (i.e., your message method will indirectly call itself).

class habitat.message_server.ThreadedSink(server)

A class for sinks that need to execute in another thread to inherit from.

Each call to message will be executed in a new thread, though if too many threads are currently running for this Sink the call will block until a space is available.

Children can set self._max_workers in setup() to change the maximum number of spawned threads (for instance, set it to 1 to ensure that only one thread is active at once).

Children also get access to self.manager where they can store data that will persist between calls (anything set in self will vanish when the thread terminates). This is only available after setup() - things set on self will be accessible at self.manager (persistant) or just self (temporary) in message().

Yes, this is very confusing. Welcome to threading.

push_message(message)

Spawn a new thread to deal with the incoming message. If too many threads are active, wait for the oldest one to finish.

run()

Process the message that this thread was started with.

flush()

Waits for all running workers to do their stuff.

shutdown()

Wait for all threads to stop before shutting down.

class habitat.message_server.Message(source, type, time_created, time_uploaded, data)

A Message object describes a single message that the server might handle

After initialisation, the data is available in

• message.source
• message.type
• message.time_created
• message.time_uploaded
• message.data

The following message types are available:

• Message.RECEIVED_TELEM: received telemetry string
• Message.LISTENER_INFO: listener information
• Message.LISTENER_TELEM: listener telemetry
• Message.TELEM: (parsed) telemetry data

See also

../messages

source: a Listener object

type: one of the type constants

time_created: the time that the event that eventually caused this Message to be created, e.g., for TELEM and RECEIVED_TELEM, this is the time that the telemetry string was received over the radio. (UNIX Timestamp format)

time_uploaded: the time that habitat received the message. (UNIX Timestamp)

data: a type-specific data object, which will be validated

classmethod validate_type(type)

Checks that type is an integer and a valid message type

classmethod validate_types(types)

Checks that types is a set of valid integer message types

class habitat.message_server.Listener(callsign, ip)

A Listener object describes the source from which a message came.

It has two attributes: callsign and ip. callsign is chosen by the user that created the message, and must be alphanumeric and uppercase. It cannot be “trusted”. ip in typical usage is initalised by the server receiving the message (i.e., where it came from). When comparing two Listener objects (operator overloading), only callsign is considered.

callsign: string, must be composed of alphanumeric and /_- characters only (a-zA-Z0-9/_-)

ip: string, which will be validated and converted to an IPAddress object (the ipaddr module)