ThreadedFeatureInputStream implements FeatureInputStream { @Override public void close() { iterator.close(); } ThreadedFeatureInputStream( final FeatureInputStream featureInputStream, final int queueSize ); @Override void close(); @Override FeatureCollection toCollection(); @Override Iterator<Feature> iterator(); @Override int count(); static void shutdown();  }
@Test @SuppressWarnings("unchecked") public void testConsumer() throws Exception { BlockingQueue<ProducerMessage> producerQueue = mock( BlockingQueue.class ); BlockingQueue<ConsumerMessage> consumerQueue = mock( BlockingQueue.class ); Feature[] features = new Feature[] { mock( Feature.class ), mock( Feature.class ) }; when( producerQueue.take() ).thenReturn( new ProducerFeatureMessage( features[0] ), new ProducerFeatureMessage( features[1] ), new ProducerFinishedMessage() ); Consumer consumer = new Consumer( producerQueue, consumerQueue ); assertTrue( consumer.hasNext() ); assertEquals( features[0], consumer.next() ); assertTrue( consumer.hasNext() ); assertEquals( features[1], consumer.next() ); assertFalse( consumer.hasNext() ); consumer.close(); verify( consumerQueue, never() ).put( new ConsumerClosingMessage() ); }
@Test @SuppressWarnings("unchecked") public void testConsumerAbort() throws Exception { BlockingQueue<ProducerMessage> producerQueue = mock( BlockingQueue.class ); BlockingQueue<ConsumerMessage> consumerQueue = mock( BlockingQueue.class ); Feature[] features = new Feature[] { mock( Feature.class ), mock( Feature.class ) }; when( producerQueue.take() ).thenReturn( new ProducerFeatureMessage( features[0] ), new ProducerFeatureMessage( features[1] ), new ProducerFinishedMessage() ); Consumer consumer = new Consumer( producerQueue, consumerQueue ); assertTrue( consumer.hasNext() ); assertEquals( features[0], consumer.next() ); assertTrue( consumer.hasNext() ); consumer.close(); verify( consumerQueue ).put( new ConsumerClosingMessage() ); verify( producerQueue, times( 3 ) ).take(); assertFalse( consumer.hasNext() ); }