   single element in the result sequence."
   ([f ob]
    #?(:cljs (.reduce ob #(f %1 %2))
-      :clj  (.reduce ob (as-bifunction f))))
+      :clj  (cond
+              (observable? ob)
+              (-> (.reduce ob (as-bifunction f))
+                  (as-observable))
+
+              (flowable? ob)
+              (-> (.reduce ob (as-bifunction f))
+                  (as-flowable))
+
+              :else (throw (IllegalArgumentException. "`ob` should be observable or flowable")))))
+
   ([f seed ob]
    #?(:cljs (.reduce ob #(f %1 %2) seed)
-      :clj  (.reduce ob seed (as-bifunction f)))))
+      :clj  (cond
+              (observable? ob)
+              (-> (.reduce ob seed (as-bifunction f))
+                  (as-observable))
+
+              (flowable? ob)
+              (-> (.reduce ob seed (as-bifunction f))
+                  (as-flowable))
+
+              :else (throw (IllegalArgumentException. "`ob` should be observable or flowable"))))))
 
 (defn scan
   "Applies an accumulator function over an observable

   #?(:cljs (.combineLatest ob other f)
      :clj  (let [^Iterable sources (list ob other)
                  ^Function combiner (as-function #(apply f (seq %)))]
              (Observable/combineLatest sources combiner)))))

(defn- unwrap-composite-exception
  [exc]

                           (let [v (xsink nil input)]
                             (when (reduced? v)
                               (xsink @v))))
                    disposable (on-value ob step)]
                (on-complete ob #(do (xsink nil)
                                     (sink end)))
                (fn []
                  (cancel! disposable)))))))

