Examples

[dsoprea]

Can you post a few examples? There seems to be no documentation, anywhere, on how to use nanomsg from Python.

[dsoprea]

Could I have some pointers on using the "bus" pattern?

I tried to emulate the example at https://github.com/dysinger/nanomsg-examples#bus, but it's not working quite right.

I establish a BUS socket, wait a moment (so I can manually start the other instances), and then connect to the other instances. After connection, we emit a message, and then go into a receive loop. For some reason, I receive two of every message from the other nodes.

#!/usr/bin/env python3

import time
import argparse
import pickle
import collections

import nanomsg

# Message types.

_MT_HELLO = 'hello'
_MT_HELLO_RESPONSE = 'hello_response'

# Miscellaneous constants.

_DESCRIPTION = "Establish a node on a bus"

# Types

_MESSAGE = collections.namedtuple(
            '_MESSAGE',
            ['type', 'node', 'text'])

def _parse_args():
    parser = argparse.ArgumentParser(description=_DESCRIPTION)

    parser.add_argument(
        'name',
        help='The name of this node')

    parser.add_argument(
        'node_uri',
        help='The URI to bind for this node')

    parser.add_argument(
        '-p', '--peer-uri',
        action='append',
        default=[],
        help='The URI of a peer to connect to')

    args = parser.parse_args()

    return args

def _build_message(type_name, node_name, text_template, *args):
    text = text_template % args

    return pickle.dumps((type_name, node_name, text))

def _parse_message(encoded_message):
    parts = pickle.loads(encoded_message)
    return _MESSAGE(*parts)

def _process_messages(s, node_name):
    print("Entering receive loop.")

    while 1:
        encoded_message = s.recv()
        message = _parse_message(encoded_message)

        print("Received message of type [%s] from node [%s]: [%s]" % 
              (message.type, message.node, message.text))

def _start_server(node_name, node_uri, peer_uri_list):
    print("Starting server: [%s] [%s]" % (node_name, node_uri))

    s = nanomsg.Socket(nanomsg.BUS)

    # Configure the local port.

    s.bind(node_uri)

    # Impose a delay to allow the other instances to get to this point.

    print("Bound. Waiting to connect.")
    time.sleep(10)

    # Connect to other nodes.

    for i, peer_uri in enumerate(peer_uri_list):
        print("Connecting (%d): %s" % (i, peer_uri))
        s.connect(peer_uri)

    # Impose a delay to allow the other nodes to connect.

    print("Waiting for all connections to complete.")
    time.sleep(5)

    # Send initial message.

    print("Sending hello.")

    encoded_message = \
        _build_message(
            _MT_HELLO, 
            node_name, 
            'Hello.')

    s.send(encoded_message)

    try:
        _process_messages(s, node_name)
    finally:
        s.close()

if __name__ == '__main__':
    args = _parse_args()
    _start_server(args.name, args.node_uri, args.peer_uri)

Output:

Starting server: [node1] [tcp://0.0.0.0:1234]
Bound. Waiting to connect.
Connecting (0): tcp://127.0.0.1:1235
Connecting (1): tcp://127.0.0.1:1236
Connecting (2): tcp://127.0.0.1:1237
Waiting for all connections to complete.
Sending hello.
Entering receive loop.
Received message of type [hello] from node [node2]: [Hello.]
Received message of type [hello] from node [node2]: [Hello.]
Received message of type [hello] from node [node3]: [Hello.]
Received message of type [hello] from node [node3]: [Hello.]
Received message of type [hello] from node [node4]: [Hello.]
Received message of type [hello] from node [node4]: [Hello.]

[tersfeld]

Hi !

I have the same problem as you, did you manage to make it work in the end ?

Thanks :)

[dsoprea]

Nope. I was relying on someone in the project to actually want to do this.

Recommend using a different solution as this project is unmaintained and, as it's considered "promising" rather than "finished", I'd classify it as more of a PoC and too high-risk for serious consideration.

#43