def _Main(self):
    """The main loop."""
    self._StartProfiling(self._processing_configuration.profiling)

    if self._serializers_profiler:
      self._storage_writer.SetSerializersProfiler(self._serializers_profiler)

    if self._storage_profiler:
      self._storage_writer.SetStorageProfiler(self._storage_profiler)

    logger.debug('Analysis plugin: {0!s} (PID: {1:d}) started'.format(
        self._name, self._pid))

    # Creating the threading event in the constructor will cause a pickle
    # error on Windows when an analysis process is created.
    self._foreman_status_wait_event = threading.Event()
    self._status = definitions.STATUS_INDICATOR_ANALYZING

    task = tasks.Task()
    # TODO: temporary solution.
    task.identifier = self._analysis_plugin.plugin_name

    self._task = task

    storage_writer = self._storage_writer.CreateTaskStorage(task)

    if self._serializers_profiler:
      storage_writer.SetSerializersProfiler(self._serializers_profiler)

    if self._storage_profiler:
      storage_writer.SetStorageProfiler(self._storage_profiler)

    storage_writer.Open()

    self._analysis_mediator = analysis_mediator.AnalysisMediator(
        storage_writer, self._knowledge_base, data_location=self._data_location)

    # TODO: set event_filter_expression in mediator.

    storage_writer.WriteTaskStart()

    try:
      logger.debug(
          '{0!s} (PID: {1:d}) started monitoring event queue.'.format(
              self._name, self._pid))

      while not self._abort:
        try:
          event = self._event_queue.PopItem()

        except (errors.QueueClose, errors.QueueEmpty) as exception:
          logger.debug('ConsumeItems exiting with exception {0:s}.'.format(
              type(exception)))
          break

        if isinstance(event, plaso_queue.QueueAbort):
          logger.debug('ConsumeItems exiting, dequeued QueueAbort object.')
          break

        self._ProcessEvent(self._analysis_mediator, event)

        self._number_of_consumed_events += 1

        if self._guppy_memory_profiler:
          self._guppy_memory_profiler.Sample()

      logger.debug(
          '{0!s} (PID: {1:d}) stopped monitoring event queue.'.format(
              self._name, self._pid))

      if not self._abort:
        self._status = definitions.STATUS_INDICATOR_REPORTING

        self._analysis_mediator.ProduceAnalysisReport(self._analysis_plugin)

    # All exceptions need to be caught here to prevent the process
    # from being killed by an uncaught exception.
    except Exception as exception:  # pylint: disable=broad-except
      logger.warning(
          'Unhandled exception in process: {0!s} (PID: {1:d}).'.format(
              self._name, self._pid))
      logger.exception(exception)

      self._abort = True

    finally:
      storage_writer.WriteTaskCompletion(aborted=self._abort)

      storage_writer.Close()

      if self._serializers_profiler:
        storage_writer.SetSerializersProfiler(None)

      if self._storage_profiler:
        storage_writer.SetStorageProfiler(None)

    try:
      self._storage_writer.FinalizeTaskStorage(task)
    except IOError:
      pass

    if self._abort:
      self._status = definitions.STATUS_INDICATOR_ABORTED
    else:
      self._status = definitions.STATUS_INDICATOR_COMPLETED

    self._foreman_status_wait_event.wait(self._FOREMAN_STATUS_WAIT)

    logger.debug('Analysis plugin: {0!s} (PID: {1:d}) stopped'.format(
        self._name, self._pid))

    if self._serializers_profiler:
      self._storage_writer.SetSerializersProfiler(None)

    if self._storage_profiler:
      self._storage_writer.SetStorageProfiler(None)

    self._StopProfiling()

    self._analysis_mediator = None
    self._foreman_status_wait_event = None
    self._storage_writer = None
    self._task = None

    try:
      self._event_queue.Close(abort=self._abort)
    except errors.QueueAlreadyClosed:
      logger.error('Queue for {0:s} was already closed.'.format(self.name))
def _Main ( self ) : self . _StartProfiling ( self . _processing_configuration . profiling ) if self . _serializers_profiler : self . _storage_writer . SetSerializersProfiler ( self . _serializers_profiler ) if self . _storage_profiler : self . _storage_writer . SetStorageProfiler ( self . _storage_profiler ) logger . debug ( 'Analysis plugin: {0!s} (PID: {1:d}) started' . format ( self . _name , self . _pid ) ) self . _foreman_status_wait_event = threading . Event ( ) self . _status = definitions . STATUS_INDICATOR_ANALYZING task = tasks . Task ( ) task . identifier = self . _analysis_plugin . plugin_name self . _task = task storage_writer = self . _storage_writer . CreateTaskStorage ( task ) if self . _serializers_profiler : storage_writer . SetSerializersProfiler ( self . _serializers_profiler ) if self . _storage_profiler : storage_writer . SetStorageProfiler ( self . _storage_profiler ) storage_writer . Open ( ) self . _analysis_mediator = analysis_mediator . AnalysisMediator ( storage_writer , self . _knowledge_base , data_location = self . _data_location ) storage_writer . WriteTaskStart ( ) try : logger . debug ( '{0!s} (PID: {1:d}) started monitoring event queue.' . format ( self . _name , self . _pid ) ) while not self . _abort : try : event = self . _event_queue . PopItem ( ) except ( errors . QueueClose , errors . QueueEmpty ) as exception : logger . debug ( 'ConsumeItems exiting with exception {0:s}.' . format ( type ( exception ) ) ) break if isinstance ( event , plaso_queue . QueueAbort ) : logger . debug ( 'ConsumeItems exiting, dequeued QueueAbort object.' ) break self . _ProcessEvent ( self . _analysis_mediator , event ) self . _number_of_consumed_events += 1 if self . _guppy_memory_profiler : self . _guppy_memory_profiler . Sample ( ) logger . debug ( '{0!s} (PID: {1:d}) stopped monitoring event queue.' . format ( self . _name , self . _pid ) ) if not self . _abort : self . _status = definitions . STATUS_INDICATOR_REPORTING self . _analysis_mediator . ProduceAnalysisReport ( self . _analysis_plugin ) except Exception as exception : logger . warning ( 'Unhandled exception in process: {0!s} (PID: {1:d}).' . format ( self . _name , self . _pid ) ) logger . exception ( exception ) self . _abort = True finally : storage_writer . WriteTaskCompletion ( aborted = self . _abort ) storage_writer . Close ( ) if self . _serializers_profiler : storage_writer . SetSerializersProfiler ( None ) if self . _storage_profiler : storage_writer . SetStorageProfiler ( None ) try : self . _storage_writer . FinalizeTaskStorage ( task ) except IOError : pass if self . _abort : self . _status = definitions . STATUS_INDICATOR_ABORTED else : self . _status = definitions . STATUS_INDICATOR_COMPLETED self . _foreman_status_wait_event . wait ( self . _FOREMAN_STATUS_WAIT ) logger . debug ( 'Analysis plugin: {0!s} (PID: {1:d}) stopped' . format ( self . _name , self . _pid ) ) if self . _serializers_profiler : self . _storage_writer . SetSerializersProfiler ( None ) if self . _storage_profiler : self . _storage_writer . SetStorageProfiler ( None ) self . _StopProfiling ( ) self . _analysis_mediator = None self . _foreman_status_wait_event = None self . _storage_writer = None self . _task = None try : self . _event_queue . Close ( abort = self . _abort ) except errors . QueueAlreadyClosed : logger . error ( 'Queue for {0:s} was already closed.' . format ( self . name ) )
The main loop.