@Test public void consumerShutsDown_whenNothingQueuedAndProducersShutdown() { assertThat(consumer.call(), is(0L)); }
@Override public Long call() { long objectCount = 0; try { while (true) { if (numProducersShutdown.get() < numProducers || !queue.isEmpty()) { OWLCompositeObject owlObject = queue.poll(1, TimeUnit.MINUTES); if (null == owlObject) { continue; } if (0 == queue.size() % 100_000) { logger.info("Currently " + queue.size() + " objects remaining in the queue"); } try { visitor.setOntology(owlObject.getOntology()); owlObject.getObject().accept(visitor); } catch (RuntimeException e) { logger.log(Level.WARNING, e.getMessage(), e); } objectCount++; } else { break; } } } catch (InterruptedException consumed) { logger.log(Level.WARNING, consumed.getMessage(), consumed); } logger.info("Ontology consumer shutting after processing " + objectCount + " objects..."); return objectCount; }
OwlOntologyConsumer implements Callable<Long> { @Override public Long call() { long objectCount = 0; try { while (true) { if (numProducersShutdown.get() < numProducers || !queue.isEmpty()) { OWLCompositeObject owlObject = queue.poll(1, TimeUnit.MINUTES); if (null == owlObject) { continue; } if (0 == queue.size() % 100_000) { logger.info("Currently " + queue.size() + " objects remaining in the queue"); } try { visitor.setOntology(owlObject.getOntology()); owlObject.getObject().accept(visitor); } catch (RuntimeException e) { logger.log(Level.WARNING, e.getMessage(), e); } objectCount++; } else { break; } } } catch (InterruptedException consumed) { logger.log(Level.WARNING, consumed.getMessage(), consumed); } logger.info("Ontology consumer shutting after processing " + objectCount + " objects..."); return objectCount; } }
OwlOntologyConsumer implements Callable<Long> { @Override public Long call() { long objectCount = 0; try { while (true) { if (numProducersShutdown.get() < numProducers || !queue.isEmpty()) { OWLCompositeObject owlObject = queue.poll(1, TimeUnit.MINUTES); if (null == owlObject) { continue; } if (0 == queue.size() % 100_000) { logger.info("Currently " + queue.size() + " objects remaining in the queue"); } try { visitor.setOntology(owlObject.getOntology()); owlObject.getObject().accept(visitor); } catch (RuntimeException e) { logger.log(Level.WARNING, e.getMessage(), e); } objectCount++; } else { break; } } } catch (InterruptedException consumed) { logger.log(Level.WARNING, consumed.getMessage(), consumed); } logger.info("Ontology consumer shutting after processing " + objectCount + " objects..."); return objectCount; } @Inject  OwlOntologyConsumer(BlockingQueue<OWLCompositeObject> queue, Graph graph, @IndicatesNumberOfProducerThreads int numProducers,
      @IndicatesMappedProperties List<MappedProperty> mappedProperties,
      @IndicatesNumberOfShutdownProducers AtomicInteger numProducersShutdown); }
OwlOntologyConsumer implements Callable<Long> { @Override public Long call() { long objectCount = 0; try { while (true) { if (numProducersShutdown.get() < numProducers || !queue.isEmpty()) { OWLCompositeObject owlObject = queue.poll(1, TimeUnit.MINUTES); if (null == owlObject) { continue; } if (0 == queue.size() % 100_000) { logger.info("Currently " + queue.size() + " objects remaining in the queue"); } try { visitor.setOntology(owlObject.getOntology()); owlObject.getObject().accept(visitor); } catch (RuntimeException e) { logger.log(Level.WARNING, e.getMessage(), e); } objectCount++; } else { break; } } } catch (InterruptedException consumed) { logger.log(Level.WARNING, consumed.getMessage(), consumed); } logger.info("Ontology consumer shutting after processing " + objectCount + " objects..."); return objectCount; } @Inject  OwlOntologyConsumer(BlockingQueue<OWLCompositeObject> queue, Graph graph, @IndicatesNumberOfProducerThreads int numProducers,
      @IndicatesMappedProperties List<MappedProperty> mappedProperties,
      @IndicatesNumberOfShutdownProducers AtomicInteger numProducersShutdown); @Override Long call(); }
OwlOntologyConsumer implements Callable<Long> { @Override public Long call() { long objectCount = 0; try { while (true) { if (numProducersShutdown.get() < numProducers || !queue.isEmpty()) { OWLCompositeObject owlObject = queue.poll(1, TimeUnit.MINUTES); if (null == owlObject) { continue; } if (0 == queue.size() % 100_000) { logger.info("Currently " + queue.size() + " objects remaining in the queue"); } try { visitor.setOntology(owlObject.getOntology()); owlObject.getObject().accept(visitor); } catch (RuntimeException e) { logger.log(Level.WARNING, e.getMessage(), e); } objectCount++; } else { break; } } } catch (InterruptedException consumed) { logger.log(Level.WARNING, consumed.getMessage(), consumed); } logger.info("Ontology consumer shutting after processing " + objectCount + " objects..."); return objectCount; } @Inject  OwlOntologyConsumer(BlockingQueue<OWLCompositeObject> queue, Graph graph, @IndicatesNumberOfProducerThreads int numProducers,
      @IndicatesMappedProperties List<MappedProperty> mappedProperties,
      @IndicatesNumberOfShutdownProducers AtomicInteger numProducersShutdown); @Override Long call();  }