'Does the real work with a single flow.'

def _ProcessMessages(self, notification, queue_manager):

    flow_obj = None
    session_id = notification.session_id
    try:
        flow_name = session_id.FlowName()
        if (flow_name in self.well_known_flows):
            expected_flow = self.well_known_flows[flow_name].__class__
            flow_obj = aff4.FACTORY.CreateWithLock(session_id, expected_flow, lease_time=self.well_known_flow_lease_time, blocking=False, token=self.token)
        else:
            flow_obj = aff4.FACTORY.OpenWithLock(session_id, lease_time=self.flow_lease_time, blocking=False, token=self.token)
        now = time.time()
        logging.debug('Got    lock    on    %s', session_id)
        queue_manager.DeleteNotification(session_id, end=notification.timestamp)
        if (flow_name in self.well_known_flows):
            stats.STATS.IncrementCounter('well_known_flow_requests', fields=[str(session_id)])
            with flow_obj:
                responses = flow_obj.FetchAndRemoveRequestsAndResponses(session_id)
            flow_obj.ProcessResponses(responses, self.__class__.thread_pool)
        else:
            with flow_obj:
                self._ProcessRegularFlowMessages(flow_obj, notification)
        elapsed = (time.time() - now)
        logging.debug('Done    processing    %s:    %s    sec', session_id, elapsed)
        stats.STATS.RecordEvent('worker_flow_processing_time', elapsed, fields=[flow_obj.Name()])
        self.queued_flows.ExpireObject(session_id)
    except aff4.LockError:
        stats.STATS.IncrementCounter('worker_flow_lock_error')
    except FlowProcessingError:
        pass
    except Exception as e:
        logging.exception('Error    processing    session    %s:    %s', session_id, e)
        stats.STATS.IncrementCounter('worker_session_errors', fields=[str(type(e))])
        queue_manager.DeleteNotification(session_id)
