ParallelGatewaySenderEventProcessor extends AbstractGatewaySenderEventProcessor { @Override public int eventQueueSize() { ParallelGatewaySenderQueue queue = (ParallelGatewaySenderQueue) getQueue(); return queue == null ? 0 : queue.localSize(); } protected  ParallelGatewaySenderEventProcessor(AbstractGatewaySender sender,
      ThreadsMonitoring tMonitoring, boolean cleanQueues); protected  ParallelGatewaySenderEventProcessor(AbstractGatewaySender sender,
      Set<Region> userRegions, int id, int nDispatcher, ThreadsMonitoring tMonitoring,
      boolean cleanQueues); @Override int eventQueueSize(); @Override void enqueueEvent(EnumListenerEvent operation, EntryEvent event, Object substituteValue,
      boolean isLastEventInTransaction); @Override void clear(PartitionedRegion pr, int bucketId); @Override void notifyEventProcessorIfRequired(int bucketId); @Override BlockingQueue<GatewaySenderEventImpl> getBucketTmpQueue(int bucketId); @Override PartitionedRegion getRegion(String prRegionName); @Override void removeShadowPR(String prRegionName); @Override void conflateEvent(Conflatable conflatableObject, int bucketId, Long tailKey); @Override void addShadowPartitionedRegionForUserPR(PartitionedRegion pr); @Override void addShadowPartitionedRegionForUserRR(DistributedRegion userRegion); @Override void initializeEventDispatcher();  }
@Test public void eventQueueSizeReturnsQueueLocalSize() { ParallelGatewaySenderEventProcessor processor = spy(new ParallelGatewaySenderEventProcessor(sender, mock( ThreadsMonitoring.class), false)); doReturn(queue).when(processor).getQueue(); processor.eventQueueSize(); verify(queue).localSize(); }
@Test public void eventQueueSizeReturnsZeroIfQueueIsNull() { ParallelGatewaySenderEventProcessor processor = spy(new ParallelGatewaySenderEventProcessor(sender, mock( ThreadsMonitoring.class), false)); doReturn(null).when(processor).getQueue(); assertThat(processor.eventQueueSize()).isEqualTo(0); verify(queue, never()).localSize(); }