LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void startQuery(final String ryaInstance, final StreamsQuery query) throws QueryExecutorException { requireNonNull(ryaInstance); requireNonNull(query); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { final Set<String> topics = Sets.newHashSet( KafkaTopics.statementsTopic(ryaInstance), KafkaTopics.queryResultsTopic(ryaInstance, query.getQueryId())); createKafkaTopic.createTopics(topics, 1, 1, Optional.empty()); final KafkaStreams streams = streamsFactory.make(ryaInstance, query); streams.start(); ryaInstanceById.put(query.getQueryId(), ryaInstance); idByRyaInstance.put(ryaInstance, query.getQueryId()); byQueryId.put(query.getQueryId(), streams); } catch (final KafkaStreamsFactoryException e) { throw new QueryExecutorException("Could not start query " + query.getQueryId(), e); } 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; }
@Test(expected = IllegalStateException.class) public void startQuery_serviceNotStarted() throws Exception { final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class)); executor.startQuery("rya", new StreamsQuery(UUID.randomUUID(), "query", true, false)); }
@Test public void startQuery() throws Exception { final String ryaInstance = "rya"; final StreamsQuery query = new StreamsQuery(UUID.randomUUID(), "SELECT * WHERE { ?a ?b ?c. }", true, false); final KafkaStreamsFactory jobFactory = mock(KafkaStreamsFactory.class); final KafkaStreams queryJob = mock(KafkaStreams.class); when(jobFactory.make(eq(ryaInstance), eq(query))).thenReturn(queryJob); final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory); executor.startAndWait(); try { executor.startQuery(ryaInstance, query); verify(queryJob).start(); } finally { executor.stopAndWait(); } }