def _build_header(self, raw_msg, raw_headers):
        """
        Builds the header for this Process-level RPC conversation.
        https://confluence.oceanobservatories.org/display/syseng/CIAD+COI+OV+Common+Message+Format
        """

        header = EndpointUnit._build_header(self, raw_msg, raw_headers)

        # add our process identity to the headers
        header.update({'sender-name': self._process.name or 'unnamed-process',     # @TODO
                       'sender': self._process.id})

        if hasattr(self._process, 'process_type'):
            header.update({'sender-type': self._process.process_type or 'unknown-process-type'})
            if self._process.process_type == 'service' and hasattr(self.channel, '_send_name'):
                header.update({'sender-service': "%s,%s" % (self.channel._send_name.exchange, self._process.name)})

        context = self.get_context()
        #log.debug('ProcessEndpointUnitMixin._build_header has context of: %s', context)


        # use context to set security attributes forward
        if isinstance(context, dict):
            new_header = self.build_security_headers(context)
            header.update(new_header)
        else:
            # no context? we're the originator of the message then
            container_id                    = BaseEndpoint._get_container_instance().id
            header['origin-container-id']   = container_id

            #This is the originating conversation
            if 'conv-id' in raw_headers:
                header['original-conv-id'] = raw_headers['conv-id']

        return header

    def _build_header(self, raw_msg, raw_headers):
        """
        Builds the header for this Process-level RPC conversation.
        """
        header = EndpointUnit._build_header(self, raw_msg, raw_headers)

        # Add our process identity to the headers (as sender)
        header.update({'sender-name': self._process.name or 'unnamed-process',     # @TODO
                       'sender': self._process.id})

        if hasattr(self._process, 'process_type'):
            header.update({'sender-type': self._process.process_type or 'unknown-process-type'})
            if self._process.process_type == 'service' and hasattr(self.channel, '_send_name'):
                header.update({'sender-service': "%s,%s" % (self.channel._send_name.exchange, self._process.name)})

        # Use received message headers context to set security attributes forward
        context = self.get_context()
        if isinstance(context, dict):
            new_header = self.build_security_headers(context)
            header.update(new_header)
        else:
            # no context? we're the originator of the message then
            container_id = BaseEndpoint._get_container_instance().id
            header['origin-container-id'] = container_id

            # This is the originating conversation
            if 'conv-id' in raw_headers:
                header['original-conv-id'] = raw_headers['conv-id']

        return header

    def _intercept_msg_in(self, inv):
        """
        Override for incoming message interception.

        This is a request, so the order should be Message, Process
        """
        inv_one = EndpointUnit._intercept_msg_in(self, inv)
        inv_two = process_interceptors(interceptors["process_incoming"] if "process_incoming" in interceptors else [], inv_one)
        return inv_two

    def _intercept_msg_out(self, inv):
        """
        Override for outgoing message interception.

        This is request, so the order should be Process, Message
        """
        inv_one = process_interceptors(interceptors["process_outgoing"] if "process_outgoing" in interceptors else [], inv)
        inv_two = EndpointUnit._intercept_msg_out(self, inv_one)

        return inv_two

    def _build_header(self, raw_msg):
        """
        Builds the header for this Process-level RPC conversation.
        https://confluence.oceanobservatories.org/display/syseng/CIAD+COI+OV+Common+Message+Format
        """

        header = EndpointUnit._build_header(self, raw_msg)

        # add our process identity to the headers
        header.update({'sender-name'  : self._process.name or 'unnamed-process',     # @TODO
                       'sender'       : self._process.id })

        if hasattr(self._process,'process_type' ):
            header.update({'sender-type'  : self._process.process_type or 'unknown-process-type' })
            if self._process.process_type == 'service':
                header.update({ 'sender-service' : "%s,%s" % ( self.channel._send_name.exchange,self._process.name) })

        context = self._process.get_context()
        log.debug('ProcessEndpointUnitMixin._build_header has context of: %s', context)

        # use context to set security attributes forward
        if isinstance(context, dict):
            # fwd on actor specific information, according to common message format spec
            actor_id            = context.get('ion-actor-id', None)
            actor_roles         = context.get('ion-actor-roles', None)
            actor_tokens        = context.get('ion-actor-tokens', None)
            expiry              = context.get('expiry', None)
            container_id        = context.get('origin-container-id', None)

            #If an actor-id is specified then there may be other associated data that needs to be passed on
            if actor_id:
                header['ion-actor-id']  = actor_id
                if actor_roles:     header['ion-actor-roles']   = actor_roles
                if actor_tokens:    header['ion-actor-tokens']  = actor_tokens

            if expiry:          header['expiry']                = expiry
            if container_id:    header['origin-container-id']   = container_id
        else:
            # no context? we're the originator of the message then
            container_id                    = BaseEndpoint._get_container_instance().id
            header['origin-container-id']   = container_id

        return header

    def _build_header(self, raw_msg, raw_headers):
        """
        Builds the header for this Process-level RPC conversation.
        https://confluence.oceanobservatories.org/display/syseng/CIAD+COI+OV+Common+Message+Format
        """

        header = EndpointUnit._build_header(self, raw_msg, raw_headers)

        # add our process identity to the headers
        header.update({'sender-name': self._process.name or 'unnamed-process',     # @TODO
                       'sender': self._process.id})

        if hasattr(self._process, 'process_type'):
            header.update({'sender-type': self._process.process_type or 'unknown-process-type'})
            if self._process.process_type == 'service' and hasattr(self.channel, '_send_name'):
                header.update({'sender-service': "%s,%s" % (self.channel._send_name.exchange, self._process.name)})

        context = self.get_context()
        log.debug('ProcessEndpointUnitMixin._build_header has context of: %s', context)

        #Check for a field with the ResourceId decorator and if found, then set resource-id
        # in the header with that field's value or if the decorator specifies a field within an object,
        #then use the object's field value ( ie. _id)
        try:
            if isinstance(raw_msg, IonObjectBase):
                decorator = 'ResourceId'
                field = raw_msg.find_field_for_decorator(decorator)
                if field is not None and hasattr(raw_msg,field):
                    deco_value = raw_msg.get_decorator_value(field, decorator)
                    if deco_value:
                        #Assume that if there is a value, then it is specifying a field in the object
                        fld_value = getattr(raw_msg,field)
                        header['resource-id'] = getattr(fld_value, deco_value)
                    else:
                        header['resource-id'] = getattr(raw_msg,field)

        except Exception, ex:
            log.exception(ex)

    def _build_invocation(self, **kwargs):
        newkwargs = kwargs.copy()
        newkwargs.update({'process':self._process})

        inv = EndpointUnit._build_invocation(self, **newkwargs)
        return inv

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

 def setUp(self):
     self._endpoint_unit = EndpointUnit(interceptors={})

class TestEndpointUnit(PyonTestCase):

    def setUp(self):
        self._endpoint_unit = EndpointUnit(interceptors={})

    def test_attach_channel(self):
        ch = Mock(spec=BaseChannel)
        self._endpoint_unit.attach_channel(ch)

        self.assertTrue(self._endpoint_unit.channel is not None)
        self.assertEquals(self._endpoint_unit.channel, ch)

    @patch('pyon.net.endpoint.get_ion_ts', Mock(return_value=sentinel.ts))
    def test_send(self):

        # need a channel to send on
        self.assertRaises(AttributeError, self._endpoint_unit.send, "fake")

        ch = Mock(spec=SendChannel)
        self._endpoint_unit.attach_channel(ch)

        self._endpoint_unit.send("hi", {'header':'value'})
        ch.send.assert_called_once_with('hi', {'header':'value', 'ts':sentinel.ts})

    def test_close(self):
        ch = Mock(spec=BaseChannel)
        self._endpoint_unit.attach_channel(ch)
        self._endpoint_unit.close()
        ch.close.assert_called_once_with()

    def test_build_header(self):
        head = self._endpoint_unit._build_header({'fake': 'content'}, {})
        self.assertTrue(isinstance(head, dict))

    def test_build_payload(self):
        fakemsg = {'fake':'content'}
        msg = self._endpoint_unit._build_payload(fakemsg, {'fake':'header'})
        self.assertEquals(msg, fakemsg)

    def test_build_msg(self):
        fakemsg = {'fake':'content'}
        msg, headers = self._endpoint_unit._build_msg(fakemsg, {})

        self.assertEquals(msg, fakemsg)
        self.assertEquals(headers, {'ts':ANY})

    def test_intercept_in(self):
        self._endpoint_unit._build_invocation = Mock()
        self._endpoint_unit._intercept_msg_in = Mock()

        self._endpoint_unit.intercept_in(sentinel.msg, sentinel.headers)

        self._endpoint_unit._build_invocation.assert_called_once_with(path=Invocation.PATH_IN,
                                                                      message=sentinel.msg,
                                                                      headers=sentinel.headers)
        self.assertTrue(self._endpoint_unit._intercept_msg_in.called)

    def test__message_received(self):
        self._endpoint_unit.message_received  = Mock()
        self._endpoint_unit.message_received.return_value = sentinel.msg_return

        retval = self._endpoint_unit._message_received(sentinel.msg, sentinel.headers)

        self.assertEquals(retval, sentinel.msg_return)

        self.assertTrue(self._endpoint_unit.message_received.called)

 def __init__(self, opt=None, **kwargs):
     self._opt = opt
     EndpointUnit.__init__(self, **kwargs)

 def setUp(self):
     self._endpoint_unit = EndpointUnit()

class TestEndpointUnit(PyonTestCase):
    def setUp(self):
        self._endpoint_unit = EndpointUnit()

    def test_attach_channel(self):
        ch = Mock(spec=BaseChannel)
        self._endpoint_unit.attach_channel(ch)

        self.assertTrue(self._endpoint_unit.channel is not None)
        self.assertEquals(self._endpoint_unit.channel, ch)

    @patch("pyon.net.endpoint.get_ion_ts", Mock(return_value=sentinel.ts))
    def test_send(self):

        # need a channel to send on
        self.assertRaises(AttributeError, self._endpoint_unit.send, "fake")

        ch = Mock(spec=SendChannel)
        self._endpoint_unit.attach_channel(ch)

        self._endpoint_unit.send("hi", {"header": "value"})
        ch.send.assert_called_once_with("hi", {"header": "value", "ts": sentinel.ts})

    def test_close(self):
        ch = Mock(spec=BaseChannel)
        self._endpoint_unit.attach_channel(ch)
        self._endpoint_unit.close()
        ch.close.assert_called_once_with()

    def test_spawn_listener(self):
        def recv():
            ar = event.AsyncResult()
            ar.wait()

        ch = Mock(spec=BidirClientChannel)
        ch.recv.side_effect = recv
        self._endpoint_unit.attach_channel(ch)

        self._endpoint_unit.spawn_listener()

        self._endpoint_unit.close()
        self.assertTrue(self._endpoint_unit._recv_greenlet.ready())

    def test_build_header(self):
        head = self._endpoint_unit._build_header({"fake": "content"})
        self.assertTrue(isinstance(head, dict))

    def test_build_payload(self):
        fakemsg = {"fake": "content"}
        msg = self._endpoint_unit._build_payload(fakemsg)
        self.assertEquals(msg, fakemsg)

    def test_build_msg(self):
        fakemsg = {"fake": "content"}
        msg = self._endpoint_unit._build_msg(fakemsg)

    #        self.assertTrue(isinstance(msg, dict))
    #        self.assertTrue(msg.has_key('header'))
    #        self.assertTrue(msg.has_key('payload'))
    #        self.assertTrue(isinstance(msg['header'], dict))
    #        self.assertEquals(fakemsg, msg['payload'])

    def test__message_received(self):
        self._endpoint_unit._build_invocation = Mock()
        self._endpoint_unit._intercept_msg_in = Mock()
        self._endpoint_unit.message_received = Mock()
        self._endpoint_unit.message_received.return_value = sentinel.msg_return

        retval = self._endpoint_unit._message_received(sentinel.msg, sentinel.headers)

        self.assertEquals(retval, sentinel.msg_return)

        self._endpoint_unit._build_invocation.assert_called_once_with(
            path=Invocation.PATH_IN, message=sentinel.msg, headers=sentinel.headers
        )
        self.assertTrue(self._endpoint_unit._intercept_msg_in.called)
        self.assertTrue(self._endpoint_unit.message_received.called)

class TestEndpointUnit(PyonTestCase):

    def setUp(self):
        self._endpoint_unit = EndpointUnit()

    def test_attach_channel(self):
        ch = Mock(spec=BaseChannel)
        self._endpoint_unit.attach_channel(ch)

        self.assertTrue(self._endpoint_unit.channel is not None)
        self.assertEquals(self._endpoint_unit.channel, ch)

    def test_send(self):

        # need a channel to send on
        self.assertRaises(AttributeError, self._endpoint_unit.send, "fake")

        ch = Mock(spec=SendChannel)
        self._endpoint_unit.attach_channel(ch)

        self._endpoint_unit.send("hi", {'header':'value'})
        ch.send.assert_called_once_with('hi', {'header':'value'})

    def test_close(self):
        ch = Mock(spec=BaseChannel)
        self._endpoint_unit.attach_channel(ch)
        self._endpoint_unit.close()
        ch.close.assert_called_once_with()

    def test_spawn_listener(self):
        ch = Mock(spec=BidirClientChannel)
        self._endpoint_unit.attach_channel(ch)

        self._endpoint_unit.spawn_listener()

        self._endpoint_unit.close()
        self.assertTrue(self._endpoint_unit._recv_greenlet.ready())

    def test_build_header(self):
        head = self._endpoint_unit._build_header({'fake': 'content'})
        self.assertTrue(isinstance(head, dict))

    def test_build_payload(self):
        fakemsg = {'fake':'content'}
        msg = self._endpoint_unit._build_payload(fakemsg)
        self.assertEquals(msg, fakemsg)

    def test_build_msg(self):
        fakemsg = {'fake':'content'}
        msg = self._endpoint_unit._build_msg(fakemsg)
#        self.assertTrue(isinstance(msg, dict))
#        self.assertTrue(msg.has_key('header'))
#        self.assertTrue(msg.has_key('payload'))
#        self.assertTrue(isinstance(msg['header'], dict))
#        self.assertEquals(fakemsg, msg['payload'])

    def test__message_received(self):
        self._endpoint_unit._build_invocation = Mock()
        self._endpoint_unit._intercept_msg_in = Mock()
        self._endpoint_unit.message_received  = Mock()
        self._endpoint_unit.message_received.return_value = sentinel.msg_return


        retval = self._endpoint_unit._message_received(sentinel.msg, sentinel.headers)

        self.assertEquals(retval, sentinel.msg_return)

        self._endpoint_unit._build_invocation.assert_called_once_with(path=Invocation.PATH_IN,
                                                                      message=sentinel.msg,
                                                                      headers=sentinel.headers)
        self.assertTrue(self._endpoint_unit._intercept_msg_in.called)
        self.assertTrue(self._endpoint_unit.message_received.called)