실행단계 최초 구현

src/main/kotlin/org/javabom/bomscheduler/config/JobManagerConfig.kt

@@ -0,0 +1,44 @@
+package org.javabom.bomscheduler.config
+
+import org.javabom.bomscheduler.job.JobCollection
+import org.javabom.bomscheduler.job.JobCoordinator
+import org.javabom.bomscheduler.job.JobManager
+import org.javabom.bomscheduler.schedule.BomSchedule
+import org.javabom.bomscheduler.schedule.BomScheduleInterceptor
+import org.javabom.bomscheduler.task.TaskManager
+import org.javabom.bomscheduler.task.TaskRunner
+import org.springframework.aop.Advisor
+import org.springframework.aop.support.DefaultPointcutAdvisor
+import org.springframework.aop.support.annotation.AnnotationMatchingPointcut
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.Import
+
+@Configuration
+class JobManagerConfig {
+
+    @Bean
+    fun jobManager(jobCoordinator: JobCoordinator, jobCollection: JobCollection): JobManager {
+        return JobManager(jobCoordinator, jobCollection)
+    }
+
+    @Bean
+    fun taskRunner(): TaskRunner {
+        return TaskRunner()
+    }
+
+    @Bean
+    fun TaskManager(
+        jobManager: JobManager,
+        jobCollection: JobCollection
+    ): TaskManager {
+        return TaskManager(jobManager, jobCollection, taskRunner())
+    }
+
+    @Bean
+    fun bomScheduleInterceptor(): Advisor {
+        val interceptor = BomScheduleInterceptor(taskRunner())
+        val pointcut = AnnotationMatchingPointcut(BomSchedule::class.java)
+        return DefaultPointcutAdvisor(pointcut, interceptor)
+    }
+}

src/main/kotlin/org/javabom/bomscheduler/job/Job.kt

@@ -9,4 +9,10 @@ data class Job(
     val version: Int = 0,
     val lastExecutionTime: LocalDateTime = LocalDateTime.now(),
     val instanceExpiredTime: LocalDateTime? = null
-)
+) {
+
+    fun isExpired(): Boolean {
+        return instanceExpiredTime
+            ?.isBefore(LocalDateTime.now())?: false
+    }
+}

src/main/kotlin/org/javabom/bomscheduler/job/JobCollection.kt

@@ -1,8 +1,16 @@
 package org.javabom.bomscheduler.job
 
-class JobCollection(definedJobs : List<Job> = emptyList()) {
+class JobCollection(
+    private var definedJobs: List<Job> = emptyList()
+) {
 
     init {
         definedJobs.forEach{ println("Register Job ${it.name}")}
     }
+
+    fun getDefinedJobs(): List<Job> = definedJobs
+
+    fun replaceJobs(jobs: List<Job>) {
+        definedJobs = jobs
+    }
 }

src/main/kotlin/org/javabom/bomscheduler/job/JobManager.kt

@@ -0,0 +1,63 @@
+package org.javabom.bomscheduler.job
+
+import org.javabom.bomscheduler.config.ScheduleConfig
+import org.springframework.context.SmartLifecycle
+import java.time.LocalDateTime
+
+class JobManager(
+    private val jobCoordinator: JobCoordinator,
+    private val jobCollection: JobCollection
+): SmartLifecycle {
+
+    private var running = false
+    private var keepGoing = false
+
+    override fun start() {
+        running = true
+        keepGoing = true
+
+        Thread {
+            updateJob()
+        }.start()
+    }
+
+    override fun stop() {
+        this.keepGoing = false
+    }
+
+    override fun isRunning(): Boolean {
+        return running
+    }
+
+    private fun updateJob() {
+        while (keepGoing) {
+            val (currentJobs, anotherJobs) = jobCoordinator.getDefinedJob()
+                .partition { it.instanceName == ScheduleConfig.HOST_NAME }
+
+            val now = LocalDateTime.now()
+
+            val exchangedJobs = anotherJobs.filter { it.isExpired() }
+                .map {
+                    val ttl = it.ttl
+
+                    it.copy(
+                        instanceName = ScheduleConfig.HOST_NAME,
+                        instanceExpiredTime = now.plusNanos(ttl)
+                    )
+                }
+
+
+            if (exchangedJobs.isNotEmpty()) {
+                jobCoordinator.updateJobs(exchangedJobs)
+            }
+
+            jobCollection.replaceJobs(currentJobs + exchangedJobs)
+        }
+
+        running = false
+    }
+
+    fun updateJobExecutionTime(job: Job) {
+        jobCoordinator.updateJobs(listOf(job))
+    }
+}

src/main/kotlin/org/javabom/bomscheduler/job/JobMetadata.kt

@@ -0,0 +1,9 @@
+package org.javabom.bomscheduler.job
+
+import java.time.LocalDateTime
+
+data class JobMetadata(
+    val jobName: String,
+    val lastExecutionTime: LocalDateTime,
+    val ttl: Long
+)

src/main/kotlin/org/javabom/bomscheduler/schedule/BomScheduleInterceptor.kt

@@ -0,0 +1,23 @@
+package org.javabom.bomscheduler.schedule
+
+import org.aopalliance.intercept.MethodInterceptor
+import org.aopalliance.intercept.MethodInvocation
+import org.javabom.bomscheduler.task.TaskManager
+import org.javabom.bomscheduler.task.TaskRunner
+
+class BomScheduleInterceptor(
+    private val taskRunner: TaskRunner
+) : MethodInterceptor {
+    override fun invoke(invocation: MethodInvocation): Any? {
+        val jobName = invocation.method.getAnnotation(BomSchedule::class.java).jobName
+
+        val task = taskRunner.findTaskByJobName(jobName)
+
+        if (task != null) {
+            task.consuming()
+            return invocation.proceed()
+        }
+
+        return null
+    }
+}

src/main/kotlin/org/javabom/bomscheduler/task/Task.kt

@@ -0,0 +1,11 @@
+package org.javabom.bomscheduler.task
+
+data class Task (
+    val jobName: String,
+    val callback: () -> Unit
+) {
+
+    fun consuming() {
+        callback.invoke()
+    }
+}

src/main/kotlin/org/javabom/bomscheduler/task/TaskManager.kt

@@ -0,0 +1,62 @@
+package org.javabom.bomscheduler.task
+
+import org.javabom.bomscheduler.config.ScheduleConfig
+import org.javabom.bomscheduler.job.JobCollection
+import org.javabom.bomscheduler.job.JobManager
+import org.springframework.context.SmartLifecycle
+import java.time.LocalDateTime
+
+class TaskManager(
+    private val jobManager: JobManager,
+    private val jobCollection: JobCollection,
+    private val taskRunner: TaskRunner
+): SmartLifecycle {
+    private var running = false
+    private var keepGoing = false
+
+
+    override fun start() {
+        running = true
+        keepGoing = true
+
+        Thread {
+            createExecutableTasks()
+        }.start()
+    }
+
+    override fun stop() {
+        keepGoing = false
+    }
+
+    override fun isRunning(): Boolean {
+        return running
+    }
+
+    private fun createExecutableTasks() {
+        val executableTasks = jobCollection.getDefinedJobs()
+            .filter { it.instanceName == ScheduleConfig.HOST_NAME }
+            .map {
+                Task(
+                    jobName = it.name,
+                    callback = {
+                        val executionTime = LocalDateTime.now()
+                        val ttl = it.ttl
+
+                        taskRunner.consumeTask(it.name)
+                        jobManager.updateJobExecutionTime(
+                            it.copy(
+                                lastExecutionTime = executionTime,
+                                instanceExpiredTime = executionTime.plusNanos(ttl)
+                            )
+                        )
+                    }
+                )
+            }
+
+        while (keepGoing) {
+            taskRunner.putTasks(executableTasks)
+        }
+
+        running = false
+    }
+}

src/main/kotlin/org/javabom/bomscheduler/task/TaskRunner.kt

@@ -0,0 +1,19 @@
+package org.javabom.bomscheduler.task
+
+import org.javabom.bomscheduler.job.Job
+import org.javabom.bomscheduler.job.JobMetadata
+import org.springframework.stereotype.Component
+
+class TaskRunner {
+    private val executableTasks: MutableMap<String, Task> = mutableMapOf()
+
+    fun putTasks(tasks: List<Task>) {
+        tasks.forEach {
+            executableTasks.putIfAbsent(it.jobName, it)
+        }
+    }
+
+    fun findTaskByJobName(jobName: String) = executableTasks[jobName]
+
+    fun consumeTask(jobName: String) = executableTasks.remove(jobName)
+}