diff options
| -rw-r--r-- | src/weepinbell/processor/facade.clj | 5 | ||||
| -rw-r--r-- | test/weepinbell/processor_starter_test.clj | 62 |
2 files changed, 50 insertions, 17 deletions
diff --git a/src/weepinbell/processor/facade.clj b/src/weepinbell/processor/facade.clj index 29febc2..1360b30 100644 --- a/src/weepinbell/processor/facade.clj +++ b/src/weepinbell/processor/facade.clj @@ -1,6 +1,6 @@ (ns weepinbell.processor.facade
(:require [clojure.spec.alpha :as s]
- [clojure.core.async :refer [go-loop <! put! close! chan <!!]]
+ [clojure.core.async :refer [go-loop <! put! chan <!!]]
[taoensso.timbre :as timbre]
[weepinbell.utils.digest :refer [digest]]
[weepinbell.storage.spec]
@@ -15,6 +15,9 @@ (s/def ::callbacks
(s/keys :req-un [::on-receive ::on-create ::on-delete ::on-update]))
(s/def ::storage (partial s/valid? :weepinbell.storage.spec/vtable))
+(s/def ::base-url string?)
+(s/def ::allowed-target-domains sequential?)
+(s/def ::pipeline-concurrency int?)
(s/def ::config
(s/keys :req-un [::base-url ::allowed-target-domains ::pipeline-concurrency]))
(s/def ::processor-meta
diff --git a/test/weepinbell/processor_starter_test.clj b/test/weepinbell/processor_starter_test.clj index 8b604a5..500bd9f 100644 --- a/test/weepinbell/processor_starter_test.clj +++ b/test/weepinbell/processor_starter_test.clj @@ -1,12 +1,17 @@ (ns weepinbell.processor-starter-test
(:require [clojure.test :refer [deftest testing is]]
- [clojure.core.async :refer [chan put!]]
+ [clojure.core.async :refer [chan put! poll!]]
[weepinbell.processor.facade :refer [start-processor]]))
-(defn- test-processor [test-wmc received?]
- {:pre [(instance? clojure.lang.IPending received?)]}
+(defn- test-processor
+ ([test-wmc received?]
+ (test-processor test-wmc received? 1 nil))
+ ([test-wmc received? concurrency]
+ (test-processor test-wmc received? concurrency nil))
+ ([test-wmc received? concurrency on-receive]
(let [pmeta {:wmc test-wmc
- :callbacks {:on-receive (fn [_] (deliver received? true))
+ :callbacks {:on-receive (or on-receive
+ (fn [_] (put! received? true)))
:on-create (fn [_])
:on-delete (fn [_])
:on-update (fn [_])}
@@ -17,22 +22,47 @@ :get-by-domain (fn [& _])}
:config {:base-url "https://test.com"
:allowed-target-domains ["test.com"]
- :pipeline-concurrency 1}}]
- (start-processor pmeta)))
+ :pipeline-concurrency concurrency}}]
+ (start-processor pmeta))))
(deftest start-processor-test
- (testing "graceful shutdown"
- (let [test-wmc (chan)
- received? (promise)
+ (testing "processor starting and taking webmention"
+ (let [test-wmc (chan), received? (chan)
shutdown (test-processor test-wmc received?)]
- (shutdown)
(put! test-wmc {:s "s" :t "t"})
- (is (not (deref received? 50 false)))))
- (testing "processor starting and taking webmention"
- (let [test-wmc (chan)
- received? (promise)
+ (Thread/sleep 50)
+ (is (poll! received?))
+ (shutdown)))
+ (testing "graceful shutdown, should not take webmention after shutdown"
+ (let [test-wmc (chan), received? (chan)
shutdown (test-processor test-wmc received?)]
+ (shutdown)
(put! test-wmc {:s "s" :t "t"})
- (is (deref received? 50 false))
- (shutdown))))
+ (Thread/sleep 50)
+ (is (nil? (poll! received?))))))
+
+(deftest concurrent-processor-test
+ (testing "processor processing all webmentions before shutting down"
+ (let [test-wmc (chan), received? (chan)
+ concurrency 3
+ shutdown (test-processor test-wmc received? 3)]
+ (is (fn? shutdown))
+ (dotimes [_ concurrency] (put! test-wmc {:s "s" :t "t"}))
+ (shutdown)
+ (Thread/sleep 50)
+ (dotimes [_ concurrency] (is (poll! received?)))))
+ (testing "processor still works if Exceptions were thrown"
+ (let [test-wmc (chan), received? (chan), fail-once (chan)]
+ (put! fail-once true)
+ (let [on-receive (fn [_]
+ ;; throw Exception at the first call
+ (if (poll! fail-once)
+ (throw (Exception.))
+ (put! received? true)))
+ concurrency 5
+ shutdown (test-processor test-wmc received? concurrency on-receive)]
+ (dotimes [_ concurrency] (put! test-wmc {:s "s" :t "t"}))
+ (Thread/sleep 50)
+ (dotimes [_ (- concurrency 1)] (is (poll! received?)))
+ (shutdown)))))
|