private void checkRunningListenersAndStartIfPossible ( DispatchQueueSelector queueSelector ) { synchronized ( queuedListenerTasks ) { // if either // - running for object-independent tasks or // - running for object-dependent tasks, but queue is empty or not present Queue < Runnable > queue = queueSelector == null ? null : queuedListenerTasks . get ( queueSelector ) ; if ( queue == null || queue . isEmpty ( ) ) { // if no object-independent tasks to be processed everything is fine, return if ( queuedListenerTasks . get ( null ) . isEmpty ( ) ) { return ; } // if object-independent tasks to be processed // check whether there are still queued object-dependent tasks boolean moreObjectDependentTasks = queuedListenerTasks . entrySet ( ) . stream ( ) . filter ( entry -> ! entry . getValue ( ) . isEmpty ( ) ) . anyMatch ( entry -> entry . getKey ( ) != null ) ; // if there are object-dependent tasks scheduled or there are still tasks running => retry later if ( moreObjectDependentTasks || ! runningListeners . isEmpty ( ) ) { return ; } // force object-independent task being handled // after all queued object-dependent tasks are finished // and before new object-dependent tasks will get queued queueSelector = null ; queue = queuedListenerTasks . get ( null ) ; } DispatchQueueSelector finalQueueSelector = queueSelector ; Queue < Runnable > taskQueue = queue ; // if there is something to execute and there is task running already if ( ! queue . isEmpty ( ) && runningListeners . add ( finalQueueSelector ) ) { AtomicReference < Future < ? > > activeListener = new AtomicReference <> ( ) ; activeListener . set ( api . getThreadPool ( ) . getExecutorService ( ) . submit ( ( ) -> { if ( finalQueueSelector instanceof ServerImpl ) { Object serverReadyNotifier = new Object ( ) ; ( ( ServerImpl ) finalQueueSelector ) . addServerReadyConsumer ( s -> { synchronized ( serverReadyNotifier ) { serverReadyNotifier . notifyAll ( ) ; } } ) ; while ( ! ( ( ServerImpl ) finalQueueSelector ) . isReady ( ) ) { try { synchronized ( serverReadyNotifier ) { serverReadyNotifier . wait ( 5000 ) ; } } catch ( InterruptedException ignored ) { } } } // Add the future to the list of active listeners activeListeners . put ( activeListener , new Object [ ] { System . nanoTime ( ) , finalQueueSelector } ) ; try { taskQueue . poll ( ) . run ( ) ; } catch ( Throwable t ) { logger . error ( "Unhandled exception in {}!" , ( ) -> getThreadType ( finalQueueSelector ) , ( ) -> t ) ; } activeListeners . remove ( activeListener ) ; runningListeners . remove ( finalQueueSelector ) ; // Inform the dispatchEvent method that it maybe can queue new listeners now synchronized ( queuedListenerTasks ) { queuedListenerTasks . notifyAll ( ) ; } checkRunningListenersAndStartIfPossible ( finalQueueSelector ) ; } ) ) ; } } }
Checks if there are listeners running for the given object and if not it takes one from the queue for the object and executes it .