Call this method to process events. This method creates a background thread
than runs {@link #processEvents()} and performs the heavy lifting.
176224
public void runAsync ()
  {
    final Thread aThread = new Thread (this::processEvents,
                                       "WatchDir-" + m_aStartDir + "-" + RandomHelper.getRandom ().nextInt ());
    aThread.setDaemon (true);
    aThread.start ();
  }
987843
def dispatch_events!
      loop do
        event = @event_queue.pop

        callbacks.each do |callback|
          Thread.new do
            callback.call event
          end
        end
      end
    end
1097246
def run(self):
        
        self._stop = False
        while not self._stop:
            have_sources = self._timers or self._readers or self._writers
            if not self._processor.pending and not have_sources:
                break
            events = QEventLoop.AllEvents
            if not self._processor.pending:
                events |= QEventLoop.WaitForMoreEvents
            self._qapp.processEvents(events)
            if self._processor.pending:
                self._processor.run()
136637
public void startListening() throws FileNotFoundException, ClassNotFoundException, IOException,
        ResourceNotExistingException, TTException {
        mProcessingThread = new Thread() {
            public void run() {
                try {
                    processFileNotifications();
                } catch (InterruptedException | TTException | IOException e) {
                }
            }
        };
        mProcessingThread.start();

        initSessions();
    }
1366863
public void run() {
    lock.readLock().lock();
    try {
      if (Collections.isEmptyCollection(listeners)) {
        return;
      }
      for (ProcessorListener listener : listeners) {
        executorService.execute(listener);
      }
    } finally {
      lock.readLock().unlock();
    }
  }
422517
public void processEvent(ContentEvent event, int counter) {
		if (this.piInstances == null || this.piInstances.size() < this.getParallelism())
			throw new IllegalStateException("ThreadsWorkerProcessingItem(s) need to be setup before process any event (i.e. in ThreadsTopology.start()).");
		
		ThreadsProcessingItemInstance piInstance = this.piInstances.get(counter);
		ThreadsEventRunnable runnable = new ThreadsEventRunnable(piInstance, event);
		ThreadsEngine.getThreadWithIndex(piInstance.getThreadIndex()).submit(runnable);
	}
177777
private void processLoggingEvents(List<LoggingEvent> loggingEvents) {
        // Lazy initialization of the appender. This is needed because the
        // original appenders configuration may be available only after the
        // complete
        // log4j initialization.
        while (appenders.getAllAppenders() == null) {
            if ((batcher == null) || (batcher.isPaused())) {
                try {
                    Thread.sleep(SLEEP_TIME_MS);
                } catch (InterruptedException ignore) {

                }
                continue;
            }

            org.apache.log4j.Logger asyncLogger = LoggerCache.getInstance()
                    .getOrCreateLogger(LOGGER_ASYNC_APPENDER);
            Appender originalAppender = asyncLogger
                    .getAppender(originalAppenderName);
            if (originalAppender == null) {
                try {
                    Thread.sleep(SLEEP_TIME_MS);
                } catch (InterruptedException ignore) {

                }
                continue;
            }
            appenders.addAppender(originalAppender);
        }
        // First take the overflown summary events and put it back in the queue
        for (Iterator<Entry<String, LogSummary>> iter = logSummaryMap
                .entrySet().iterator(); iter.hasNext();) {
            Entry<String, LogSummary> mapEntry = (Entry<String, LogSummary>) iter
                    .next();
            // If the space is not available, then exit immediately
            if (batcher.isSpaceAvailable()) {
                LogSummary logSummary = mapEntry.getValue();
                LoggingEvent event = logSummary.createEvent();
                // Put the event in the queue and remove the event from the summary
                if (batcher.process(event)) {
                    iter.remove();
                } else {
                    break;
                }
            } else {
                break;
            }

        }
        // Process the events from the queue and call the underlying
        // appender
        for (LoggingEvent event : loggingEvents) {
            appenders.appendLoopOnAppenders(event);

        }

    }
718789
private void executeHandlers(List<Event> filteredEvents) {
        for (EventHandler current : eventHandlers) {
            for (Event event : filteredEvents) {
                current.handleEvent(event);
            }
        }
    }
877245
public void start() {
        if(actionInBackground != null) {

            Runnable jobToRun = new Runnable() {
                @Override
                public void run() {
                    result = (JobResult) actionInBackground.doAsync();
                    onResult();
                }
            };

            if(getExecutorService() != null) {
               asyncFutureTask = (FutureTask) getExecutorService().submit(jobToRun);
            } else {
                asyncThread = new Thread(jobToRun);
                asyncThread.start();
            }
        }
    }
198796
public void process(EventPipeline eventPipeline, EventBase<?> event) {
        try {
            for (HandlerReference hdlr : this) {
                try {
                    hdlr.invoke(event);
                    if (event.isStopped()) {
                        break;
                    }
                } catch (AssertionError t) {
                    // JUnit support
                    CoreUtils.setAssertionError(t);
                    event.handlingError(eventPipeline, t);
                } catch (Error e) { // NOPMD
                    // Wouldn't have caught it, if it was possible.
                    throw e;
                } catch (Throwable t) { // NOPMD
                    // Errors have been rethrown, so this should work.
                    event.handlingError(eventPipeline, t);
                }
            }
        } finally { // NOPMD
            try {
                event.handled();
            } catch (AssertionError t) {
                // JUnit support
                CoreUtils.setAssertionError(t);
                event.handlingError(eventPipeline, t);
            } catch (Error e) { // NOPMD
                // Wouldn't have caught it, if it was possible.
                throw e;
            } catch (Throwable t) { // NOPMD
                // Errors have been rethrown, so this should work.
                event.handlingError(eventPipeline, t);
            }
        }
    }
1272444
void start() {
    final ContextStart contextStart = new ContextStartImpl(this.identifier);
    for (final EventHandler<ContextStart> startHandler : this.contextStartHandlers) {
      startHandler.onNext(contextStart);
    }
  }
86319
public void handleEvent(Event event) {
        String topic = event.getTopic();
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "Received event topic: " + topic);
        }
        HttpInboundLink link = event.getProperty(CallbackIDs.CALLBACK_HTTPICL.getName(), HttpInboundLink.class);
        if (null != link) {
            if (tc.isEventEnabled()) {
                Tr.event(tc, "Processing request on new thread; " + link);
            }
            link.ready(link.getVirtualConnection());
        } else {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Unable to process pipelined request with null link");
            }
        }
    }
645004
def run(self):
        """"""
        self.api = self.context.get("cls")(self.context)
        self.context["inst"].append(self)  # Adapters used by strategies

        for call, calldata in self.context.get("calls", {}).items():
            def loop():
                """Loop on event scheduler, calling calls"""
                while not self.stopped.wait(calldata.get("delay", None)):
                    self.call(call, calldata.get("arguments", None))

            self.thread[call] = Process(target=loop)
            self.thread[call].start()
1191547
def process(event, watch_type = nil)
      @zk.raw_event_handler(event)

      logger.debug { "EventHandler#process dispatching event for #{watch_type.inspect}: #{event.inspect}" }# unless event.type == -1
      event.zk = @zk

      cb_keys = 
        if event.node_event?
          [event.path, ALL_NODE_EVENTS_KEY]
        elsif event.session_event?
          [state_key(event.state), ALL_STATE_EVENTS_KEY]
        else
          raise ZKError, "don't know how to process event: #{event.inspect}"
        end

      cb_ary = synchronize do 
        clear_watch_restrictions(event, watch_type)

        @callbacks.values_at(*cb_keys)
      end

      cb_ary.flatten! # takes care of not modifying original arrays
      cb_ary.compact!

      # we only filter for node events
      if event.node_event?
        interest_key = event.interest_key
        cb_ary.select! { |sub| sub.interests.include?(interest_key) }
      end

      safe_call(cb_ary, event)
    end
150688
@MainThread
    private void handleEventResult(Event event, EventResult result) {
        if (!activeEvents.contains(event)) {
            Utils.logE(event.getKey(), "Cannot send result of finished event");
            return;
        }

        scheduleResultCallbacks(event, result);
        executeTasks(false);
    }
405221
def process_events
      until @dead
        @mutex.synchronize do
          while @queue.empty?
            break if @dead
            @not_empty.wait(@mutex)
          end
          event = @queue.pop
          break unless event
          notify_listeners(event)
          event.dispatch
        end
      end
    rescue Exception => ex
      Logger.error "Error while running event: #{Logger.format_error(ex)}"
    end
198730
def __postCallAction(self, event):
        
        aThread = event.get_thread()
        retval  = self._get_return_value(aThread)
        self.__callHandler(self.__postCB, event, retval)