'Processes messages for a given flow.'

def _ProcessRegularFlowMessages(self, flow_obj, notification):

    session_id = notification.session_id
    if (not isinstance(flow_obj, flow.FlowBase)):
        logging.warn('%s    is    not    a    proper    flow    object    (got    %s)', session_id, type(flow_obj))
        stats.STATS.IncrementCounter('worker_bad_flow_objects', fields=[str(type(flow_obj))])
        raise FlowProcessingError('Not    a    GRRFlow.')
    runner = flow_obj.GetRunner()
    try:
        runner.ProcessCompletedRequests(notification, self.__class__.thread_pool)
    except Exception as e:
        runner.context.state = rdf_flows.FlowContext.State.ERROR
        runner.context.backtrace = traceback.format_exc()
        logging.error('Flow    %s:    %s', flow_obj, e)
        raise FlowProcessingError(e)
