Fixes #282.

changes.md

@@ -3,6 +3,7 @@
 - **API breaking change**: update signature of `onyx.api/await-job-completion` to take an opts map.
 - Allow functions in leaf position of a workflow. [#198](https://github.com/onyx-platform/onyx/issues/198)
 - Bug fix: flow-conditions retry default action should emit segments [#262](https://github.com/onyx-platform/onyx/issues/262)
+- API: New catalog entry option `:onyx/n-peers` to automatically expand to make `:onyx/min-peers` and `:onyx/max-peers` peers the same value. [#282](https://github.com/onyx-platform/onyx/issues/282)
 
 #### 0.7.3
 

doc/user-guide/information-model.md

@@ -47,6 +47,7 @@ This section specifies what a valid catalog, workflow, and flow conditions look
 |-----------------------|------------|----------------------|------------|-----------------------------------------------------------------------|
 |`:onyx/batch-timeout`  | `integer`  | `>= 0`               | `1000`     |                                                                       |
 |`:onyx/max-peers`      | `integer`  | `> 0`                |            |                                                                       |
+|`:onyx/n-peers`        | `integer`  | `> 0`                |            | Expands to make `:onyx/min-peers` and `:onyx/max-peers this value.    |
 |`:onyx/language`       | `keyword`  | `:clojure`, `:java`  | `:clojure` | Affects `:onyx/fn` and `:onyx/plugin` function and plugin resolution  |
 |`:onyx/restart-pred-fn`| `keyword`  | `any`                |            | Keyword pointing to function taking an exception which returns a boolean for whether a peer is restartable following that exception  |
 

src/onyx/api.clj

@@ -100,8 +100,17 @@
         exempt-tasks (filter (fn [task] (some #{(:name task)} exempt-set)) tasks)]
     (map :id exempt-tasks)))
 
+(defn ^{:no-doc true} expand-n-peers [catalog]
+  (mapv
+   (fn [entry]
+     (if-let [n (:onyx/n-peers entry)]
+       (assoc entry :onyx/min-peers n :onyx/max-peers n)
+       entry))
+   catalog))
+
 (defn ^{:no-doc true} create-submit-job-entry [id config job tasks]
   (let [task-ids (map :id tasks)
+        job (update job :catalog expand-n-peers)
         scheduler (:task-scheduler job)
         sat (saturation (:catalog job))
         task-saturation (task-saturation (:catalog job) tasks)

src/onyx/static/validation.clj

@@ -28,13 +28,19 @@
                   (:onyx/max-peers entry))
       (throw (ex-info ":onyx/min-peers must be <= :onyx/max-peers" {:entry entry})))))
 
+(defn min-max-n-peers-mutually-exclusive [entry]
+  (when (or (and (:onyx/min-peers entry) (:onyx/n-peers entry))
+            (and (:onyx/max-peers entry) (:onyx/n-peers entry)))
+    (throw (ex-info ":onyx/n-peers cannot be used with :onyx/min-peers or :onyx/max-peers" {:entry entry}))))
+
 (defn validate-catalog
   [catalog]
   (no-duplicate-entries catalog)
   (schema/validate Catalog catalog)
   (doseq [entry catalog]
     (name-and-type-not-equal entry)
-    (min-and-max-peers-sane entry)))
+    (min-and-max-peers-sane entry)
+    (min-max-n-peers-mutually-exclusive entry)))
 
 (defn validate-workflow-names [{:keys [workflow catalog]}]
   (when-let [missing-names (->> workflow