code: public final void collect ()
    {
      if (m_aPerformer == null)
        throw new IllegalStateException ("No performer set!");
  
      try
      {
        // The temporary list that contains all objects to be delivered
        final ICommonsList <DATATYPE> aObjectsToPerform = new CommonsArrayList <> ();
        boolean bQueueIsStopped = false;
  
        while (true)
        {
          // Block until the first object is in the queue
          Object aCurrentObject = m_aQueue.take ();
          if (EqualsHelper.identityEqual (aCurrentObject, STOP_QUEUE_OBJECT))
            break;
  
          // add current object
          aObjectsToPerform.add (GenericReflection.uncheckedCast (aCurrentObject));
  
          // take all messages that are in the queue and handle them at once.
          // Handle at last m_nMaxPerformSize objects
          while (aObjectsToPerform.size () < m_nMaxPerformCount && !m_aQueue.isEmpty ())
          {
            // Explicitly handle the "stop queue message" (using "=="!!!)
            aCurrentObject = m_aQueue.take ();
            if (EqualsHelper.identityEqual (aCurrentObject, STOP_QUEUE_OBJECT))
            {
              bQueueIsStopped = true;
              break;
            }
  
            // add current object
            aObjectsToPerform.add (GenericReflection.uncheckedCast (aCurrentObject));
          }
  
          _perform (aObjectsToPerform);
  
          // In case we received a stop message while getting the bulk messages
          // above -> break the loop manually
          // Note: do not include in while-loop above because the conditions may
          // not execute in the correct order since "take" is blocking!
          if (bQueueIsStopped)
            break;
        }
  
        // perform any remaining actions
        _perform (aObjectsToPerform);
      }
      catch (final Exception ex)
      {
        LOGGER.error ("Error taking elements from queue - queue has been interrupted!!!", ex);
      }
    }
docs: This method starts the collector by taking objects from the internal
  {@link BlockingQueue}. So this method blocks and must be invoked from a
  separate thread. This method runs until {@link #stopQueuingNewObjects()} is
  new called and the queue is empty.
  
  @throws IllegalStateException
  if no performer is set - see
  {@link #setPerformer(IConcurrentPerformer)}