QueueSynchronizationProcessor extends ReplyProcessor21 { @Override public void process(DistributionMessage msg) { try { reply = (QueueSynchronizationReplyMessage) msg; } finally { super.process(msg); } } QueueSynchronizationProcessor(DistributionManager dm, InternalDistributedMember primary); @Override void process(DistributionMessage msg);  }
@Test public void processMessageSetsReply() { QueueSynchronizationProcessor processor = spy(new QueueSynchronizationProcessor(manager, primary)); replyMessage = mock(QueueSynchronizationProcessor.QueueSynchronizationReplyMessage.class); processor.process(replyMessage); assertThat(processor.reply).isEqualTo(replyMessage); }
@Test public void processQueueSynchronizationMessageSendsReply() { message = spy(new QueueSynchronizationProcessor.QueueSynchronizationMessage()); setupMessage(); message.process(manager); verify(replyMessage).setEventIds(dispatched); verify(replyMessage).setSuccess(); verifyReplyMessageSent(); }