PostTroll

PostTroll is a message system for pytroll.

A typical use is for event-driven production chains, using messages for notifications.

To get the software, take a look on github.

Use Example

The main use of this library is the posttroll.message.Message, posttroll.subscriber.Subscribe and posttroll.publisher.Publish classes, but the nameserver script is also necessary. The nameserver scripts allows to register data publishers and then for the subscribers to find them. Here is the usage of the nameserver script:

usage: nameserver [-h] [-d {start,stop,status,restart}] [-l LOG] [-v]

optional arguments:
  -h, --help            show this help message and exit
  -d {start,stop,status,restart}, --daemon {start,stop,status,restart}
                        Run as a daemon
  -l LOG, --log LOG     File to log to (defaults to stdout)
  -v, --verbose         print debug messages too
  --no-multicast        disable address broadcasting via multicasting

So, after starting the nameserver, making two processes communicate is fairly easy. Here is an example of publishing code:

from posttroll.publisher import Publish
from posttroll.message import Message
import time

try:
    with Publish("a_service", 9000) as pub:
        counter = 0
        while True:
            counter += 1
            message = Message("/counter", "info", str(counter))
            print "publishing", message
            pub.send(str(message))
            time.sleep(3)
except KeyboardInterrupt:
    print "terminating publisher..."

And the subscribing code:

from posttroll.subscriber import Subscribe

with Subscribe("a_service", "counter",) as sub:
    for msg in sub.recv():
        print msg

There is also a threaded container for the listener that can be used eg. inside a class for continuously monitoring incoming messages:

from posttroll.publisher import NoisyPublisher
from posttroll.listener import ListenerContainer
from posttroll.message import Message
import time

pub = NoisyPublisher("test")
pub.start()
sub = ListenerContainer(topics=["/counter"])
# Wait that both sub and pub to register to nameserver
time.sleep(3)
for counter in range(5):
    msg_out = Message("/counter", "info", str(counter))
    pub.send(str(msg_out))
    print "published", str(msg_out)
    msg_in = sub.output_queue.get(True, 1)
    print "received", str(msg_in), ""
pub.stop()
sub.stop()

If you do not want to broadcast addresses via multicasting to nameservers in your network, you can start the nameserver with the argument –no-multicast. Doing that, you have to specify the nameserver(s) explicitly in the publishing code:

from posttroll.publisher import Publish
from posttroll.message import Message
import time

try:
    with Publish("a_service", 9000, nameservers=['localhost']) as pub:
        counter = 0
        while True:
            counter += 1
            message = Message("/counter", "info", str(counter))
            print "publishing", message
            pub.send(str(message))
            time.sleep(3)
except KeyboardInterrupt:
    print "terminating publisher..."

See also

posttroll.publisher.Publish and posttroll.subscriber.Subscribe

Setting TCP keep-alive

If the network connection between a publisher and a subscriber seem to be dropping, it is possible to set TCP keep-alive settings via environment variables. Below are some rudimentary example values:

import os

os.environ["POSTTROLL_TCP_KEEPALIVE"] = "1"
os.environ["POSTTROLL_TCP_KEEPALIVE_CNT"] = "10"
os.environ["POSTTROLL_TCP_KEEPALIVE_IDLE"] = "1"
os.environ["POSTTROLL_TCP_KEEPALIVE_INTVL"] = "1"

These values need to be set before any subscriber/publisher are created to have them take any effect. Another option is to set these in the shell initialization, like $HOME/.bashrc.

For further information on the 0MQ TCP keep-alive, see zmq_setsockopts for relevant socket options.

Converting from older posttroll versions

Migrating from older versions of posttroll (pre v0.2), so some adaptations have to be made. Instead of data types, the services now have aliases. So, for the publishing, the following call:

with Publish("a_service", ["data_type1", "data_type2"], 9000) as pub:

would translate into:

with Publish("a_service", 9000, ["data_type1", "data_type2"]) as pub:

On the subscriber side, the following:

with Subscribe("data_type1") as sub:

would have to be changed to:

with Subscribe("a_service") as sub:

Note that the behaviour is changed: all the messages comming from the publisher a_service would be iterated over, including messages that have another data type than the one you want. This is why there is now the possibility to add a subject filter directly inside the posttroll.subscriber.Subscribe call:

with Subscribe("a_service", "data_type1") as sub:

This means that the subjects of the messages you are interested in should start with “data_type1” though…

API

Publisher

The publisher module gives high-level tools to publish messages on a port.

class posttroll.publisher.NoisyPublisher(name, port=0, aliases=None, broadcast_interval=2, nameservers=None, min_port=None, max_port=None)

Same as a Publisher, but with broadcasting of its own name and address.

Setting the name to a meaningful value is import since it will be searchable in the nameserver. The port is to be provided as an int, and setting to 0 means it will be set to a random free port. aliases is a list of alternative names for the process. broadcast_interval, in seconds (2 by default) says how often the current name and address should be broadcasted. If nameservers is non-empty, multicasting will be deactivated and the publisher registers on these nameservers only

close()

Alias for stop.

send(msg)

Send a msg.

start()

Start the publisher.

stop()

Stop the publisher.

class posttroll.publisher.Publish(name, port=0, aliases=None, broadcast_interval=2, nameservers=None, min_port=None, max_port=None)

The publishing context.

See Publisher and NoisyPublisher for more information on the arguments.

The publisher is selected based on the arguments, see :function:`create_publisher_from_dict_config` for information how the selection is done.

Example on how to use the Publish context:

from posttroll.publisher import Publish
from posttroll.message import Message
import time

try:
    with Publish("my_service", port=9000) as pub:
        counter = 0
        while True:
            counter += 1
            message = Message("/counter", "info", str(counter))
            print("publishing", message)
            pub.send(message.encode())
            time.sleep(3)
except KeyboardInterrupt:
    print("terminating publisher...")

class posttroll.publisher.Publisher(address, name='', min_port=None, max_port=None)

The publisher class.

address is the current address of the Publisher, e.g.:

tcp://localhost:1234

Setting the port to 0 means that a random free port will be chosen for you. It is still possible to limit the range from which the port is selected by either setting environment variables POSTTROLL_PUB_MIN_PORT and POSTTROLL_PUB_MAX_PORT, or passing the values, as integers, using arguments min_port and max_port when creating the Publisher.

name is simply the name of the publisher.

An example on how to use the Publisher:

from posttroll.publisher import Publisher, get_own_ip
from posttroll.message import Message
import time

pub_address = "tcp://" + str(get_own_ip()) + ":9000"
pub = Publisher(pub_address)

try:
    counter = 0
    while True:
        counter += 1
        message = Message("/counter", "info", str(counter))
        pub.send_string(message.encode())
        time.sleep(3)
except KeyboardInterrupt:
    print("terminating publisher...")
    pub.stop()

bind()

close()

Alias for stop.

heartbeat(min_interval=0)

Send a heartbeat … but only if min_interval seconds has passed since last beat.

send(msg)

Send the given message.

start()

Start the publisher.

stop()

Stop the publisher.

posttroll.publisher.create_publisher_from_dict_config(settings)

Create a publisher based on dictionary of configuration items.

The publisher is created based on the given options in the following way:

• setting settings[‘port’] to non-zero integer AND settings[‘nameservers’] to False will disable nameserver connections and address broadcasting, and publish the messages only on the localhost on the given port

• setting settings[‘nameservers’] to a list of hostnames will connect to nameservers running on those servers, and in addition publish the messages on a random port on the localhost

• setting settings[‘port’] to zero and settings[‘nameservers’] to None will broadcast the publisher address and port with multicast, and publish the messages on a random port.

The last two cases will require settings[‘name’] to be set. Additional options are described in the docstrings of the respective classes, namely Publisher and NoisyPublisher.

posttroll.publisher.get_own_ip()

Get the host’s ip number.

Subscriber

Simple library to subscribe to messages.

class posttroll.subscriber.NSSubscriber(services='', topics='pytroll:/', addr_listener=False, addresses=None, timeout=10, translate=False, nameserver='localhost')

Automatically subscribe to services.

Automatic subscriptions are done by requesting addresses from the nameserver. If topics are specified, filter the messages through the beginning of the subject. addr_listener allows to add new services on the fly as they appear on the network. Additional addresses to subscribe to can be specified, and address translation can be performed if translate is set to True (False by default). The timeout here is specified in seconds. The nameserver tells which host should be used for nameserver requests, defaulting to “localhost”.

Note: ‘services = None’, means no services, and ‘services =””’ means all services. Default is to listen to all services.

close()

Alias for stop.

start()

Start the subscriber.

stop()

Stop the subscriber.

class posttroll.subscriber.Subscribe(services='', topics='pytroll:/', addr_listener=False, addresses=None, timeout=10, translate=False, nameserver='localhost', message_filter=None)

Subscriber context.

See NSSubscriber and Subscriber for initialization parameters.

The subscriber is selected based on the arguments, see :function:`create_subscriber_from_dict_config` for information how the selection is done.

Example:

from posttroll.subscriber import Subscribe

with Subscribe("a_service", "my_topic",) as sub:
    for msg in sub.recv():
        print(msg)

class posttroll.subscriber.Subscriber(addresses, topics='', message_filter=None, translate=False)

Class for subscribing to message streams.

Subscribes to addresses for topics, and perform address translation of translate is true. The function message_filter can be used to discriminate some messages on the subscriber side. topics on the other hand performs filtering on the publishing side (from zeromq 3).

Example:

from posttroll.subscriber import Subscriber, get_pub_address

addr = get_pub_address(service, timeout=2)
sub = Subscriber([addr], 'my_topic')
try:
    for msg in sub(timeout=2):
        print("Consumer got", msg)

except KeyboardInterrupt:
    print("terminating consumer...")
    sub.close()

add(address, topics=None)

Add address to the subscribing list for topics.

It topics is None we will subscibe to already specified topics.

add_hook_pull(address, callback)

Specify a PULL callback in the same stream (thread) as the main receive loop.

The callback will be called with the received messages from the specified subscription. Good for pushed ‘inproc’ messages from another thread.

add_hook_sub(address, topics, callback)

Specify a SUB callback in the same stream (thread) as the main receive loop.

The callback will be called with the received messages from the specified subscription.

Good for operations, which is required to be done in the same thread as the main recieve loop (e.q operations on the underlying sockets).

property addresses

Get the addresses.

close()

Close the subscriber: stop it and close the local subscribers.

recv(timeout=None)

Receive, optionally with timeout in seconds.

remove(address)

Remove address from the subscribing list for topics.

stop()

Stop the subscriber.

property subscribers

Get the subscribers.

update(addresses)

Update with a set of addresses.

posttroll.subscriber.create_subscriber_from_dict_config(settings)

Get a subscriber class instance defined by a dictionary of configuration options.

The subscriber is created based on the given options in the following way:

• setting settings[‘addresses’] to a non-empty list AND settings[‘nameserver’] to False will disable nameserver connections and connect only the listed addresses

• setting settings[‘nameserver’] to a string will connect to nameserver running on the indicated server

• if settings[‘nameserver’] is missing or None, the nameserver on localhost is assumed.

Additional options are described in the docstrings of the respective classes, namely Subscriber and NSSubscriber.

Messages

A Message goes like: <subject> <type> <sender> <timestamp> <version> [mime-type data]

Message('/DC/juhu', 'info', 'jhuuuu !!!')

will be encoded as (at the right time and by the right user at the right host):

pytroll://DC/juhu info henry@prodsat 2010-12-01T12:21:11.123456 v1.01 application/json "jhuuuu !!!"

Note: It’s not optimized for BIG messages.

class posttroll.message.Message(subject='', atype='', data='', binary=False, rawstr=None)

A Message.

• Has to be initialized with a rawstr (encoded message to decode) OR

• Has to be initialized with a subject, type and optionally data, in which case:

  • It will add a few extra attributes.

  • It will make a Message pickleable.

static decode(rawstr)

Decode a raw string into a Message.

encode()

Encode a Message to a raw string.

property head

Return header of a message (a message without the data part).

property host

Try to return a host from a sender.

property user

Try to return a user from a sender.

exception posttroll.message.MessageError

This modules exceptions.

posttroll.message.datetime_decoder(dct)

Decode datetimes to python objects.

posttroll.message.datetime_encoder(obj)

Encodes datetimes into iso format.

posttroll.message.is_valid_data(obj)

Check if data is JSON serializable.

posttroll.message.is_valid_sender(obj)

Currently we only check for empty strings.

posttroll.message.is_valid_subject(obj)

Currently we only check for empty strings.

posttroll.message.is_valid_type(obj)

Currently we only check for empty strings.

Address receiver

Receive broadcasted addresses in a standard pytroll Message.

It will look like: /<server-name>/address info … host:port

class posttroll.address_receiver.AddressReceiver(max_age=datetime.timedelta(seconds=600), port=None, do_heartbeat=True, multicast_enabled=True, restrict_to_localhost=False)

General thread to receive broadcast addresses.

get(name='')

Get the address(es).

is_running()

Check if the receiver is alive.

start()

Start the receiver.

stop()

Stop the receiver.

posttroll.address_receiver.getaddress

alias of AddressReceiver

Name server

Manage other’s subscriptions.

Default port is 5557, if $NAMESERVER_PORT is not defined.

class posttroll.ns.NameServer(max_age=datetime.timedelta(seconds=600), multicast_enabled=True, restrict_to_localhost=False)

The name server.

run(*args)

Run the listener and answer to requests.

stop()

Stop the name server.

exception posttroll.ns.TimeoutError

A timeout.

posttroll.ns.get_active_address(name, arec)

Get the addresses of the active modules for a given publisher name.

posttroll.ns.get_pub_address(name, timeout=10, nameserver='localhost')

Get the address of the publisher for a given publisher name from the nameserver on nameserver (localhost by default).

posttroll.ns.get_pub_addresses(names=None, timeout=10, nameserver='localhost')

Get the address of the publisher for a given list of publisher names from the nameserver on nameserver (localhost by default).

Multicasting

Context

class posttroll.message_broadcaster.AddressBroadcaster(name, address, interval, nameservers)

Class to broadcast stuff.

class posttroll.message_broadcaster.MessageBroadcaster(msg, port, interval, designated_receivers=None)

Class to broadcast stuff.

If interval is 0 or negative, no broadcasting is done.

is_running()

Are we running.

start()

Start the broadcasting.

stop()

Stop the broadcasting.

posttroll.message_broadcaster.sendaddress

alias of AddressBroadcaster

Multicast code

Send/receive UDP multicast packets.

Requires that your OS kernel supports IP multicast.

This is based on python-examples Demo/sockets/mcast.py

class posttroll.bbmcast.MulticastReceiver(port, mcgroup='225.0.0.212')

Multicast receiver on port for an mcgroup.

BUFSIZE = 1024

close()

Close the receiver.

settimeout(tout=None)

A timeout will throw a ‘socket.timeout’.

class posttroll.bbmcast.MulticastSender(port, mcgroup='225.0.0.212')

Multicast sender on port and mcgroup.

close()

Close the sender.

posttroll.bbmcast.SocketTimeout

alias of TimeoutError

posttroll.bbmcast.mcast_receiver(port, mcgroup='225.0.0.212')

Open a UDP socket, bind it to a port and select a multicast group.

posttroll.bbmcast.mcast_sender(mcgroup='225.0.0.212')

Non-object interface for sending multicast messages.

Connections

Misc

Posttroll packages.

posttroll.get_context()

Provide the context to use.

This function takes care of creating new contexts in case of forks.

posttroll.strp_isoformat(strg)

Decode an ISO formatted string to a datetime object.

Allow a time-string without microseconds.

We handle input like: 2011-11-14T12:51:25.123456

Indices and tables

• Index

• Module Index

• Search Page