QueueSynchronizationProcessor extends ReplyProcessor21 { static List<EventID> getDispatchedEvents(final DistributionManager dm, final InternalDistributedMember primary, String regionName, List<EventID> events) { QueueSynchronizationProcessor processor = new QueueSynchronizationProcessor(dm, primary); QueueSynchronizationMessage message = new QueueSynchronizationMessage(); message.setEventIdList(events); message.setProcessorId(processor.getProcessorId()); message.setRegionName(regionName); message.setRecipient(primary); dm.putOutgoing(message); try { processor.waitForRepliesUninterruptibly(); } catch (ReplyException e) { e.handleCause(); } if (processor.reply != null) { return processor.reply.getDispatchedEvents(); } return null; } QueueSynchronizationProcessor(DistributionManager dm, InternalDistributedMember primary); @Override void process(DistributionMessage msg);  }
@Test public void getDispatchedEventsReturnsNullIfQueueIsNull() { message = spy(new QueueSynchronizationProcessor.QueueSynchronizationMessage()); String regionName = "queueName"; message.setRegionName(regionName); when(cache.getRegion(regionName)).thenReturn(null); assertThat(message.getDispatchedEvents(cache)).isNull(); }
@Test public void getDispatchedEventsReturns() { message = spy(new QueueSynchronizationProcessor.QueueSynchronizationMessage()); String regionName = "queueName"; message.setRegionName(regionName); message.setEventIdList(eventIDs); HARegion region = mock(HARegion.class); HARegionQueue queue = mock(HARegionQueue.class); when(cache.getRegion(regionName)).thenReturn(uncheckedCast(region)); when(region.getOwner()).thenReturn(queue); when(queue.getDispatchedEvents(eventIDs)).thenReturn(dispatched); assertThat(message.getDispatchedEvents(cache)).isEqualTo(dispatched); }