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):
    """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.
    """
    if data_store.RelationalDBEnabled():
      return self.ReceiveMessagesRelationalFlows(client_id, messages)

    now = time.time()
    with queue_manager.QueueManager(token=self.token) as manager:
      for session_id, msgs in iteritems(
          collection.Group(messages, operator.attrgetter("session_id"))):

        # Remove and handle messages to WellKnownFlows
        leftover_msgs = self.HandleWellKnownFlows(msgs)

        unprocessed_msgs = []
        for msg in leftover_msgs:
          if (msg.auth_state == msg.AuthorizationState.AUTHENTICATED or
              msg.session_id == self.unauth_allowed_session_id):
            unprocessed_msgs.append(msg)

        if len(unprocessed_msgs) < len(leftover_msgs):
          logging.info("Dropped %d unauthenticated messages for %s",
                       len(leftover_msgs) - len(unprocessed_msgs), client_id)

        if not unprocessed_msgs:
          continue

        for msg in unprocessed_msgs:
          manager.QueueResponse(msg)

        for msg in unprocessed_msgs:
          # Messages for well known flows should notify even though they don't
          # have a status.
          if msg.request_id == 0:
            manager.QueueNotification(session_id=msg.session_id)
            # Those messages are all the same, one notification is enough.
            break
          elif msg.type == rdf_flows.GrrMessage.Type.STATUS:
            # If we receive a status message from the client it means the client
            # has finished processing this request. We therefore can de-queue it
            # from the client queue. msg.task_id will raise if the task id is
            # not set (message originated at the client, there was no request on
            # the server), so we have to check .HasTaskID() first.
            if msg.HasTaskID():
              manager.DeQueueClientRequest(msg)

            manager.QueueNotification(
                session_id=msg.session_id, last_status=msg.request_id)

            stat = rdf_flows.GrrStatus(msg.payload)
            if stat.status == rdf_flows.GrrStatus.ReturnedStatus.CLIENT_KILLED:
              # A client crashed while performing an action, fire an event.
              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())
              events.Events.PublishEvent(
                  "ClientCrash", crash_details, token=self.token)

    logging.debug("Received %s messages from %s in %s sec", len(messages),
                  client_id,
                  time.time() - now)