Cfengine3 Messaging

> what would be killer is a combination of the latest cfengine
> (with promise theory) and 0mq.

Source: http://lists.zeromq.org/pipermail/zeromq-dev/2011-February/009491.html

This part of the Webhuis CFEngine initiative is research and therefore subject to change. :-)
It is the objective to have messaging incorporated in CFEngine and thus enabling nodes, by cf-agent that is, to send out messages to whatever listeners out there. The listeners can be cf-messages daemons or other nodes configured to process messages delivered by CFEngine Nodes.
CFEngine will need to implement a new promise type: messages:.
Parts of the works is to be transferred to and discussions will take place on:

https://github.com/Webhuis/Data

So let's go and deliver that kill!

0mq

The search for a suitable messaging mechanism eventually led to 0mq, just a library not an application. Could it be less in weight?
For reasons of symplicity the code examples, so far are in Python. CFEngine is independent of anything but libc and thus the end results of CFEngine messaging have to be developed in C.

Basic patterns

ZeroMQ comes with five basic patterns

• Synchronous Request/Response
• Asynchronous Request/Response
• Publish/Subscribe
• Push/Pull
• Exclusive Pair

Nodes carry different kinds of information

• Common information
• Static information
• Dynamic Information
• State Information

In the analysis use cases of types of messages has to be mapped on the basic patterns. The listeners mirror the nodes mappings.

Synchronous Request/Response

Client Requests

socket = context.socket(zmq.REQ)
socket.connect("tcp://wbhs-pkg.webhuis.nl:8080")

#  Do 10 requests, waiting each time for a response
print("Sending request %s ..." % request)
socket.send(b"Hello")

#  Get the reply.
message = socket.recv()
print("Received reply %s [ %s ]" % (request, message))

Server Replies

socket = context.socket(zmq.REP)
socket.bind("tcp://10.68.71.184:8080")

#  Wait for next request from client
message = socket.recv()
print("Received request: %s" % message)

#  Send reply back to client
socket.send(b"World")

Publish/Subscribe

Server publishes

ctx = zmq.Context()
publisher = ctx.socket(zmq.PUB)

publisher.bind("tcp://10.68.71.184:5556")

time.sleep(5)

sequence = 0
id = 0
data =1000

while sequence < 20:
  id += 1;
  data += 1000;
  publisher.send("%i %i %i" % (sequence, id, data));
  sequence += 1

Client subscriber receives

context = zmq.Context()
socket = context.socket(zmq.SUB)

print("Collecting updates from server...")
socket.connect("tcp://10.68.71.184:5556")

socket.setsockopt(zmq.SUBSCRIBE, b'')

for update_nbr in range(5):
    string = socket.recv_string()
    print string

Output

0 1 2000
1 2 3000
2 3 4000
3 4 5000
4 5 6000

Exclusive Pair

---

Return to: Cfengine