feat(core): Support multiple activators (#41)

spinnaker · Jun 1, 2018 · 7528847 · 7528847
1 parent f0ea49d
commit 7528847
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 15 deletions.
diff --git a/keiko-core/src/main/kotlin/com/netflix/spinnaker/q/QueueProcessor.kt b/keiko-core/src/main/kotlin/com/netflix/spinnaker/q/QueueProcessor.kt
@@ -38,13 +38,12 @@ class QueueProcessor(
   private val queue: Queue,
   private val executor: QueueExecutor<*>,
   private val handlers: Collection<MessageHandler<*>>,
-  private val activator: Activator,
+  private val activators: List<Activator>,
   private val publisher: EventPublisher,
   private val deadMessageHandler: DeadMessageCallback,
   private val fillExecutorEachCycle: Boolean = false,
   private val requeueDelay: Duration = Duration.ofSeconds(0),
-  private val requeueMaxJitter: Duration = Duration.ofSeconds(0),
-  private val enabled: Boolean = true
+  private val requeueMaxJitter: Duration = Duration.ofSeconds(0)
 ) {
   private val log: Logger = getLogger(javaClass)
   private val random: Random = Random()
@@ -116,8 +115,8 @@ class QueueProcessor(
   }
 
   private fun ifEnabled(fn: () -> Unit) {
-    if (enabled) {
-      activator.ifEnabled(fn)
+    if (activators.all { it.enabled }) {
+      fn.invoke()
     }
   }
 

diff --git a/keiko-core/src/test/kotlin/com/netflix/spinnaker/q/QueueProcessorTest.kt b/keiko-core/src/test/kotlin/com/netflix/spinnaker/q/QueueProcessorTest.kt
@@ -65,7 +65,7 @@ object QueueProcessorTest : Spek({
         queue,
         BlockingQueueExecutor(),
         emptyList(),
-        activator,
+        listOf(activator),
         publisher,
         deadMessageHandler
       )
@@ -83,9 +83,7 @@ object QueueProcessorTest : Spek({
 
     describe("when enabled") {
       beforeEachTest {
-        whenever(activator.ifEnabled(any())) doStub { callback: () -> Unit ->
-          callback.invoke()
-        }
+        whenever(activator.enabled) doReturn true
       }
 
       and("there is no capacity in the thread pool") {
@@ -95,7 +93,7 @@ object QueueProcessorTest : Spek({
           queue,
           executor,
           emptyList(),
-          activator,
+          listOf(activator),
           publisher,
           deadMessageHandler
         )
@@ -127,7 +125,7 @@ object QueueProcessorTest : Spek({
           queue,
           BlockingQueueExecutor(),
           listOf(simpleMessageHandler, parentMessageHandler),
-          activator,
+          listOf(activator),
           publisher,
           deadMessageHandler
         )

diff --git a/keiko-spring/src/main/kotlin/com/netflix/spinnaker/config/QueueConfiguration.kt b/keiko-spring/src/main/kotlin/com/netflix/spinnaker/config/QueueConfiguration.kt
@@ -18,6 +18,7 @@ package com.netflix.spinnaker.config
 
 import com.netflix.spinnaker.q.Activator
 import com.netflix.spinnaker.q.DeadMessageCallback
+import com.netflix.spinnaker.q.EnabledActivator
 import com.netflix.spinnaker.q.MessageHandler
 import com.netflix.spinnaker.q.Queue
 import com.netflix.spinnaker.q.QueueExecutor
@@ -59,23 +60,25 @@ class QueueConfiguration {
     queue: Queue,
     executor: QueueExecutor<*>,
     handlers: Collection<MessageHandler<*>>,
-    activator: Activator,
+    activators: List<Activator>,
     publisher: EventPublisher,
     queueProperties: QueueProperties,
     deadMessageHandler: DeadMessageCallback
   ) = QueueProcessor(
     queue,
     executor,
     handlers,
-    activator,
+    activators,
     publisher,
     deadMessageHandler,
     queueProperties.fillExecutorEachCycle,
     Duration.ofSeconds(queueProperties.requeueDelaySeconds),
-    Duration.ofSeconds(queueProperties.requeueMaxJitterSeconds),
-    queueProperties.enabled
+    Duration.ofSeconds(queueProperties.requeueMaxJitterSeconds)
   )
 
+  @Bean
+  fun enabledActivator(queueProperties: QueueProperties) = EnabledActivator(queueProperties.enabled)
+
   @Bean
   fun queueEventPublisher(
     applicationEventPublisher: ApplicationEventPublisher

diff --git a/keiko-spring/src/main/kotlin/com/netflix/spinnaker/q/EnabledActivator.kt b/keiko-spring/src/main/kotlin/com/netflix/spinnaker/q/EnabledActivator.kt
@@ -0,0 +1,18 @@
+/*
+ * Copyright 2018 Netflix, Inc.
+ *
+ * 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.netflix.spinnaker.q
+
+class EnabledActivator(override val enabled: Boolean = true) : Activator