Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276972
--- Diff:
extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java
---
@@ -0,0 +1,884 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.rya.streams.querymanager;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.rya.streams.api.entity.StreamsQuery;
+import org.apache.rya.streams.api.queries.ChangeLogEntry;
+import org.apache.rya.streams.api.queries.InMemoryQueryRepository;
+import org.apache.rya.streams.api.queries.QueryChange;
+import org.apache.rya.streams.api.queries.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import
org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import
org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+import
com.google.common.util.concurrent.AbstractScheduledService.Scheduler;
+import com.google.common.util.concurrent.AbstractService;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams
system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log =
LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound
to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * How long blocking operations will be attempted before potentially
trying again.
+ */
+ private final long blockingValue;
+
+ /**
+ * The units for {@link #blockingValue}.
+ */
+ private final TimeUnit blockingUnits;
+
+ /**
+ * Used to inform threads that the application is shutting down, so
they must stop work.
+ */
+ private final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ /**
+ * This thread pool manages the two thread used to work the {@link
LogEvent}s
+ * and the {@link QueryEvent}s.
+ */
+ private final ExecutorService executor =
Executors.newFixedThreadPool(2);
+
+ /**
+ * Creates a new {@link QueryManager}.
+ *
+ * @param queryExecutor - Runs the active {@link StreamsQuery}s. (not
null)
+ * @param source - The {@link QueryChangeLogSource} of
QueryChangeLogs. (not null)
+ * @param blockingValue - How long blocking operations will try before
looping. (> 0)
+ * @param blockingUnits - The units of the {@code blockingValue}. (not
null)
+ */
+ public QueryManager(
+ final QueryExecutor queryExecutor,
+ final QueryChangeLogSource source,
+ final long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value
must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ protected void doStart() {
+ log.info("Starting a QueryManager.");
+
+ // A work queue of discovered Query Change Logs that need to be
handled.
+ // This queue exists so that the source notifying thread may be
released
+ // immediately instead of calling into blocking functions.
+ final BlockingQueue<LogEvent> logEvents = new
ArrayBlockingQueue<>(1024);
+
+ // A work queue of discovered Query Changes from the monitored
Query Change Logs
+ // that need to be handled. This queue exists so that the Query
Repository notifying
+ // thread may be released immediately instead of calling into
blocking functions.
+ final BlockingQueue<QueryEvent> queryEvents = new
ArrayBlockingQueue<>(1024);
+
+ try {
+ // Start up a LogEventWorker using the executor service.
+ executor.submit(new LogEventWorker(logEvents, queryEvents,
blockingValue, blockingUnits, shutdownSignal));
+
+ // Start up a QueryEvent Worker using the executor service.
+ executor.submit(new QueryEventWorker(queryEvents,
queryExecutor, blockingValue, blockingUnits, shutdownSignal));
+
+ // Start up the query execution framework.
+ queryExecutor.startAndWait();
+
+ // Startup the source that discovers new Query Change Logs.
+ changeLogSource.startAndWait();
+
+ // Subscribe the source a listener that writes to the
LogEventWorker's work queue.
+ changeLogSource.subscribe(new LogEventWorkGenerator(logEvents,
blockingValue, blockingUnits, shutdownSignal));
+ } catch(final RejectedExecutionException |
UncheckedExecutionException e) {
+ log.error("Could not start up a QueryManager.", e);
+ notifyFailed(e);
+ }
+
+ // Notify the service was successfully started.
+ notifyStarted();
+
+ log.info("QueryManager has finished starting.");
+ }
+
+ @Override
+ protected void doStop() {
+ log.info("Stopping a QueryManager.");
+
+ // Set the shutdown flag so that all components that rely on that
signal will stop processing.
+ shutdownSignal.set(true);
+
+ // Stop the workers and wait for them to die.
+ executor.shutdownNow();
+ try {
+ if(!executor.awaitTermination(10, TimeUnit.SECONDS)) {
+ log.warn("Waited 10 seconds for the worker threads to die,
but they are still running.");
+ }
+ } catch (final InterruptedException e) {
+ log.warn("Waited 10 seconds for the worker threads to die, but
they are still running.");
+ }
+
+ // Stop the source of new Change Logs.
+ try {
+ changeLogSource.stopAndWait();
+ } catch(final UncheckedExecutionException e) {
+ log.warn("Could not stop the Change Log Source.", e);
+ }
+
+ // Stop the query execution framework.
+ try {
+ queryExecutor.stopAndWait();
+ } catch(final UncheckedExecutionException e) {
+ log.warn("Could not stop the Query Executor", e);
+ }
+
+ // Notify the service was successfully stopped.
+ notifyStopped();
+
+ log.info("QueryManager has finished stopping.");
+ }
+
+ /**
+ * Offer a unit of work to a blocking queue until it is either
accepted, or the
+ * shutdown signal is set.
+ *
+ * @param workQueue - The blocking work queue to write to. (not null)
+ * @param event - The event that will be offered to the work queue.
(not null)
+ * @param offerValue - How long to wait when offering new work.
+ * @param offerUnits - The unit for the {@code offerValue}. (not null)
+ * @param shutdownSignal - Used to signal application shutdown has
started, so
+ * this method may terminate without ever placing the event on the
queue. (not null)
+ * @return {@code true} if the evet nwas added to the queue, otherwise
false.
+ */
+ private static <T> boolean offerUntilAcceptedOrShutdown(
+ final BlockingQueue<T> workQueue,
+ final T event, final
+ long offerValue,
+ final TimeUnit offerUnits,
+ final AtomicBoolean shutdownSignal) {
+ requireNonNull(workQueue);
+ requireNonNull(event);
+ requireNonNull(shutdownSignal);
+
+ boolean submitted = false;
+ while(!submitted && !shutdownSignal.get()) {
+ try {
+ submitted = workQueue.offer(event, offerValue, offerUnits);
+ if(!submitted) {
+ log.debug("An event could not be added to a work queue
after waiting 5 seconds. Trying again...");
+ }
+ } catch (final InterruptedException e) {
+ log.debug("An event could not be added to a work queue
after waiting 5 seconds. Trying again...");
+ }
+ }
+ return submitted;
+ }
+
+ /**
+ * An observation that a {@link QueryChangeLog} was created within or
+ * removed from a {@link QueryChangeLogSource}.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class LogEvent {
+
+ /**
+ * The types of events that may be observed.
+ */
+ static enum LogEventType {
+ /**
+ * A {@link QueryChangeLog} was created within a {@link
QueryChangeLogSource}.
+ */
+ CREATE,
+
+ /**
+ * A {@link QueryChangeLog} was deleted from a {@link
QueryChangeLogSource}.
+ */
+ DELETE;
+ }
+
+ private final String ryaInstance;
+ private final LogEventType eventType;
+ private final Optional<QueryChangeLog> log;
+
+ /**
+ * Constructs an instance of {@link LogEvent}.
+ *
+ * @param ryaInstance - The Rya Instance the log is/was for. (not
null)
+ * @param eventType - The type of event that was observed. (not
null)
+ * @param log - The log if this is a create event. (not null)
+ */
+ private LogEvent(final String ryaInstance, final LogEventType
eventType, final Optional<QueryChangeLog> log) {
+ this.ryaInstance = requireNonNull(ryaInstance);
+ this.eventType = requireNonNull(eventType);
+ this.log = requireNonNull(log);
+ }
+
+ /**
+ * @return The Rya Instance whose log was either created or
deleted.
+ */
+ public String getRyaInstanceName() {
+ return ryaInstance;
+ }
+
+ /**
+ * @return The type of event that was observed.
+ */
+ public LogEventType getEventType() {
+ return eventType;
+ }
+
+ /**
+ * @return The {@link QueryChangeLog} if this is a CREATE event.
+ */
+ public Optional<QueryChangeLog> getQueryChangeLog() {
+ return log;
+ }
+
+ @Override
+ public String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * Make a {@link LogEvent} that indicates a {@link QueryChangeLog}
was created within a
+ * {@link QueryChangeLogSource}.
+ *
+ * @param ryaInstance - The Rya Instance the created log is for.
(not null)
+ * @param log - The created {@link QueryChangeLog. (not null)
+ * @return A {@link LogEvent} built using the provided values.
+ */
+ public static LogEvent create(final String ryaInstance, final
QueryChangeLog log) {
+ return new LogEvent(ryaInstance, LogEventType.CREATE
,Optional.of(log));
+ }
+
+ /**
+ * Make a {@link LogEvent} that indicates a {@link QueryChangeLog}
was deleted from
+ * a {@link QueryChangeLogSource}.
+ *
+ * @param ryaInstance - The Rya Instance whose log was deleted.
(not null)
+ * @return A {@link LogEvent} built using the provided values.
+ */
+ public static LogEvent delete(final String ryaInstance) {
+ return new LogEvent(ryaInstance, LogEventType.DELETE,
Optional.empty());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or
not
+ * via the provided {@link QueryExecutor}.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class QueryEvent {
+
+ /**
+ * The type of events that may be observed.
+ */
+ public static enum QueryEventType {
+ /**
+ * Indicates a {@link StreamsQuery} needs to be executing.
+ */
+ EXECUTING,
+
+ /**
+ * Indicates a {@link StreamsQuery} needs to be stopped.
+ */
+ STOPPED,
+
+ /**
+ * Indicates all {@link StreamsQuery}s for a Rya instance need
to be stopped.
+ */
+ STOP_ALL;
+ }
+
+ private final String ryaInstance;
+ private final QueryEventType type;
+ private final Optional<UUID> queryId;
+ private final Optional<StreamsQuery> query;
+
+ /**
+ * Constructs an instance of {@link QueryEvent}.
+ *
+ * @param ryaInstance - The Rya instance that generated the event.
(not null)
+ * @param type - Indicates whether the query needs to be executing
or not. (not null)
+ * @param queryId - If stopped, the ID of the query that must not
be running. (not null)
+ * @param query - If executing, the StreamsQuery that defines what
should be executing. (not null)
+ */
+ private QueryEvent(
+ final String ryaInstance,
+ final QueryEventType type,
+ final Optional<UUID> queryId,
+ final Optional<StreamsQuery> query) {
+ this.ryaInstance = requireNonNull(ryaInstance);
+ this.type = requireNonNull(type);
+ this.queryId = requireNonNull(queryId);
+ this.query = requireNonNull(query);
+ }
+
+ /**
+ * @return The Rya instance that generated the event.
+ */
+ public String getRyaInstance() {
+ return ryaInstance;
+ }
+
+ /**
+ * @return Indicates whether the query needs to be executing or
not.
+ */
+ public QueryEventType getType() {
+ return type;
+ }
+
+ /**
+ * @return If stopped, the ID of the query that must not be
running. Otherwise absent.
+ */
+ public Optional<UUID> getQueryId() {
+ return queryId;
+ }
+
+ /**
+ * @return If executing, the StreamsQuery that defines what should
be executing. Otherwise absent.
+ */
+ public Optional<StreamsQuery> getStreamsQuery() {
+ return query;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(ryaInstance, type, queryId, query);
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if(o instanceof QueryEvent) {
+ final QueryEvent other = (QueryEvent) o;
+ return Objects.equals(ryaInstance, other.ryaInstance) &&
+ Objects.equals(type, other.type) &&
+ Objects.equals(queryId, other.queryId) &&
+ Objects.equals(query, other.query);
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder string = new StringBuilder();
+ string.append("Query Event {\n")
+ .append(" Rya
Instance:").append(ryaInstance).append(",\n")
+ .append(" Type: ").append(type).append(",\n");
+ switch(type) {
+ case EXECUTING:
+ append(string, query.get());
+ break;
+ case STOPPED:
+ string.append(" Query ID:
").append(queryId.get()).append("\n");
+ break;
+ case STOP_ALL:
+ break;
+ default:
+ // Default to showing everything that is in the object.
+ string.append(" Query ID:
").append(queryId.get()).append("\n");
+ append(string, query.get());
+ break;
+ }
+ string.append("}");
+ return string.toString();
+ }
+
+ private void append(final StringBuilder string, final StreamsQuery
query) {
+ requireNonNull(string);
+ requireNonNull(query);
+ string.append(" Streams Query {\n")
+ .append(" Query ID:
").append(query.getQueryId()).append(",\n")
+ .append(" Is Active:
").append(query.isActive()).append(",\n")
+ .append(" SPARQL:
").append(query.getSparql()).append("\n")
+ .append(" }");
+ }
+
+ /**
+ * Create a {@link QueryEvent} that indicates a query needs to be
executing.
+ *
+ * @param ryaInstance - The Rya instance that generated the event.
(not null)
+ * @param query - The StreamsQuery that defines what should be
executing. (not null)
+ * @return A {@link QueryEvent} built using the provided values.
+ */
+ public static QueryEvent executing(final String ryaInstance, final
StreamsQuery query) {
+ return new QueryEvent(ryaInstance, QueryEventType.EXECUTING,
Optional.empty(), Optional.of(query));
+ }
+
+ /**
+ * Create a {@link QueryEvent} that indicates a query needs to be
stopped.
+ *
+ * @param ryaInstance - The Rya instance that generated the event.
(not null)
+ * @param queryId - The ID of the query that must not be running.
(not null)
+ * @return A {@link QueryEvent} built using the provided values.
+ */
+ public static QueryEvent stopped(final String ryaInstance, final
UUID queryId) {
+ return new QueryEvent(ryaInstance, QueryEventType.STOPPED,
Optional.of(queryId), Optional.empty());
+ }
+
+ /**
+ * Create a {@link QueryEvent} that indicates all queries for a
Rya instance needs to be stopped.
+ *
+ * @param ryaInstance - The Rya instance that generated the event.
(not null)
+ * @return A {@link QueryEvent} built using the provided values.
+ */
+ public static QueryEvent stopALL(final String ryaInstance) {
+ return new QueryEvent(ryaInstance, QueryEventType.STOP_ALL,
Optional.empty(), Optional.empty());
+ }
+ }
+
+ /**
+ * Listens to a {@link QueryChangeLogSource} and adds observations to
the provided
+ * work queue. It does so until the provided shutdown signal is set.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class LogEventWorkGenerator implements SourceListener {
+
+ private final BlockingQueue<LogEvent> workQueue;
+ private final AtomicBoolean shutdownSignal;
+ private final long offerValue;
+ private final TimeUnit offerUnits;
+
+ /**
+ * Constructs an instance of {@link QueryManagerSourceListener}.
+ *
+ * @param workQueue - A blocking queue that will have {@link
LogEvent}s offered to it. (not null)
+ * @param offerValue - How long to wait when offering new work.
+ * @param offerUnits - The unit for the {@code offerValue}. (not
null)
+ * @param shutdownSignal - Indicates to this listener that it
needs to stop adding events
+ * to the work queue because the application is shutting down.
(not null)
+ */
+ public LogEventWorkGenerator(
+ final BlockingQueue<LogEvent> workQueue,
+ final long offerValue,
+ final TimeUnit offerUnits,
+ final AtomicBoolean shutdownSignal) {
+ this.workQueue = requireNonNull(workQueue);
+ this.shutdownSignal = requireNonNull(shutdownSignal);
+ this.offerValue = offerValue;
+ this.offerUnits = requireNonNull(offerUnits);
+ }
+
+ @Override
+ public void notifyCreate(final String ryaInstanceName, final
QueryChangeLog changeLog) {
+ log.info("A new Query Change Log has been discovered for Rya
Instance " + ryaInstanceName + ". All " +
+ "queries that are set to active within it will be
started.");
+
+ // Create an event that summarizes this notification.
+ final LogEvent event = LogEvent.create(ryaInstanceName,
changeLog);
+
+ // Offer it to the worker until there is room for it in the
work queue, or we are shutting down.
+ offerUntilAcceptedOrShutdown(workQueue, event, offerValue,
offerUnits, shutdownSignal);
+ }
+
+ @Override
+ public void notifyDelete(final String ryaInstanceName) {
+ log.info("The Query Change Log for Rya Instance " +
ryaInstanceName + " has been deleted. All of the " +
+ "queries related to that instance will be stopped.");
+
+ // Create an event that summarizes this notification.
+ final LogEvent event = LogEvent.delete(ryaInstanceName);
+
+ // Offer it to the worker until there is room for it in the
work queue, or we are shutting down.
+ offerUntilAcceptedOrShutdown(workQueue, event, offerValue,
offerUnits, shutdownSignal);
+ }
+ }
+
+ /**
+ * Processes a work queue of {@link LogEvent}s.
+ * <p/>
+ * Whenever a new log has been created, then it registers a {@link
QueryEventWorkGenerator}
+ * that generates {@link QueryEvent}s based on the content and updates
to the discovered
+ * {@link QueryChagneLog}.
+ * <p/>
+ * Whenever a log is deleted, then the generator is stopped and a stop
all {@link QueryEvent}
+ * is written to the work queue.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class LogEventWorker implements Runnable {
+
+ /**
+ * A map of Rya Instance name to he Query Repository for that
instance.
+ */
+ private final Map<String, QueryRepository> repos = new HashMap<>();
+
+ private final BlockingQueue<LogEvent> logWorkQueue;
+ private final BlockingQueue<QueryEvent> queryWorkQueue;
+ private final long blockingValue;
+ private final TimeUnit blockingUnits;
+ private final AtomicBoolean shutdownSignal;
+
+ /**
+ * Constructs an instance of {@link LogEventWorker}.
+ *
+ * @param logWorkQueue - A queue of {@link LogEvent}s that will be
worked by this object. (not null)
+ * @param queryWorkQueue - A queue where {@link QueryEvent}s will
be placed by this object. (not null)
+ * @param blockingValue - How long to wait when polling/offering
new work.
+ * @param blockingUnits - The unit for the {@code blockingValue}.
(not null)
+ * @param shutdownSignal - Indicates when the application has been
shutdown, so the executing thread
+ * may exit the {@link #run()} method. (not null)
+ */
+ public LogEventWorker(
+ final BlockingQueue<LogEvent> logWorkQueue,
+ final BlockingQueue<QueryEvent> queryWorkQueue,
+ final long blockingValue,
+ final TimeUnit blockingUnits,
+ final AtomicBoolean shutdownSignal) {
+ this.logWorkQueue = requireNonNull(logWorkQueue);
+ this.queryWorkQueue = requireNonNull(queryWorkQueue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ this.shutdownSignal = requireNonNull(shutdownSignal);
+ }
+
+ @Override
+ public void run() {
+ // Run until the shutdown signal is set.
+ while(!shutdownSignal.get()) {
+ try {
+ // Pull a unit of work from the queue.
+ log.debug("LogEventWorker - Polling the work queue for
a new LogEvent.");
+ final LogEvent logEvent =
logWorkQueue.poll(blockingValue, blockingUnits);
+ if(logEvent == null) {
+ // Poll again if nothing was found.
+ continue;
+ }
+
+ log.info("LogEventWorker - handling: \n" + logEvent);
+ final String ryaInstance =
logEvent.getRyaInstanceName();
+
+ switch(logEvent.getEventType()) {
+ case CREATE:
+ // If we see a create message for a Rya
Instance we are already maintaining,
+ // then don't do anything.
+ if(repos.containsKey(ryaInstance)) {
+ log.warn("LogEventWorker - A repository is
already being managed for the Rya Instance " +
+ ryaInstance + ". This message will
be ignored.");
+ continue;
+ }
+
+ // Create and start a QueryRepository for the
discovered log. Hold onto the repository
+ // so that it may be shutdown later.
+ final Scheduler scheduler =
Scheduler.newFixedRateSchedule(0, blockingValue, blockingUnits);
+ final QueryRepository repo = new
InMemoryQueryRepository(logEvent.getQueryChangeLog().get(), scheduler);
+ repo.startAndWait();
+ repos.put(ryaInstance, repo);
+
+ // Subscribe a worker that adds the Query
Events to the queryWorkQueue queue.
+ // A count down latch is used to ensure the
returned set of queries are handled
+ // prior to any notifications from the
repository.
+ final CountDownLatch subscriptionWorkFinished
= new CountDownLatch(1);
+ final QueryEventWorkGenerator
queryWorkGenerator =
+ new
QueryEventWorkGenerator(ryaInstance, subscriptionWorkFinished, queryWorkQueue,
+ blockingValue, blockingUnits,
shutdownSignal);
+
+ log.debug("LogEventWorker - Setting up a
QueryWorkGenerator...");
+ final Set<StreamsQuery> queries =
repo.subscribe(queryWorkGenerator);
+ log.debug("LogEventWorker - Finished setting
up a QueryWorkGenerator.");
+
+ // Handle the view of the queries within the
repository as it existed when
+ // the subscription was registered.
+ queries.stream()
+ .forEach(query -> {
+ // Create a QueryEvent that represents the
active state of the existing query.
+ final QueryEvent queryEvent =
query.isActive() ?
+ QueryEvent.executing(ryaInstance,
query) : QueryEvent.stopped(ryaInstance, query.getQueryId());
+ log.debug("LogEventWorker - offering: " +
queryEvent);
+
+ // Offer it to the worker until there is
room for it in the work queue, or we are shutting down.
+
offerUntilAcceptedOrShutdown(queryWorkQueue, queryEvent, blockingValue,
blockingUnits, shutdownSignal);
+ });
+
+ // Indicate the subscription work is finished
so that the registered listener may start
+ // adding work to the queue.
+ log.info("LogEventWorker - Counting down the
subscription work latch.");
+ subscriptionWorkFinished.countDown();
+ break;
+
+ case DELETE:
+ if(repos.containsKey(ryaInstance)) {
+ // Shut down the query repository for the
Rya instance. This ensures the listener will
+ // not receive any more work that needs to
be done.
+ final QueryRepository deletedRepo =
repos.remove(ryaInstance);
+ deletedRepo.stopAndWait();
+
+ // Add work that stops all of the queries
related to the instance.
+ final QueryEvent stopAllEvent =
QueryEvent.stopALL(ryaInstance);
+
offerUntilAcceptedOrShutdown(queryWorkQueue, stopAllEvent, blockingValue,
blockingUnits, shutdownSignal);
+ }
+ break;
+ }
+ } catch (final InterruptedException e) {
+ log.debug("LogEventWorker did not see any new events
over the past 5 seconds. Polling again...");
+ }
+ }
+
+ log.info("LogEventWorker shutting down...");
+
+ // Shutdown all of the QueryRepositories that were started.
+ repos.values().forEach(repo -> repo.stopAndWait());
+
+ log.info("LogEventWorker shut down.");
+ }
+ }
+
+ /**
+ * Listens to a {@link QueryRepository} and adds observations to the
provided work queue.
+ * It does so until the provided shutdown signal is set.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class QueryEventWorkGenerator implements QueryChangeLogListener
{
+
+ private final String ryaInstance;
+ private final CountDownLatch subscriptionWorkFinished;
+ private final BlockingQueue<QueryEvent> queryWorkQueue;
+ private final long blockingValue;
+ private final TimeUnit blockingUnits;
+ private final AtomicBoolean shutdownSignal;
+
+ /**
+ * Constructs an instance of {@link QueryEventWorkGenerator}.
+ *
+ * @param ryaInstance - The rya instance whose log this objects is
watching. (not null)
+ * @param subscriptionWorkFinished - Indicates when work that
needs to be completed before this
+ * listener handles notifications is completed. (not null)
+ * @param queryWorkQueue - A queue where {@link QueryEvent}s will
be placed by this object. (not null)
+ * @param blockingValue - How long to wait when polling/offering
new work.
+ * @param blockingUnits - The unit for the {@code blockingValue}.
(not null)
+ * @param shutdownSignal - Indicates to this listener that it
needs to stop adding events
+ * to the work queue because the application is shutting down.
(not null)
+ */
+ public QueryEventWorkGenerator(
+ final String ryaInstance,
+ final CountDownLatch subscriptionWorkFinished,
+ final BlockingQueue<QueryEvent> queryWorkQueue,
+ final long blockingValue,
+ final TimeUnit blockingUnits,
+ final AtomicBoolean shutdownSignal) {
+ this.ryaInstance = requireNonNull(ryaInstance);
+ this.subscriptionWorkFinished =
requireNonNull(subscriptionWorkFinished);
+ this.queryWorkQueue = requireNonNull(queryWorkQueue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ this.shutdownSignal = requireNonNull(shutdownSignal);
+ }
+
+ @Override
+ public void notify(final ChangeLogEntry<QueryChange>
queryChangeEvent, final Optional<StreamsQuery> newQueryState) {
+ requireNonNull(queryChangeEvent);
+ requireNonNull(newQueryState);
+
+ // Wait for the subscription work to be finished.
+ try {
+ log.debug("Waiting for Subscription Work Finished latch to
release...");
+ while(!shutdownSignal.get() &&
!subscriptionWorkFinished.await(blockingValue, blockingUnits)) {
+ log.debug("Still waiting...");
+ }
+ log.debug("Subscription Work Finished latch to released.");
+ } catch (final InterruptedException e) {
+ log.warn("Interrupted while waiting for the Subscription
Work Finished latch to be " +
+ "released. Shutting down?", e);
+ }
+
+ // If we left the loop because of a shutdown, return
immediately.
+ if(shutdownSignal.get()) {
+ log.debug("Not processing notification. Shutting down.");
+ return;
+ }
+
+ // Generate work from the notification.
+ final QueryChange change = queryChangeEvent.getEntry();
+ switch(change.getChangeType()) {
+ case CREATE:
+ if(newQueryState.isPresent()) {
+ log.info("Rya Instance " + ryaInstance + " created
Rya Streams query " + newQueryState + ".");
+ final StreamsQuery newQuery = newQueryState.get();
+ if(newQuery.isActive()) {
+ final QueryEvent executeNewQuery =
QueryEvent.executing(ryaInstance, newQuery);
+ offerUntilAcceptedOrShutdown(queryWorkQueue,
executeNewQuery, blockingValue, blockingUnits, shutdownSignal);
+ }
+ } else {
+ log.error("Received a CREATE QueryChange for Rya
Instance: " + ryaInstance +
+ ", Query ID: " + change.getQueryId() + ",
but the QueryRepository did not supply a " +
+ "StreamsQuery representing the created
query. The query will not be processed.");
+ }
+ break;
+
+ case DELETE:
+ final UUID deletedQueryId = change.getQueryId();
+ log.info("Rya Instance " + ryaInstance + " deleted Rya
Streams query with ID " + deletedQueryId);
+ final QueryEvent stopDeletedQuery =
QueryEvent.stopped(ryaInstance, deletedQueryId);
+ offerUntilAcceptedOrShutdown(queryWorkQueue,
stopDeletedQuery, blockingValue, blockingUnits, shutdownSignal);
+ break;
+
+ case UPDATE:
+ if(newQueryState.isPresent()) {
+ final StreamsQuery updatedQuery =
newQueryState.get();
+ if(updatedQuery.isActive()) {
+ log.info("Rya Instance " + ryaInstance + "
updated Rya Streams query with ID " +
+ updatedQuery.getQueryId() + " to be
active.");
+ final QueryEvent executeUpdatedQuery =
QueryEvent.executing(ryaInstance, updatedQuery);
+ offerUntilAcceptedOrShutdown(queryWorkQueue,
executeUpdatedQuery, blockingValue, blockingUnits, shutdownSignal);
+ } else {
+ log.info("Rya Instance " + ryaInstance + "
updated Rya Streams query with ID " +
+ updatedQuery.getQueryId() + " to be
inactive.");
+ final QueryEvent stopUpdatedQuery =
QueryEvent.stopped(ryaInstance, updatedQuery.getQueryId());
+ offerUntilAcceptedOrShutdown(queryWorkQueue,
stopUpdatedQuery, blockingValue, blockingUnits, shutdownSignal);
+ }
+ } else {
+ log.error("Received an UPDATE QueryChange for Rya
Instance: " + ryaInstance +
+ ", Query ID: " + change.getQueryId() + ",
but the QueryRepository did not supply a " +
+ "StreamsQuery representing the created
query. The query will not be processed.");
+ }
+ break;
+ }
+ }
+ }
+
+ /**
+ * Processes a work queue of {@link QueryEvent}s.
+ * <p/>
+ * Each type of event maps the to corresponding method on {@link
QueryExecutor} that is called into.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class QueryEventWorker implements Runnable {
+
+ private final BlockingQueue<QueryEvent> workQueue;
+ private final QueryExecutor queryExecutor;
+ private final long pollingValue;
+ private final TimeUnit pollingUnits;
+ private final AtomicBoolean shutdownSignal;
+
+ /**
+ * Constructs an instance of {@link QueryEventWorker}.
+ *
+ * @param logWorkQueue - A queue of {@link QueryEvent}s that will
be worked by this object. (not null)
--- End diff --
Done.
---
