console

$> python ./test-notification-server.py

$> python test-notification.py --config-file ./test.conf --debug
$> git reset --hard 5315399
$> python test-notification-old.py --config-file ./test.conf --debug

test-notification-old.py

import eventlet

eventlet.monkey_patch(os=False)

import sys

from nova import config
from nova import context
from nova.openstack.common import log as logging
from nova.openstack.common.notifier import api as notifier_api

config.parse_args(sys.argv)

logging.setup("nova")

notifier_api.notify(context.RequestContext('user', 'project', is_admin=True),
                    'compute.blaa', 'foo', notifier_api.INFO, {})

test-notification-server.py

class Kombu(object):

    def __init__(self):

        self.params = {
            'hostname': 'localhost',
            'port': 5672,
            'virtual_host': '/',
            }

        self.exchange_name = 'nova'
        self.topic = 'notifications.info'

    def connect(self):
        import kombu.connection
        import kombu.entity

        self.connection = kombu.connection.BrokerConnection(**self.params)
        self.connection.connect()

        self.channel = self.connection.channel()

        exchange = kombu.entity.Exchange(name=self.exchange_name,
                                         type='topic',
                                         durable=False,
                                         auto_delete=False)

        queue = kombu.entity.Queue(name=self.topic,
                                   exchange=exchange,
                                   routing_key=self.topic,
                                   channel=self.channel,
                                   durable=False,
                                   auto_delete=False,
                                   exclusive=False)
        queue.declare()

        queue.consume(consumer_tag=str(1),
                      nowait=False,
                      callback=self.callback)

    def callback(self, raw_message):
        message = self.channel.message_to_python(raw_message)
        print message.payload
        message.ack()

    @property
    def sock(self):
        return self.connection.connection.sock

    def process_pending(self):
        self.connection.drain_events(timeout=0)


import select
import socket
import errno

k = Kombu()
k.connect()

p = select.poll()
p.register(k.sock, select.POLLIN|select.POLLPRI)

while True:
    try:
        k.process_pending()
    except socket.error as e:
        if e.errno != errno.EAGAIN:
            raise e
        p.poll()

test-notification.py

import eventlet

eventlet.monkey_patch(os=False)

import sys

from nova import config
from nova import context
from nova.openstack.common import log as logging
from nova import rpc

config.parse_args(sys.argv)

logging.setup("nova")

notifier = rpc.get_notifier(service='compute', host='blaa')
notifier.info(context.RequestContext('user', 'project', is_admin=True), 'foo', {})

test.conf

[DEFAULT]
rpc_backend = nova.openstack.common.rpc.impl_kombu
rabbit_host = localhost
rabbit_port = 5672
rabbit_virtual_host = /
control_exchange = nova

notification_driver = nova.openstack.common.notifier.rpc_notifier
notification_topics = notifications