core.async difference between pipe and pipeline

channels.clj

(into [] (partition-by #(not= % :spliter)) [1 2 3 4 :spliter 4 3 2 1 :spliter])

;; ^^^^ this is like the base case. output is [[1 2 3 4] [:spliter] [4 3 2 1] [:spliter]]


(def in (chan))
(def out (chan 1 (partition-by #(not= % :spliter))))

(pipe in out)

;; ^^^^ works in the same way
;; the out chan collects stuff and dumps a collection of values
;; so that out chan will have 4 vals at the end [1 2 3 4], [:spliter], [4 3 2 1] and [:spliter]
;; last [:spliter] is only on chan close ofc...


(def in (chan))
(def out (chan))

(pipeline 1 out (partition-by #(not= % :spliter)) in)

;; ^^^^ not really what i'd expect. 
;; basically this just runs the parition-by on each element
;; in essence transforming each input element into single-element vector
;; out chan will have 10 vals at the end of everything [1], [2], [3], [4], [:spliter], [4], [3]...

;; this is because each input is processed idependently 
;; the '1' passed as the first arg is the paralelization factor. i.e. 1 => do 1 thing is parallel, 5 => do 5 things in parallel


(comment
  
  (go-loop []
    (when-not (closed? out)
      (println (<! out))
      (recur)))


  (go
    (doseq [x [1 2 3 4 :spliter 4 3 2 1 :spliter]]
      (>! in x)))

)