SchedulerClient.java

/**
 * Copyright (C) Gustav Karlsson
 *
 * Licensed 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 com.github.kagkarlsson.scheduler;

import com.github.kagkarlsson.scheduler.exceptions.TaskInstanceNotFoundException;
import com.github.kagkarlsson.scheduler.exceptions.TaskInstanceCurrentlyExecutingException;
import com.github.kagkarlsson.scheduler.jdbc.DefaultJdbcCustomization;
import com.github.kagkarlsson.scheduler.jdbc.JdbcCustomization;
import com.github.kagkarlsson.scheduler.jdbc.JdbcTaskRepository;
import com.github.kagkarlsson.scheduler.serializer.Serializer;
import com.github.kagkarlsson.scheduler.stats.StatsRegistry;
import com.github.kagkarlsson.scheduler.task.Execution;
import com.github.kagkarlsson.scheduler.task.SchedulableInstance;
import com.github.kagkarlsson.scheduler.task.Task;
import com.github.kagkarlsson.scheduler.task.TaskInstance;
import com.github.kagkarlsson.scheduler.task.TaskInstanceId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.function.Consumer;

import static java.util.Optional.ofNullable;

public interface SchedulerClient {

    /**
     * Schedule a new execution.
     *
     * @param taskInstance  Task-instance, optionally with data
     * @param executionTime Instant it should run
     * @return void
     * @see java.time.Instant
     * @see com.github.kagkarlsson.scheduler.task.TaskInstance
     */
    <T> void schedule(TaskInstance<T> taskInstance, Instant executionTime);

    <T> void schedule(SchedulableInstance<T> schedulableInstance);

    /**
     * Update an existing execution to a new execution-time. If the execution does not exist or if it is currently
     * running, an exception is thrown.
     *
     * @param taskInstanceId
     * @param newExecutionTime the new execution-time
     * @see java.time.Instant
     * @see com.github.kagkarlsson.scheduler.task.TaskInstanceId
     */
    void reschedule(TaskInstanceId taskInstanceId, Instant newExecutionTime);

    /**
     * Update an existing execution with a new execution-time and new task-data. If the execution does not exist or if
     * it is currently running, an exception is thrown.
     *
     * @param taskInstanceId
     * @param newExecutionTime the new execution-time
     * @param newData          the new task-data
     * @see java.time.Instant
     * @see com.github.kagkarlsson.scheduler.task.TaskInstanceId
     */
    <T> void reschedule(TaskInstanceId taskInstanceId, Instant newExecutionTime, T newData);

    /**
     * Update an existing execution with a new execution-time and new task-data. If the execution does not exist or if
     * it is currently running, an exception is thrown.
     *
     * @param schedulableInstance the updated instance
     */
    <T> void reschedule(SchedulableInstance<T> schedulableInstance);

    /**
     * Removes/Cancels an execution.
     *
     * @param taskInstanceId
     * @return void
     * @see com.github.kagkarlsson.scheduler.task.TaskInstanceId
     */
    void cancel(TaskInstanceId taskInstanceId);

    /**
     * Gets all scheduled executions and supplies them to the provided Consumer. A Consumer is used to
     * avoid forcing the SchedulerClient to load all executions in memory. Currently running executions are not returned.
     *
     * @param consumer Consumer for the executions
     * @return void
     */
    void fetchScheduledExecutions(Consumer<ScheduledExecution<Object>> consumer);
    void fetchScheduledExecutions(ScheduledExecutionsFilter filter, Consumer<ScheduledExecution<Object>> consumer);

    /**
     * @see #fetchScheduledExecutions(Consumer)
     */
    default List<ScheduledExecution<Object>> getScheduledExecutions() {
        List<ScheduledExecution<Object>> executions = new ArrayList<>();
        fetchScheduledExecutions(executions::add);
        return executions;
    }

    /**
     * @see #fetchScheduledExecutions(Consumer)
     */
    default List<ScheduledExecution<Object>> getScheduledExecutions(ScheduledExecutionsFilter filter) {
        List<ScheduledExecution<Object>> executions = new ArrayList<>();
        fetchScheduledExecutions(filter, executions::add);
        return executions;
    }

    /**
     * Gets all scheduled executions for a task and supplies them to the provided Consumer. A Consumer is used to
     * avoid forcing the SchedulerClient to load all executions in memory. Currently running executions are not returned.
     *
     * @param taskName  the name of the task to get scheduled-executions for
     * @param dataClass the task data-class the data will be serialized and cast to
     * @param consumer  Consumer for the executions
     * @return void
     */
    <T> void fetchScheduledExecutionsForTask(String taskName, Class<T> dataClass, Consumer<ScheduledExecution<T>> consumer);
    <T> void fetchScheduledExecutionsForTask(String taskName, Class<T> dataClass, ScheduledExecutionsFilter filter, Consumer<ScheduledExecution<T>> consumer);

    /**
     * @see #fetchScheduledExecutionsForTask(String, Class, Consumer)
     */
    default <T> List<ScheduledExecution<T>> getScheduledExecutionsForTask(String taskName, Class<T> dataClass) {
        List<ScheduledExecution<T>> executions = new ArrayList<>();
        fetchScheduledExecutionsForTask(taskName, dataClass, executions::add);
        return executions;
    }

    /**
     * @see #fetchScheduledExecutionsForTask(String, Class, Consumer)
     */
    default <T> List<ScheduledExecution<T>> getScheduledExecutionsForTask(String taskName, Class<T> dataClass, ScheduledExecutionsFilter filter) {
        List<ScheduledExecution<T>> executions = new ArrayList<>();
        fetchScheduledExecutionsForTask(taskName, dataClass, filter, executions::add);
        return executions;
    }

    /**
     * Gets the details for a specific scheduled execution. Currently running executions are also returned.
     *
     * @param taskInstanceId
     * @return Optional.empty() if no matching execution found
     * @see com.github.kagkarlsson.scheduler.task.TaskInstanceId
     * @see com.github.kagkarlsson.scheduler.ScheduledExecution
     */
    Optional<ScheduledExecution<Object>> getScheduledExecution(TaskInstanceId taskInstanceId);

    class Builder {

        private final DataSource dataSource;
        private List<Task<?>> knownTasks;
        private Serializer serializer = Serializer.DEFAULT_JAVA_SERIALIZER;
        private String tableName = JdbcTaskRepository.DEFAULT_TABLE_NAME;
        private JdbcCustomization jdbcCustomization;

        private Builder(DataSource dataSource, List<Task<?>> knownTasks) {
            this.dataSource = dataSource;
            this.knownTasks = knownTasks;
        }

        public static Builder create(DataSource dataSource, Task<?>... knownTasks) {
            return new Builder(dataSource, Arrays.asList(knownTasks));
        }

        public static Builder create(DataSource dataSource, List<Task<?>> knownTasks) {
            return new Builder(dataSource, knownTasks);
        }

        public Builder serializer(Serializer serializer) {
            this.serializer = serializer;
            return this;
        }

        public Builder tableName(String tableName) {
            this.tableName = tableName;
            return this;
        }

        public Builder jdbcCustomization(JdbcCustomization jdbcCustomization) {
            this.jdbcCustomization = jdbcCustomization;
            return this;
        }

        public SchedulerClient build() {
            TaskResolver taskResolver = new TaskResolver(StatsRegistry.NOOP, knownTasks);
            final SystemClock clock = new SystemClock();

            TaskRepository taskRepository = new JdbcTaskRepository(
                dataSource,
                false,
                ofNullable(jdbcCustomization).orElse(new DefaultJdbcCustomization()),
                tableName,
                taskResolver,
                new SchedulerClientName(),
                serializer,
                clock);

            return new StandardSchedulerClient(taskRepository, clock);
        }
    }

    class StandardSchedulerClient implements SchedulerClient {

        private static final Logger LOG = LoggerFactory.getLogger(StandardSchedulerClient.class);
        protected final TaskRepository taskRepository;
        private final Clock clock;
        private SchedulerClientEventListener schedulerClientEventListener;

        StandardSchedulerClient(TaskRepository taskRepository, Clock clock) {
            this(taskRepository, SchedulerClientEventListener.NOOP, clock);
        }

        StandardSchedulerClient(TaskRepository taskRepository, SchedulerClientEventListener schedulerClientEventListener, Clock clock) {
            this.taskRepository = taskRepository;
            this.schedulerClientEventListener = schedulerClientEventListener;
            this.clock = clock;
        }

        @Override
        public <T> void schedule(TaskInstance<T> taskInstance, Instant executionTime) {
            boolean success = taskRepository.createIfNotExists(SchedulableInstance.of(taskInstance, executionTime));
            if (success) {
                notifyListeners(ClientEvent.EventType.SCHEDULE, taskInstance, executionTime);
            }
        }
        @Override
        public <T> void schedule(SchedulableInstance<T> schedulableInstance) {
            schedule(schedulableInstance.getTaskInstance(), schedulableInstance.getNextExecutionTime(clock.now()));
        }

        @Override
        public void reschedule(TaskInstanceId taskInstanceId, Instant newExecutionTime) {
            reschedule(taskInstanceId, newExecutionTime, null);
        }

        @Override
        public <T> void reschedule(SchedulableInstance<T> schedulableInstance) {
            reschedule(schedulableInstance, schedulableInstance.getNextExecutionTime(clock.now()), schedulableInstance.getTaskInstance().getData());
        }

        @Override
        public <T> void reschedule(TaskInstanceId taskInstanceId, Instant newExecutionTime, T newData) {
            String taskName = taskInstanceId.getTaskName();
            String instanceId = taskInstanceId.getId();
            Optional<Execution> execution = taskRepository.getExecution(taskName, instanceId);
            if (execution.isPresent()) {
                if (execution.get().isPicked()) {
                    throw new TaskInstanceCurrentlyExecutingException(taskName, instanceId);
                }

                boolean success;
                if (newData == null) {
                    success = taskRepository.reschedule(execution.get(), newExecutionTime, null, null, 0);
                } else {
                    success = taskRepository.reschedule(execution.get(), newExecutionTime, newData, null, null, 0);
                }

                if (success) {
                    notifyListeners(ClientEvent.EventType.RESCHEDULE, taskInstanceId, newExecutionTime);
                }
            } else {
                throw new TaskInstanceNotFoundException(taskName, instanceId);
            }
        }

        @Override
        public void cancel(TaskInstanceId taskInstanceId) {
            String taskName = taskInstanceId.getTaskName();
            String instanceId = taskInstanceId.getId();
            Optional<Execution> execution = taskRepository.getExecution(taskName, instanceId);
            if (execution.isPresent()) {
                if (execution.get().isPicked()) {
                    throw new TaskInstanceCurrentlyExecutingException(taskName, instanceId);
                }

                taskRepository.remove(execution.get());
                notifyListeners(ClientEvent.EventType.CANCEL, taskInstanceId, execution.get().executionTime);
            } else {
                throw new TaskInstanceNotFoundException(taskName, instanceId);
            }
        }

        @Override
        public void fetchScheduledExecutions(Consumer<ScheduledExecution<Object>> consumer) {
            fetchScheduledExecutions(ScheduledExecutionsFilter.all().withPicked(false), consumer);
        }

        @Override
        public void fetchScheduledExecutions(ScheduledExecutionsFilter filter, Consumer<ScheduledExecution<Object>> consumer) {
            taskRepository.getScheduledExecutions(filter,
                execution -> consumer.accept(new ScheduledExecution<>(Object.class, execution)));
        }

        @Override
        public <T> void fetchScheduledExecutionsForTask(String taskName, Class<T> dataClass, Consumer<ScheduledExecution<T>> consumer) {
            fetchScheduledExecutionsForTask(taskName, dataClass, ScheduledExecutionsFilter.all().withPicked(false), consumer);
        }

        @Override
        public <T> void fetchScheduledExecutionsForTask(String taskName, Class<T> dataClass, ScheduledExecutionsFilter filter,
                                                        Consumer<ScheduledExecution<T>> consumer) {
            taskRepository.getScheduledExecutions(filter, taskName,
                execution -> consumer.accept(new ScheduledExecution<>(dataClass, execution)));
        }

        @Override
        public Optional<ScheduledExecution<Object>> getScheduledExecution(TaskInstanceId taskInstanceId) {
            Optional<Execution> e = taskRepository.getExecution(taskInstanceId.getTaskName(), taskInstanceId.getId());
            return e.map(oe -> new ScheduledExecution<>(Object.class, oe));
        }

        private void notifyListeners(ClientEvent.EventType eventType, TaskInstanceId taskInstanceId, Instant executionTime) {
            try {
                schedulerClientEventListener.newEvent(new ClientEvent(new ClientEvent.ClientEventContext(eventType, taskInstanceId, executionTime)));
            } catch (Exception e) {
                LOG.error("Error when notifying SchedulerClientEventListener.", e);
            }
        }
    }


    class SchedulerClientName implements SchedulerName {
        @Override
        public String getName() {
            return "SchedulerClient";
        }

    }

}