ContentBasedJmsQueueRecordWriter implements RecordWriter<P> { @Override public void writeRecords(Batch<P> batch) throws Exception { DefaultPredicate<P> defaultPredicate = new DefaultPredicate<>(); QueueSender defaultQueue = queueMap.get(defaultPredicate); for (Record<P> record : batch) { boolean matched = false; Message payload = (Message) record.getPayload(); for (Map.Entry<Predicate<P>, QueueSender> entry : queueMap.entrySet()) { Predicate<P> predicate = entry.getKey(); if (!(predicate instanceof DefaultPredicate) && predicate.matches(record)) { queueMap.get(predicate).send(payload); matched = true; break; } } if (!matched && defaultQueue != null) { defaultQueue.send(payload); } } } ContentBasedJmsQueueRecordWriter(Map<Predicate<P>, QueueSender> queueMap); @Override void writeRecords(Batch<P> batch);  }
@Test public void orangeRecordShouldBeDispatchedToOrangeQueue() throws Exception { recordWriter.writeRecords(new Batch(orangeRecord)); verify(orangeQueue).send(orangeRecord.getPayload()); verifyNoInteractions(defaultQueue); }
@Test public void nonOrangeRecordShouldBeDispatchedToDefaultQueue() throws Exception { recordWriter.writeRecords(new Batch(appleRecord)); verify(defaultQueue).send(appleRecord.getPayload()); verifyNoInteractions(orangeQueue); }