AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory { @Override public AsyncEventQueue create(String asyncQueueId, AsyncEventListener listener) { if (listener == null) { throw new IllegalArgumentException( "AsyncEventListener cannot be null"); } AsyncEventQueue asyncEventQueue; if (cache instanceof CacheCreation) { asyncEventQueue = new AsyncEventQueueCreation(asyncQueueId, gatewaySenderAttributes, listener); if (pauseEventsDispatching) { ((AsyncEventQueueCreation) asyncEventQueue).setPauseEventDispatching(true); } ((CacheCreation) cache).addAsyncEventQueue(asyncEventQueue); } else { if (logger.isDebugEnabled()) { logger.debug("Creating GatewaySender that underlies the AsyncEventQueue"); } addAsyncEventListener(listener); InternalGatewaySender sender = (InternalGatewaySender) create(getSenderIdFromAsyncEventQueueId(asyncQueueId)); AsyncEventQueueImpl asyncEventQueueImpl = new AsyncEventQueueImpl(sender, listener); asyncEventQueue = asyncEventQueueImpl; cache.addAsyncEventQueue(asyncEventQueueImpl); if (pauseEventsDispatching) { sender.setStartEventProcessorInPausedState(); } if (!gatewaySenderAttributes.isManualStart()) { sender.start(); } } if (logger.isDebugEnabled()) { logger.debug("Returning AsyncEventQueue" + asyncEventQueue); } return asyncEventQueue; } AsyncEventQueueFactoryImpl(InternalCache cache);   AsyncEventQueueFactoryImpl(InternalCache cache, GatewaySenderAttributes gatewaySenderAttributes,
      int batchTimeInterval); @Override AsyncEventQueueFactory setBatchSize(int size); @Override AsyncEventQueueFactory setPersistent(boolean isPersistent); @Override AsyncEventQueueFactory setDiskStoreName(String name); @Override AsyncEventQueueFactory setMaximumQueueMemory(int memory); @Override AsyncEventQueueFactory setDiskSynchronous(boolean isSynchronous); @Override AsyncEventQueueFactory setBatchTimeInterval(int batchTimeInterval); @Override AsyncEventQueueFactory setBatchConflationEnabled(boolean isConflation); @Override AsyncEventQueueFactory setDispatcherThreads(int numThreads); @Override AsyncEventQueueFactory setOrderPolicy(OrderPolicy policy); @Override AsyncEventQueueFactory addGatewayEventFilter(GatewayEventFilter filter); @Override AsyncEventQueueFactory removeGatewayEventFilter(GatewayEventFilter filter); @Override AsyncEventQueueFactory setGatewayEventSubstitutionListener(
      GatewayEventSubstitutionFilter filter); AsyncEventQueueFactory removeGatewayEventAlternateValueProvider(
      GatewayEventSubstitutionFilter provider); AsyncEventQueueFactory addAsyncEventListener(AsyncEventListener listener); @Override AsyncEventQueue create(String asyncQueueId, AsyncEventListener listener); void configureAsyncEventQueue(AsyncEventQueue asyncQueueCreation); @Override AsyncEventQueueFactory setParallel(boolean isParallel); AsyncEventQueueFactory setBucketSorted(boolean isbucketSorted); AsyncEventQueueFactory setIsMetaQueue(boolean isMetaQueue); @Override AsyncEventQueueFactory setForwardExpirationDestroy(boolean forward); @Override AsyncEventQueueFactory pauseEventDispatching(); static final int DEFAULT_BATCH_TIME_INTERVAL; }
@Test public void testCreateAsyncEventQueueWithNullListener() { asyncEventQueueFactory = new AsyncEventQueueFactoryImpl(cache); assertThatThrownBy(() -> asyncEventQueueFactory.create("id", null)) .isInstanceOf(IllegalArgumentException.class); }