'Very simple emulator of the worker.
We wake each flow in turn and run it.
Returns:
total number of flows still alive.
Raises:
RuntimeError: if the flow terminates with an error.'

def Next(self):

    with queue_manager.QueueManager(token=self.token) as manager:
        run_sessions = []
        for queue in self.queues:
            notifications_available = manager.GetNotificationsForAllShards(queue)
            for notification in notifications_available[:1]:
                session_id = notification.session_id
                manager.DeleteNotification(session_id, end=notification.timestamp)
                run_sessions.append(session_id)
                flow_name = session_id.FlowName()
                if (flow_name in self.well_known_flows):
                    well_known_flow = self.well_known_flows[flow_name]
                    with well_known_flow:
                        responses = well_known_flow.FetchAndRemoveRequestsAndResponses(well_known_flow.well_known_session_id)
                    well_known_flow.ProcessResponses(responses, self.pool)
                    continue
                with aff4.FACTORY.OpenWithLock(session_id, token=self.token, blocking=False) as flow_obj:
                    runner = flow_obj.GetRunner()
                    cpu_used = runner.context.client_resources.cpu_usage
                    user_cpu = self.cpu_user.next()
                    system_cpu = self.cpu_system.next()
                    network_bytes = self.network_bytes.next()
                    cpu_used.user_cpu_time += user_cpu
                    cpu_used.system_cpu_time += system_cpu
                    runner.context.network_bytes_sent += network_bytes
                    runner.ProcessCompletedRequests(notification, self.pool)
                    if (self.check_flow_errors and isinstance(flow_obj, flow.GRRFlow) and (runner.context.state == rdf_flows.FlowContext.State.ERROR)):
                        logging.exception('Flow    terminated    in    state    %s    with    an    error:    %s', runner.context.current_state, runner.context.backtrace)
                        raise RuntimeError(runner.context.backtrace)
    return run_sessions
