def __init__(self, event_type=None, xp=None, process=None, **kwargs):
        """
        Constructs a publisher of events for a specific type.

        @param  event_type  The name of the event type object
        @param  xp          Exchange (AMQP) name, can be none, will use events default.
        """

        self.event_type = event_type
        self.process = process
        self._events_xp = CFG.get_safe("exchange.core.events", DEFAULT_EVENTS_XP)

        if bootstrap.container_instance and getattr(bootstrap.container_instance, 'event_repository', None):
            self.event_repo = bootstrap.container_instance.event_repository
        else:
            self.event_repo = None

        # generate an exchange name to publish events to
        container = (hasattr(self, '_process') and hasattr(self._process, 'container') and self._process.container) or BaseEndpoint._get_container_instance()
        if container and container.has_capability(container.CCAP.EXCHANGE_MANAGER):   # might be too early in chain
            xp = xp or container.create_xp(self._events_xp)
            to_name = xp
        else:
            xp = xp or self.get_events_exchange_point()
            to_name = (xp, None)

        Publisher.__init__(self, to_name=to_name, **kwargs)

    def test_last_update_cache(self):
        handle = self.start_worker()
        queue = Queue()
        o_process = handle.process
        def new_process(msg):
            o_process(msg)
            queue.put(True)
        handle.process = new_process



        definition = SBE37_CDM_stream_definition()
        publisher = Publisher()

        stream_def_id = self.pubsub_cli.create_stream_definition(container=definition)
        stream_id = self.pubsub_cli.create_stream(stream_definition_id=stream_def_id)

        time = float(0.0)

        for granule in self.make_points(definition=definition, stream_id=stream_id, N=10):

            publisher.publish(granule, to_name=(self.XP, stream_id+'.data'))
            # Determinism sucks
            try:
                queue.get(timeout=5)
            except Empty:
                self.assertTrue(False, 'Process never received the message.')

            doc = self.db.read(stream_id)
            ntime = doc.variables['time'].value
            self.assertTrue(ntime >= time, 'The documents did not sequentially get updated correctly.')
            time = ntime

    def __init__(self, xp=None, **kwargs):

        # generate a name
        xp = xp or get_events_exchange_point()
        name = (xp, None)

        Publisher.__init__(self, name=name, **kwargs)

 def launch_benchmark(transform_number=1, primer=1,message_length=4):
     import gevent
     from gevent.greenlet import Greenlet
     from pyon.util.containers import DotDict
     from pyon.net.transport import NameTrio
     from pyon.net.endpoint import Publisher
     import uuid
     num = transform_number
     msg_len = message_length
     transforms = list()
     pids = 1
     TransformBenchTesting.message_length = message_length
     cc = Container.instance
     pub = Publisher(to_name=NameTrio(get_sys_name(),str(uuid.uuid4())[0:6]))
     for i in xrange(num):
         tbt=cc.proc_manager._create_service_instance(str(pids), 'tbt', 'prototype.transforms.linear', 'TransformInPlace', DotDict({'process':{'name':'tbt%d' % pids, 'transform_id':pids}}))
         tbt.init()
         tbt.start()
         gevent.sleep(0.2)
         for i in xrange(primer):
             pub.publish(list(xrange(msg_len)))
         g = Greenlet(tbt.perf)
         g.start()
         transforms.append(tbt)
         pids += 1

    def test_sub(self):

        #start interaction observer
        io = InteractionObserver()
        io.start()

        #publish an event
        ev_pub = EventPublisher(event_type="ResourceEvent")
        ev_pub.publish_event(origin="specific", description="event")


        # publish a message
        msg_pub = Publisher()
        msg_pub.publish(to_name='anyone', msg="msg")

        # give 2 seconds for the messages to arrive
        time.sleep(2)

        #verify that two messages (an event and a message) are seen
        self.assertEquals(len(io.msg_log), 2)

        #iterate through the messages observed
        for item in io.msg_log:
            # if event
            if item[2]:
                #verify that the origin is what we sent
                self.assertEquals(item[1]['origin'], 'specific')
        dump = io._get_data(io.msg_log,{})
        sump = dump

    def test_xp_durable_send(self):
        xp = self.container.ex_manager.create_xp('an_xp')
        #self.addCleanup(xp.delete)

        xq = self.container.ex_manager.create_xn_queue('no_matter', xp)
        self.addCleanup(xq.delete)
        xq.bind('one')

        pub = Publisher(to_name=xp.create_route('one'))
        pub.publish('test')
        pub.close()


        try:
            url = self.container.ex_manager._get_management_url("queues", "%2f", xq.queue, "get")
            res = self.container.ex_manager._make_management_call(url,
                                                                  use_ems=False,
                                                                  method='post',
                                                                  data=json.dumps({'count':1, 'requeue':True,'encoding':'auto'}))

            self.assertEquals(len(res), 1)
            self.assertIn('properties', res[0])
            self.assertIn('delivery_mode', res[0]['properties'])
            self.assertEquals(2, res[0]['properties']['delivery_mode'])

        except Exception, e:
            # Rabbit 3.x does not support this command anymore apparently.
            self.assertIn('Method Not Allowed', e.message)

    def test_sub(self):
        ar = event.AsyncResult()
        def cb(*args, **kwargs):
            ar.set(args)

        sub = ConvSubscriber(callback=cb)
        pub = Publisher()
        self._listen(sub)
        pub.publish(to_name='anyone', msg="hello")


        evmsg, evheaders = ar.get(timeout=5)
        self.assertEquals(evmsg, "hello")
        self.assertAlmostEquals(int(evheaders['ts']), int(get_ion_ts()), delta=5000)

    def test_async_result(self):
        request_id = "request_foo"
        waiter = AsyncResultWaiter()
        self.assertFalse(waiter.async_res.ready())
        token = waiter.activate()
        self.assertFalse(waiter.async_res.ready())
        log.info("Wait token: %s", token)

        pub = Publisher(to_name=token)
        async_msg = AsyncResultMsg(request_id=request_id)
        pub.publish(async_msg)

        res = waiter.await(timeout=1, request_id=request_id)
        self.assertTrue(waiter.async_res.ready())
        self.assertIsInstance(res, AsyncResultMsg)
        self.assertEqual(res.__dict__, async_msg.__dict__)

    def __init__(self, xp=None, event_repo=None, **kwargs):
        """
        Constructs a publisher of events.

        @param  xp          Exchange (AMQP) name, can be none, will use events default.
        @param  event_repo  An optional repository for published events. If None, will not store
                            published events. Use the Container.event_repository for this
                            parameter if you have one.
        """

        # generate a name
        xp = xp or get_events_exchange_point()
        name = (xp, None)

        self.event_repo = event_repo

        Publisher.__init__(self, to_name=name, **kwargs)

class TestPublisher(PyonTestCase):
    def setUp(self):
        self._node = Mock(spec=NodeB)
        self._pub = Publisher(node=self._node, name="testpub")
        self._ch = Mock(spec=SendChannel)
        self._node.channel.return_value = self._ch

    def test_publish(self):
        self.assertEquals(self._node.channel.call_count, 0)

        self._pub.publish("pub")

        self._node.channel.assert_called_once_with(self._pub.channel_type)
        self.assertEquals(self._ch.send.call_count, 1)

        self._pub.publish("pub2")
        self._node.channel.assert_called_once_with(self._pub.channel_type)
        self.assertEquals(self._ch.send.call_count, 2)

class ContainerHeartbeater(object):
    """ Utility class that implements the container heartbeat publishing mechanism """
    def __init__(self, container, cfg):
        self.container = container
        self.heartbeat_cfg = cfg
        self.started = False

    def start(self):
        from pyon.net.endpoint import Publisher
        from pyon.util.async import spawn
        self.heartbeat_quit = Event()
        self.heartbeat_interval = float(self.heartbeat_cfg.get("publish_interval", 60))
        self.heartbeat_topic = self.heartbeat_cfg.get("topic", "heartbeat")
        self.heartbeat_pub = Publisher(to_name=self.heartbeat_topic)

        # Directly spawn a greenlet - we don't want this to be a supervised IonProcessThread
        self.heartbeat_gl = spawn(self.heartbeat_loop)
        self.started = True
        log.info("Started container heartbeat (interval=%s, topic=%s)", self.heartbeat_interval, self.heartbeat_topic)

    def stop(self):
        if self.started:
            self.heartbeat_quit.set()
            self.heartbeat_gl.join(timeout=1)
            self.started = False

    def heartbeat_loop(self):
        self.publish_heartbeat()
        while not self.heartbeat_quit.wait(timeout=self.heartbeat_interval):
            self.publish_heartbeat()

    def publish_heartbeat(self):
        try:
            hb_msg = self.get_heartbeat_message()
            headers = dict(expiration=60000)
            self.heartbeat_pub.publish(hb_msg, headers=headers)
        except Exception:
            log.exception("Error publishing heatbeat")

    def get_heartbeat_message(self):
        from interface.objects import ContainerHeartbeat
        hb_msg = ContainerHeartbeat(container_id=self.container.id, ts=get_ion_ts())
        return hb_msg

    def __init__(self, event_type=None, xp=None, **kwargs):
        """
        Constructs a publisher of events for a specific type.

        @param  event_type  The name of the event type object
        @param  xp          Exchange (AMQP) name, can be none, will use events default.
        """

        self.event_type = event_type

        if bootstrap.container_instance and getattr(bootstrap.container_instance, 'event_repository', None):
            self.event_repo = bootstrap.container_instance.event_repository
        else:
            self.event_repo = None

        # generate an exchange name to publish events to
        xp = xp or get_events_exchange_point()
        name = (xp, None)

        Publisher.__init__(self, to_name=name, **kwargs)

    def start(self):
        from pyon.net.endpoint import Publisher
        from pyon.util.async import spawn
        self.heartbeat_quit = Event()
        self.heartbeat_interval = float(self.heartbeat_cfg.get("publish_interval", 60))
        self.heartbeat_topic = self.heartbeat_cfg.get("topic", "heartbeat")
        self.heartbeat_pub = Publisher(to_name=self.heartbeat_topic)

        # Directly spawn a greenlet - we don't want this to be a supervised IonProcessThread
        self.heartbeat_gl = spawn(self.heartbeat_loop)
        self.started = True
        log.info("Started container heartbeat (interval=%s, topic=%s)", self.heartbeat_interval, self.heartbeat_topic)

    def on_start(self):
        TransformDataProcess.on_start(self)

        # set up subscriber to *
        self._bt_sub = Subscriber(callback=lambda m, h: self.call_process(m),
                                  from_name=NameTrio(get_sys_name(), 'bench_queue', '*'))

        # spawn listener
        self._sub_gl = spawn(self._bt_sub.listen)

        # set up publisher to anything!
        self._bt_pub = Publisher(to_name=NameTrio(get_sys_name(), str(uuid.uuid4())[0:6]))

class HeartBeater(object):
    def __init__(self, CFG, factory, log=logging):

        self._log = log
        self._log.log(logging.DEBUG, "Starting the heartbeat thread")
        self._CFG = CFG
        self._res = None
        self._interval = int(CFG.eeagent.heartbeat)
        self._res = None
        self._done = False
        self._factory = factory
        self._next_beat(datetime.datetime.now())
        self._publisher = Publisher()
        self._pd_name = CFG.eeagent.get('heartbeat_queue', 'heartbeat_queue')

        self._factory.set_state_change_callback(self._state_change_callback, None)

    def _next_beat(self, now):
        self._beat_time = now + datetime.timedelta(seconds=self._interval)

    def _state_change_callback(self, user_arg):
        # on state change set the beat time to now
        self._beat_time = datetime.datetime.now()

    def poll(self):

        now = datetime.datetime.now()
        if now > self._beat_time:
            self._next_beat(now)
            self.beat()

    def beat(self):
        try:
            beat = make_beat_msg(self._factory, self._CFG)
            message = dict(beat=beat, resource_id=self._CFG.agent.resource_id)
            to_name = self._pd_name
            self._log.debug("Send heartbeat: %s to %s", message, self._pd_name)
            self._publisher.publish(message, to_name=to_name)
        except:
            self._log.exception("beat failed")

    def test_sub(self):

        # publish 2 messages
        pub = Publisher()
        pub.publish(to_name='anyone', msg="hello1")
        pub.publish(to_name='anyone', msg="hello2")

        dsm = self.container.datastore_manager
        ds = dsm.get_datastore("conversations")

        # give at least 2 seconds for the persister to save in the repository
        # test may fail if it does not wait long enough for the persister
        no_of_conv = 0
        retried = 0

        while (no_of_conv != 2 and retried < 5):
            time.sleep(2)
            # assert that the 2 messages have been persisted
            no_of_conv = len(ds.list_objects())
            retried = retried + 1

        self.assertEquals(no_of_conv, 2)

    def __init__(self, process, stream, **kwargs):
        """
        Creates a StreamPublisher which publishes to the specified stream
        and is attached to the specified process.
        @param process   The IonProcess to attach to.
        @param stream    Name of the stream or StreamRoute object
        """
        super(StreamPublisher, self).__init__()
        if not isinstance(process, BaseService):
            raise BadRequest("No valid process provided.")
        if isinstance(stream, basestring):
            self.stream_route = StreamRoute(routing_key=stream)
        elif isinstance(stream, StreamRoute):
            self.stream_route = stream
        else:
            raise BadRequest("No valid stream information provided.")

        self.container = process.container
        self.xp_name = get_streaming_xp(self.stream_route.exchange_point)   # Fully qualified

        self.xp = self.container.ex_manager.create_xp(self.stream_route.exchange_point or DEFAULT_DATA_XP)
        self.xp_route = self.xp.create_route(self.stream_route.routing_key)

        Publisher.__init__(self, to_name=self.xp_route, **kwargs)

    def __init__(self, CFG, factory, log=logging):

        self._log = log
        self._log.log(logging.DEBUG, "Starting the heartbeat thread")
        self._CFG = CFG
        self._res = None
        self._interval = int(CFG.eeagent.heartbeat)
        self._res = None
        self._done = False
        self._factory = factory
        self._next_beat(datetime.datetime.now())
        self._publisher = Publisher()
        self._pd_name = CFG.eeagent.get('process_dispatcher', 'processdispatcher')

        self._factory.set_state_change_callback(self._state_change_callback, None)

    def __init__(self, CFG, factory, process_id, process, log=logging):

        self._log = log
        self._log.log(logging.DEBUG, "Starting the heartbeat thread")
        self._CFG = CFG
        self._res = None
        self._interval = int(CFG.eeagent.heartbeat)
        self._res = None
        self._done = False
        self._started = False
        self._factory = factory
        self.process = process
        self.process_id = process_id
        self._publisher = Publisher()
        self._pd_name = CFG.eeagent.get('heartbeat_queue', 'heartbeat_queue')

        self._factory.set_state_change_callback(self._state_change_callback, None)
        self._first_beat()

 def __init__(self, process=None, **kwargs):
     self._process = process
     Publisher.__init__(self, **kwargs)