Move to new pipeline-state interface for consuming step-result updates (

#131)

project.clj

@@ -62,6 +62,7 @@
                                        [ring-mock "0.1.5"]
                                        [org.clojars.runa/conjure "2.2.0"]
                                        [prismatic/dommy "1.0.0"]
+                                       [com.gearswithingears/shrubbery "0.4.1"]
                                        [http-kit "2.1.19"]
                                        [ring-server "0.3.1"]
                                        [ring/ring-devel "1.3.2"]

src/clj/lambdacd/state/core.clj

@@ -0,0 +1,15 @@
+(ns lambdacd.state.core
+  "Facade for all functions related to dealing with LambdaCDs state. Wraps the related interfaces to simplify compatibility and API."
+  (:require [lambdacd.state.protocols :as protocols]
+            [lambdacd.internal.pipeline-state :as legacy-pipeline-state]))
+
+(defn state-component [ctx]
+  (:pipeline-state-component ctx))
+
+(defn consume-step-result-update
+  "update a step-result in the state"
+  [ctx build-number step-id step-result]
+  (let [component (state-component ctx)]
+    (if (satisfies? lambdacd.state.protocols/StepResultUpdateConsumer component)
+      (protocols/consume-step-result-update component build-number step-id step-result)
+      (legacy-pipeline-state/update component build-number step-id step-result))))

src/clj/lambdacd/state/internal/pipeline_state_updater.clj

@@ -1,26 +1,24 @@
 (ns lambdacd.state.internal.pipeline-state-updater
-  (:refer-clojure :exclude [update])
   (:require [clojure.tools.logging :as log]
             [clojure.core.async :as async]
             [lambdacd.event-bus :as event-bus]
-            [lambdacd.internal.pipeline-state :refer [update]]
+            [lambdacd.state.core :as state]
             [lambdacd.util :as util]))
 
 (defn start-pipeline-state-updater [ctx]
   (let [step-updates-channel (util/buffered
                                (event-bus/only-payload
                                  (event-bus/subscribe ctx :step-result-updated)))
         stop-updater-channel (event-bus/only-payload
-                               (event-bus/subscribe ctx :stop-pipeline-state-updater))
-        pipeline-state (:pipeline-state-component ctx)]
+                               (event-bus/subscribe ctx :stop-pipeline-state-updater))]
     (async/go-loop []
-                   (if-let [[step-result-update ch] (async/alts! [step-updates-channel stop-updater-channel])]
-                     (when (not= stop-updater-channel ch)
-                       (let [step-result  (:step-result step-result-update)
-                             build-number (:build-number step-result-update)
-                             step-id      (:step-id step-result-update)]
-                         (update pipeline-state build-number step-id step-result)
-                         (recur)))))))
+      (if-let [[step-result-update ch] (async/alts! [step-updates-channel stop-updater-channel])]
+        (when (not= stop-updater-channel ch)
+          (let [step-result  (:step-result step-result-update)
+                build-number (:build-number step-result-update)
+                step-id      (:step-id step-result-update)]
+            (state/consume-step-result-update ctx build-number step-id step-result)
+            (recur)))))))
 
 (defn stop-pipeline-state-updater [ctx]
   (log/info "Shutting down pipeline state updater...")

src/clj/lambdacd/state/protocols.clj

@@ -0,0 +1,7 @@
+(ns lambdacd.state.protocols
+  "defines protocols that need to be implemented by a state component")
+
+(defprotocol StepResultUpdateConsumer
+  "components implementing this protocol can provide the state of a pipeline"
+  (consume-step-result-update [self build-number step-id step-result]))
+

test/clj/lambdacd/state/core_test.clj

@@ -0,0 +1,23 @@
+(ns lambdacd.state.core-test
+  (:require [clojure.test :refer :all]
+            [shrubbery.core :refer [mock received?]]
+            [lambdacd.state.core :as s]
+            [lambdacd.testsupport.data :refer [some-ctx-with]]
+            [lambdacd.state.protocols :as state-protocols]
+            [lambdacd.internal.pipeline-state :as legacy-pipeline-state]))
+
+(def some-build-number 42)
+(def some-step-id [0])
+(def some-step-result {:foo :bat})
+
+(deftest consume-step-result-update-test
+  (testing "that calls to a StepResultUpdateConsumer will just pass through"
+    (let [component (mock state-protocols/StepResultUpdateConsumer)]
+      (s/consume-step-result-update (some-ctx-with :pipeline-state-component component)
+                                    some-build-number some-step-id some-step-result)
+    (is (received? component state-protocols/consume-step-result-update [some-build-number some-step-id some-step-result]))))
+  (testing "that calls to a legacy PipelineStateComponent be mapped to that method"
+    (let [component (mock legacy-pipeline-state/PipelineStateComponent)]
+      (s/consume-step-result-update (some-ctx-with :pipeline-state-component component)
+                                    some-build-number some-step-id some-step-result)
+    (is (received? component legacy-pipeline-state/update [some-build-number some-step-id some-step-result])))))