       (s/end! s)
       (s/end! sv))))
 
-;; ;; - bufferWithTime
-;; ;; - bufferWithCount
-;; ;; - bufferWithTimeOrCount
-;; ;; - toProperty
-
-;; ;; bus
-
 (t/deftest bus-push
   (t/async done
     (let [b (s/bus)]

      (drain! s #(t/is (= % [1])))
      (s/on-end s done))))

(t/deftest observable-never
  (t/async done
    (let [n (s/never)]
      (s/on-end n done))))

;; (t/deftest observable-interval
;;   (t/async done
;;     (let [i (s/interval 10)
;;           sample (s/take 4 i)]
;;       (drain! sample #(t/is (apply = %)))
;;       (s/on-end sample done))))

;; (t/deftest observable-later
;;   (t/async done
;;     (let [s (s/later 10 42)]
;;       (drain! s #(t/is (= % [42])))
;;       (s/on-end s done))))

;; (t/deftest observable-sequentially
;;   (t/async done
;;     (let [s (s/sequentially 10 [1 2 3])
;;           sample (s/take 6 s)]
;;       (drain! sample #(t/is (= % [1 2 3])))
;;       (s/on-end sample done))))

;; (t/deftest observable-repeatedly
;;   (t/async done
;;     (let [s (s/repeatedly 10 [1 2 3])
;;           sample (s/take 6 s)]
;;       (drain! sample #(t/is (= % [1 2 3 1 2 3])))
;;       (s/on-end sample done))))

(t/deftest observable-on-value
  (t/async done
    (let [s (s/from-coll [1 2 3])
          vacc (volatile! [])]
      (s/on-value s #(vswap! vacc conj %))
      (s/on-end s #(do (t/is (= @vacc [1 2 3]))
                       (done))))))

(t/deftest observable-concat
  (t/async done


;; event stream

(defn log
  [e ob]
  (s/tap (fn [v]
           (println "[log]:" e ":" v)
           v) ob))

(t/deftest observable-from-vector
  (t/async done
    (let [coll [1 2 3]
