                                   [3 :3]])))
       (s/on-end sample done))))
 
-;; ;; interop
-
 (t/deftest pipe-to-atom
   (t/async done
     (let [st (s/from-coll [1 2 3])

      (s/end! b)
      (s/on-end b done))))


;; TODO:
;; (t/deftest bus-erro
;;   (t/async done
;;     (let [b (s/bus)]
;;       (t/is (s/bus? b))
;;       (drain! b #(t/is (= % [1]))
;;                 #(t/is (= % :oh-no)))
;;       (s/push! b 1)
;;       (s/error! b (ex-info "oh noes" {})
;;       (s/on-error b done))))

;; (t/deftest bus-plug
;;   (t/async done
;;     (let [b (s/bus)]
;;       (t/is (s/bus? b))
;;       (s/plug! b (s/from-coll [1 2 3]))
;;       (s/plug! b (s/from-coll [:four :five]))
;;       (drain! b #(t/is (= %
;;                           [1 2 3 :four :five])))
;;       (s/end! b)
;;       (s/on-end b done))))

(t/deftest filter-with-predicate
  (t/async done
    (let [s (s/from-coll [1 2 3 4 5])
          fs (s/filter #{3 5} s)]
      (drain! fs #(t/is (= % [3 5])))
      (s/on-end fs done))))

;; ;; ::todo decomplect
;; ;; - filter (pred, property)
;; ;; - map
;; ;; - mapError
;; ;; - errors
;; ;; - skipErrors
;; ;; - mapEnd
;; ;; - skipDuplicates
;; ;; - take
;; ;; - takeUntil
;; ;; - takeWhile
;; ;; - first
;; ;; - last
;; ;; - skip
;; ;; - delay
;; ;; - throttle
;; ;; - debounce
;; ;; - debounceImmediate
;; ;; - bufferingThrottle
;; ;; - doAction
;; ;; - doError
;; ;; - not
;; ;; - flatMap
;; ;; - flatMapLatest
;; ;; - flatMapFirst
;; ;; - flatMapError
;; ;; - flatMapWithConcurrencyLimit
;; ;; - flatMapConcat
;; ;; - scan
;; ;; - fold/reduce
;; ;; - diff
;; ;; - zip
;; ;; - slidingWindow
;; ;; - log
;; ;; - doLog
;; ;; - combine
;; ;; - withStateMachine
;; ;; - decode
;; ;; - awaiting
;; ;; - endOnError
;; ;; - withHandler
;; ;; - name
;; ;; - withDescription

;; ;; TODO

;; ;; - combineAsArray
;; ;; - combineWith
;; ;; - combineTemplate
;; ;; - mergeAll
;; ;; - zipAsArray
;; ;; - zipWith
;; ;; - onValues


;; ;; TODO: error handling
;; ;; - onError, ...
;; ;; - retry

;; ;; TODO: join
;; ;; - when

(t/deftest event-stream-as-functor
 (t/async done

      (s/end! s)
      (s/end! sv))))

(t/deftest bus-push
  (t/async done
    (let [b (s/bus)]
