@Test public void getRunningQueryIds_noneStarted() throws Exception { final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class)); executor.startAndWait(); try { final Set<UUID> runningQueries = executor.getRunningQueryIds(); assertTrue(runningQueries.isEmpty()); } finally { executor.stopAndWait(); } }
@Override public Set<UUID> getRunningQueryIds() throws QueryExecutorException { lock.lock(); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); try { return new HashSet<>( byQueryId.keySet() ); } finally { lock.unlock(); } }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public Set<UUID> getRunningQueryIds() throws QueryExecutorException { lock.lock(); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); try { return new HashSet<>( byQueryId.keySet() ); } finally { lock.unlock(); } } }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public Set<UUID> getRunningQueryIds() throws QueryExecutorException { lock.lock(); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); try { return new HashSet<>( byQueryId.keySet() ); } finally { lock.unlock(); } } LocalQueryExecutor(
            final CreateKafkaTopic createKafkaTopic,
            final KafkaStreamsFactory streamsFactory); }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public Set<UUID> getRunningQueryIds() throws QueryExecutorException { lock.lock(); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); try { return new HashSet<>( byQueryId.keySet() ); } finally { lock.unlock(); } } LocalQueryExecutor(
            final CreateKafkaTopic createKafkaTopic,
            final KafkaStreamsFactory streamsFactory); @Override void startQuery(final String ryaInstance, final StreamsQuery query); @Override void stopQuery(final UUID queryId); @Override void stopAll(final String ryaInstanceName); @Override Set<UUID> getRunningQueryIds(); }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public Set<UUID> getRunningQueryIds() throws QueryExecutorException { lock.lock(); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); try { return new HashSet<>( byQueryId.keySet() ); } finally { lock.unlock(); } } LocalQueryExecutor(
            final CreateKafkaTopic createKafkaTopic,
            final KafkaStreamsFactory streamsFactory); @Override void startQuery(final String ryaInstance, final StreamsQuery query); @Override void stopQuery(final UUID queryId); @Override void stopAll(final String ryaInstanceName); @Override Set<UUID> getRunningQueryIds(); static ReentrantLock lock; }