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..."
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
andNoisyPublisher
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
andNoisyPublisher
.
- 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
andSubscriber
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
andNSSubscriber
.
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