テスト用のサンプル

changed to dontwait

    def on_message(self, *args):
        print('{:.3f}: on_message'.format(elapsed_time()))
        try:
            message = self._socket.recv(nnpy.DONTWAIT)
            print('{:.3f}: subscribe> {!r}'.format(elapsed_time(), message))
        except nnpy.NNError as exc:
            if exc.error_no == nnpy.EAGAIN:
                print('{:.3f}: subscribe> got EAGAIN. skip...'.format(elapsed_time()))
            else:
                raise

"""
>>>
0.001: publish> even0
0.001: on_message
0.002: subscribe> b'even0'
1.004: publish> odd 1
1.005: on_message
1.005: subscribe> got EAGAIN. skip...
2.006: publish> even2
2.006: on_message
2.006: subscribe> b'even2'
3.010: publish> odd 3
3.010: on_message
3.010: subscribe> got EAGAIN. skip...
4.015: publish> even4
4.015: on_message
4.015: subscribe> b'even4'
5.020: publish> odd 5
5.020: on_message
5.020: subscribe> got EAGAIN. skip...
6.021: publish> even6
6.021: on_message
6.021: subscribe> b'even6'
7.022: publish> odd 7
7.022: on_message
7.022: subscribe> got EAGAIN. skip...
8.025: publish> even8
8.025: on_message
8.025: subscribe> b'even8'
9.031: publish> odd 9
9.031: on_message
9.031: subscribe> got EAGAIN. skip...
"""

test_nanornado.py

# -*- coding:utf-8 -*-
"""
'even', 'odd 'の2つのtopicで5個づつメッセージをpublishする
subscriberはtopicに空文字列を指定しているので、全てのメッセージを受け取る。これはちゃんと動く

>>>
0.001: publish> even0
0.001: on_message
0.001: subscribe> b'even0'
1.006: publish> odd 1
1.006: on_message
1.006: subscribe> b'odd 1'
2.011: publish> even2
2.011: on_message
2.011: subscribe> b'even2'
3.016: publish> odd 3
3.017: on_message
3.017: subscribe> b'odd 3'
4.021: publish> even4
4.021: on_message
4.021: subscribe> b'even4'
5.024: publish> odd 5
5.024: on_message
5.025: subscribe> b'odd 5'
6.030: publish> even6
6.030: on_message
6.030: subscribe> b'even6'
7.032: publish> odd 7
7.032: on_message
7.032: subscribe> b'odd 7'
8.034: publish> even8
8.034: on_message
8.034: subscribe> b'even8'
9.037: publish> odd 9
9.037: on_message
9.037: subscribe> b'odd 9'
"""
import time

import nnpy
from tornado import gen, ioloop


def elapsed_time():
    return time.time() - start_time
start_time = time.time()


class Subscriber(object):
    def __init__(self, socket_url, topic, timeout=None):
        self._socket = nnpy.Socket(nnpy.AF_SP, nnpy.SUB)
        self._socket.connect(socket_url)
        self._socket.setsockopt(nnpy.SUB, nnpy.SUB_SUBSCRIBE, topic)
        if timeout:
            self._socket.setsockopt(nnpy.SOL_SOCKET, nnpy.RCVTIMEO, timeout)

    def close(self):
        """Tornadoに叩かれる."""
        self._socket.close()

    def fileno(self):
        """Tornadoに叩かれる."""
        return self._socket.getsockopt(nnpy.SOL_SOCKET, nnpy.RCVFD)

    def register(self):
        ioloop.IOLoop.current().add_handler(self, self.on_message, ioloop.IOLoop.READ)

    def on_message(self, *args):
        print('{:.3f}: on_message'.format(elapsed_time()))
        message = self._socket.recv()
        print('{:.3f}: subscribe> {!r}'.format(elapsed_time(), message))


@gen.coroutine
def main(socket_url):
    publisher = nnpy.Socket(nnpy.AF_SP, nnpy.PUB)
    publisher.bind(socket_url)

    subscriber = Subscriber(socket_url, '', timeout=5000)
    subscriber.register()

    for i in range(10):
        message = '{}{}'.format(
            'odd ' if i % 2 else 'even',
            i
        )
        print('publish> {}'.format(message))
        publisher.send(message)
        yield gen.sleep(1)

ioloop.IOLoop.current().run_sync(lambda: main('ipc:///tmp/test.ipc'))