'Receives and processes the messages from the source.
For each message we update the request object, and place the
response in that request\'s queue. If the request is complete, we
send a message to the worker.
Args:
client_id: The client which sent the messages.
messages: A list of GrrMessage RDFValues.'

def ReceiveMessages(self, client_id, messages):

    now = time.time()
    with queue_manager.QueueManager(token=self.token, store=self.data_store) as manager:
        for (session_id, msgs) in utils.GroupBy(messages, operator.attrgetter('session_id')).iteritems():
            unprocessed_msgs = self.HandleWellKnownFlows(msgs)
            if (not unprocessed_msgs):
                continue
            for msg in unprocessed_msgs:
                manager.QueueResponse(msg)
            for msg in unprocessed_msgs:
                if (msg.request_id == 0):
                    manager.QueueNotification(session_id=msg.session_id, priority=msg.priority)
                    break
                elif (msg.type == rdf_flows.GrrMessage.Type.STATUS):
                    if msg.HasTaskID():
                        manager.DeQueueClientRequest(client_id, msg.task_id)
                    manager.QueueNotification(session_id=msg.session_id, priority=msg.priority, last_status=msg.request_id)
                    stat = rdf_flows.GrrStatus(msg.payload)
                    if (stat.status == rdf_flows.GrrStatus.ReturnedStatus.CLIENT_KILLED):
                        crash_details = rdf_client.ClientCrash(client_id=client_id, session_id=session_id, backtrace=stat.backtrace, crash_message=stat.error_message, nanny_status=stat.nanny_status, timestamp=rdfvalue.RDFDatetime.Now())
                        msg = rdf_flows.GrrMessage(source=client_id, payload=crash_details, auth_state=rdf_flows.GrrMessage.AuthorizationState.AUTHENTICATED)
                        events.Events.PublishEvent('ClientCrash', msg, token=self.token)
    logging.debug('Received    %s    messages    from    %s    in    %s    sec', len(messages), client_id, (time.time() - now))
