def _on_actuator_result(self, topic, headers, message, match):
     """lock result"""
     msg = jsonapi.loads(message[0])
     print 'Actuator Results:', match, msg                    
     if headers['requesterID'] == agent_id:
         if self.actuator_handler is not None:
             self.actuator_handler(match, jsonapi.loads(message[0]))

    def iter_messages(self):
        """ Yields tuples of (watcher, subtopic, stat)"""
        recv = self.pubsub_socket.recv_multipart
        with self:
            while True:
                try:
                    events = dict(self.poller.poll(self.timeout * 1000))
                except zmq.ZMQError as e:
                    if e.errno == errno.EINTR:
                        continue
                    raise

                if len(events) == 0:
                    continue

                try:
                    topic, stat = recv()
                except zmq.core.error.ZMQError as e:
                    if e.errno != errno.EINTR:
                        raise
                    else:
                        try:
                            sys.exc_clear()
                        except Exception:
                            pass
                        continue

                topic = s(topic).split('.')
                if len(topic) == 3:
                    __, watcher, subtopic = topic
                    yield watcher, subtopic, json.loads(stat)
                elif len(topic) == 2:
                    __, watcher = topic
                    yield watcher, None, json.loads(stat)

 def on_message(self, message):
     prefix, message = message.split(",", 1)
     kernel, channel = prefix.split("/", 1)
     if channel=="stdin":
         # TODO: Support the stdin channel
         # See http://ipython.org/ipython-doc/dev/development/messaging.html
         return
     try:
         if kernel == "complete":
             application = self.session.handler.application
             message = jsonapi.loads(message)
             if message["header"]["msg_type"] in ("complete_request", "object_info_request"):
                 application.completer.registerRequest(self, message)
         elif kernel not in self.channels:
             # handler may be None in certain circumstances (it seems to only be set
             # in GET requests, not POST requests, so even using it here may
             # only work with JSONP because of a race condition)
             application = self.session.handler.application
             kernel_info = application.km.kernel_info(kernel)
             self.kernel_info = {'remote_ip': kernel_info['remote_ip'],
                                 'referer': kernel_info['referer'],
                                 'timeout': kernel_info['timeout']}
             self.channels[kernel] = \
                 {"shell": ShellSockJSHandler(kernel, self.send, application),
                  "iopub": IOPubSockJSHandler(kernel, self.send, application)}
             self.channels[kernel]["shell"].open(kernel)
             self.channels[kernel]["iopub"].open(kernel)
         if kernel != "complete":
             self._log_stats(kernel, message)
             self.channels[kernel][channel].on_message(message)
     except KeyError:
         jsonmessage=jsonapi.loads(message)
         logger.info("%s message sent to deleted kernel: %s"%(jsonmessage["header"]["msg_type"], kernel))
         pass # Ignore messages to nonexistant or killed kernels

 def load_config_from_json(self):
     """load config from existing json connector files."""
     c = self.config
     # load from engine config
     with open(os.path.join(self.profile_dir.security_dir, 'ipcontroller-engine.json')) as f:
         cfg = json.loads(f.read())
     key = c.Session.key = asbytes(cfg['exec_key'])
     xport,addr = cfg['url'].split('://')
     c.HubFactory.engine_transport = xport
     ip,ports = addr.split(':')
     c.HubFactory.engine_ip = ip
     c.HubFactory.regport = int(ports)
     self.location = cfg['location']
     if not self.engine_ssh_server:
         self.engine_ssh_server = cfg['ssh']
     # load client config
     with open(os.path.join(self.profile_dir.security_dir, 'ipcontroller-client.json')) as f:
         cfg = json.loads(f.read())
     assert key == cfg['exec_key'], "exec_key mismatch between engine and client keys"
     xport,addr = cfg['url'].split('://')
     c.HubFactory.client_transport = xport
     ip,ports = addr.split(':')
     c.HubFactory.client_ip = ip
     if not self.ssh_server:
         self.ssh_server = cfg['ssh']
     assert int(ports) == c.HubFactory.regport, "regport mismatch"

def unpack_legacy_message(headers, message):
    """Unpack legacy pubsub messages for VIP agents.

    Loads JSON-formatted message parts and removes single-frame messages
    from their containing list. Does not alter headers.
    """
    if not isinstance(headers, Headers):
        headers = Headers(headers)
    try:
        content_type = headers["Content-Type"]
    except KeyError:
        return headers, message
    if isinstance(content_type, basestring):
        if content_type.lower() == "application/json":
            if isinstance(message, list) and len(message) == 1:
                return jsonapi.loads(message[0])
            if isinstance(message, basestring):
                return jsonapi.loads(message)
        if isinstance(message, list) and len(message) == 1:
            return message[0]
    if isinstance(content_type, list) and isinstance(message, list):
        parts = [
            (jsonapi.loads(msg) if str(ctype).lower() == "application/json" else msg)
            for ctype, msg in zip(content_type, message)
        ]
        parts.extend(message[len(parts) :])
        if len(parts) == len(content_type) == 1:
            return parts[0]
        return parts
    return message

def test_agent_last_update_increases(volttron_instance):
    agent = volttron_instance.build_agent()
    s = json.loads(agent.vip.health.get_status())
    dt = dateparse(s['last_updated'], fuzzy=True)
    agent.vip.health.set_status(STATUS_UNKNOWN, 'Unknown now!')
    gevent.sleep(1)
    s = json.loads(agent.vip.health.get_status())
    dt2 = dateparse(s['last_updated'], fuzzy=True)
    assert dt < dt2

    def capture_data(self, peer, sender, bus, topic, headers, message, device):
        
        timestamp_string = headers.get(headers_mod.DATE)
        timestamp, my_tz = process_timestamp(timestamp_string)
        
        try:
            # 2.0 agents compatability layer makes sender == pubsub.compat so 
            # we can do the proper thing when it is here
            if sender == 'pubsub.compat':
                message = jsonapi.loads(message[0])
                
            if isinstance(message, dict):
                values = message
            else:
                values = message[0]
                
        except ValueError as e:
            _log.error("message for {topic} bad message string: {message_string}".format(topic=topic,
                                                                                     message_string=message[0]))
            return
        except IndexError as e:
            _log.error("message for {topic} missing message string".format(topic=topic))
            return
        except Exception as e:
            _log.exception(e)
            return

        meta = {}
        try:
            # 2.0 agents compatability layer makes sender == pubsub.compat so 
            # we can do the proper thing when it is here
            if sender == 'pubsub.compat':
                if isinstance(message[1], str):
                    meta = jsonapi.loads(message[1])
            
            if not isinstance(message, dict):
                meta = message[1]
                
        except ValueError as e:
            _log.warning("meta data for {topic} bad message string: {message_string}".format(topic=topic,
                                                                                     message_string=message[0]))
        except IndexError as e:
            _log.warning("meta data for {topic} missing message string".format(topic=topic))

        if topic.startswith('analysis'):
            source = 'analysis'
        else:
            source = 'scrape'
        _log.debug("Queuing {topic} from {source} for publish".format(topic=topic,
                                                                      source=source))
        
        for key, value in values.iteritems():
            point_topic = device + '/' + key
            self._event_queue.put({'source': source,
                                   'topic': point_topic,
                                   'readings': [(timestamp,value)],
                                   'meta': meta.get(key,{})})

 def test_send(self):
     self.publisher.send('test', 'method', '{"temp": 20, "humid": 30}')
     result = self.subscriber.recv()
     self.assertEqual(result[0], b'test') # data_type
     
     msg = json.loads(result[1])
     self.assertEqual(msg[0], 'method')
     
     data = json.loads(msg[1])
     self.assertEqual(data['temp'], 20)
     self.assertEqual(data['humid'], 30)        

def test_agent_status_changes(volttron_instance):
    unknown_message = "This is unknown"
    bad_message = "Bad kitty"
    agent = volttron_instance.build_agent()
    agent.vip.health.set_status(STATUS_UNKNOWN, unknown_message)
    r = json.loads(agent.vip.health.get_status())
    assert unknown_message == r['context']
    assert STATUS_UNKNOWN == r['status']

    agent.vip.health.set_status(STATUS_BAD, bad_message)
    r = json.loads(agent.vip.health.get_status())
    assert bad_message == r['context']
    assert STATUS_BAD == r['status']

    def test_call(self):
        self.state_handler.update_sensor({'test_temperature': 22})
        self.state_handler.update_appliance({'viera': 1})        
        self.state_handler.__call__()

        result = self.subscriber.recv()
        
        msg = json.loads(result[1])
        self.assertEqual(msg[0], 'mining')
        
        data = json.loads(msg[1])
        self.assertEqual(data['sensors']['test_temperature'], 22)                
        self.assertEqual(data['appliances']['viera'], 1)

def pair_recv(msg):
    print "Received from Server:"
    try:
        try:
            j = jsonapi.loads(''.join(msg))
        except TypeError:
            j = jsonapi.loads(msg)

        if j['state'] == 'naming':
            #assign_names(j['mapping'])
            forward_from_mesh()
        elif j['state'] == 'commanding':
            send_commands(j['commanding'])
    finally:
        pass

        def _on_new_data(self, topic, headers, message, match):
            """watching for new data"""
            data = jsonapi.loads(message[0])
#             self.current_spacetemp = float(data["ZoneTemp"])
            self.current_spacetemp = 76
            droveride = bool(int(data["CoolCall2"]))
            occupied = bool(int(data["Occupied"]))
            
            if droveride and self.state not in ('IDLE', 'CLEANUP', 'STARTUP'):
                print 'User Override Initiated'
                self.cancel_event()
            
            if not occupied and self.state in ('DR_EVENT', 'RESTORE'):
                self.cancel_event()
                
            if self.state == 'IDLE' or self.state=='STARTUP':
                #self.default_coolingstpt = float(data["CoolingStPt"])
                #self.default_heatingstpt = float(data["HeatingStPt"])
                self.default_coolingstpt = 75.0
                self.default_heatingstpt = 65.0
                self.default_firststage_fanspeed = float(data["CoolSupplyFanSpeed1"])
                self.default_secondstage_fanspeed = float(data["CoolSupplyFanSpeed2"])
                self.default_damperstpt = float(data["ESMDamperMinPosition"])
                
            if self.state == 'STARTUP':
                self.state = 'IDLE'

    def routerRecv(self, message):
        """
        message = [ ... , request, image/blank]

        request = {'timestamp': timestamp, 
                   'task': 'detection'/'recognition'/'tracking',
                   'parameters': (...)}
        """
        request = loads(message[-2])

        if request["task"] == "detection":
            self.logger.debug("start detection")
            with open("image.jpg", "wb") as f:
                f.write(message[-1])
            sleep = random.randint(1, 2)  # detection
            time.sleep(sleep)
            message[-2] = dumps("detection")
            message[-1] = ""
            self.rtr.send_multipart(message)

        elif request["task"] == "tracking":
            self.logger.debug("prepare to tracking")
            message[-1] = "finish"
            tracker.Tracker(self.rtr, message)
        else:
            self.logger.debug("requested task is not supported")

 def recv_message(self, flags=0):
     '''Recieve a message as (topic, headers, message) tuple.'''
     topic = self.recv_string(flags)
     headers = self.recv_string(flags) if self.rcvmore else ''
     headers = jsonapi.loads(headers) if headers else {}
     message = self.recv_multipart(flags) if self.rcvmore else []
     return topic, Headers(headers), message

     def handle_new(self, headers, message, now):
         print 'handle new'
         requester = headers.get('requesterID')
         self.task_id = headers.get('taskID')
         priority = headers.get('priority')
         
         try:
             requests = jsonapi.loads(message[0])
             requests = requests[0]
         except (ValueError, IndexError) as ex:
             # Could be ValueError of JSONDecodeError depending
             # on if simplesjson was used.  JSONDecodeError
             # inherits from ValueError
             
             #We let the schedule manager tell us this is a bad request.
             _log.error('bad request: {request}, {error}'.format(request=requests, error=str(ex)))
             requests = []
         device, start, end = requests
 
         self.start_time = parser.parse(start, fuzzy=True)
         self.end_time = parser.parse(end, fuzzy=True)
         event = sched.Event(self.announce)
         self.schedule(self.start_time, event) 
           
         
         topic = topics.ACTUATOR_SCHEDULE_RESULT()
         headers = self.get_headers(requester, task_id=self.task_id)
         headers['type'] = SCHEDULE_ACTION_NEW
         self.publish_json(topic, headers, {'result':'SUCCESS', 
                                            'data': 'NONE', 
                                            'info': 'NONE'})

def recv_json(socket, timeout = None, ignore_exiting = False):
    """
    Receives JSON from a socket. Assumes socket is set to timeout properly.
    Raises universal.Exiting if program is exiting, or zmq.ZMQError if
    timed out.

    timeout is in milliseconds

    """

    poller = zmq.Poller()
    poller.register(socket, zmq.POLLIN)

    if timeout is not None:
        start_time = time.time() * 1000

    poll_wait_time = 1000 if timeout is None else min(timeout, 1000)
    while ignore_exiting or not universal.exiting:
        if poller.poll(poll_wait_time):
            msg = socket.recv_multipart()

            # Decode the json in the innermost frame
            msg[-1] = jsonapi.loads(msg[-1])

            # If only one frame was received simply return that frame
            if len(msg) == 1: msg = msg[0]

            return msg
        elif timeout is not None and start_time + timeout <= time.time() * 1000:
            raise Timeout()

    raise universal.Exiting()

 def on_message(self, message):
     prefix, message = message.split(",", 1)
     id = prefix.split("/", 1)[0]
     message = jsonapi.loads(message)
     logger.debug("SockJSHandler.on_message: %s", message)
     msg_type = message["header"]["msg_type"]
     app = self.session.handler.application
     if id == "complete":
         if msg_type in ("complete_request", "object_info_request"):
             app.completer.registerRequest(self, message)
         return
     try:
         kernel = app.kernel_dealer.kernel(id)
     except KeyError:
         # Ignore messages to nonexistent or killed kernels.
         logger.warning("%s sent to nonexistent kernel %s", msg_type, id)
         return
     if id not in self.channels:
         self.channels[id] = SockJSChannelsHandler(self.send)
         self.channels[id].connect(kernel)
     if msg_type == "execute_request":
         stats_logger.info(StatsMessage(
             kernel_id=id,
             remote_ip=kernel.remote_ip,
             referer=kernel.referer,
             code=message["content"]["code"],
             execute_type="request"))
     self.channels[id].send(message)

    def handle_recv(self, data):
        """called each time circusd sends an event"""
        # maintains a periodic callback to compute mem and cpu consumption for
        # each pid.
        logger.debug('Received an event from circusd: %s' % str(data))
        topic, msg = data
        try:
            topic = s(topic)
            watcher = topic.split('.')[1:-1][0]
            action = topic.split('.')[-1]
            msg = json.loads(msg)

            if action in ('reap', 'kill'):
                # a process was reaped
                pid = msg['process_pid']
                self.remove_pid(watcher, pid)
            elif action == 'spawn':
                # a process was added
                pid = msg['process_pid']
                self._append_pid(watcher, pid)
            elif action == 'stop':
                # the whole watcher was stopped.
                self.stop_watcher(watcher)
            else:
                logger.debug('Unknown action: %r' % action)
                logger.debug(msg)
        except Exception:
            logger.exception('Failed to handle %r' % msg)

    def update_override_patterns(self):
        if self._override_patterns is None:
            try:
                values = self.vip.config.get("override_patterns")
                values = jsonapi.loads(values)

                if isinstance(values, dict):
                    self._override_patterns = set()
                    for pattern, end_time in values.items():
                        # check the end_time
                        now = utils.get_aware_utc_now()
                        # If end time is indefinite, set override with indefinite duration
                        if end_time == "0.0":
                            self._set_override_on(pattern, 0.0, from_config_store=True)
                        else:
                            end_time = utils.parse_timestamp_string(end_time)
                            # If end time > current time, set override with new duration
                            if end_time > now:
                                delta = end_time - now
                                self._set_override_on(pattern, delta.total_seconds(), from_config_store=True)
                else:
                    self._override_patterns = set()
            except KeyError:
                self._override_patterns = set()
            except ValueError:
                _log.error("Override patterns is not set correctly in config store")
                self._override_patterns = set()

    def test_connect(self):
        """Tests that we can connect and send data to a zmq puller"""

        # start dummy ZMQ pull server
        gevent.spawn(self._start_zmq_puller)
        self.zmq_server_listning_event.wait(5)

        # our local zmq logger
        zmq_url = 'tcp://localhost:{0}'.format(self.zmq_tcp_port)
        client_public_key = "N[DC7+%FKdW3pJUPnaCwWxt-0/jo5Lrq&U28-GG}"
        client_secret_key = "Gwt%C0a8J/:9Jy$qpDNTy8wRzlnRD-HT8H>u7F{B"
        server_public_key = "^4b:-bZ8seRC+m2p(sg{7{skOuK*jInNeH^/Le}Q"
        zmqLogger = ZmqLogger(zmq_url, client_public_key, client_secret_key, server_public_key)
        zmqLogger.start()

        # inject some data into the logging relay singleton
        self.reportingRelay.queueLogData({'somekey': 'somedata'})

        # wait until the zmq server put something into the local testing queue
        received_data = self.testing_queue.get(5)
        received_data = received_data.split(' ', 1)
        topic, message = received_data[0], jsonapi.loads(received_data[1])

        self.assertEqual(topic, ZmqMessageTypes.HERALDING_AUTH_LOG.value)
        self.assertIn('somekey', message)
        self.assertEqual(message['somekey'], 'somedata')