Nondeterministically off by one

[luminusian]

In the output of the code below, on-value-total is on-value plus or minus one.

(def subject (rx/subject))

(def stream
  (rx/take 1000 subject))

(rx/subscribe stream
              #(println "on-value:" %))

(def total-stream
  (->> stream
       (rx/filter number?)
       (rx/map (constantly 1))
       (rx/reduce + 0)))

(rx/subscribe total-stream
              #(println "on-value-total:" %))

(do (future (run! (partial rx/push! subject)
                  (range 1 1000)))
    (future (run! (partial rx/push! subject)
                  (repeat 1000 :foo))))

One possible output:

on-value: 458
on-value-total: 459

Another possible output:

on-value: 580
on-value-total: 579

[niwinz]

Seems like some inconsistencies on rxjava and subjects, Seems like it does not push down the same number of objects on each subscription...

I have plan to update to rxjava 2.x in the near future, maybe it will be fixed.

[jespino]

Hi, this is related to a bad usage of Subject object. In the RxJava documentation (https://github.com/ReactiveX/RxJava/wiki/Subject#serializing), you can read that the Subject implementation isn't thread-safe. Only the SerializedSubject work as thread-safe, so to easy it, I have added the to-serialized function in the PR #23.

SerializerSubject add some buffering to the push, so if you want to have a similar behavior in your example code, you have to make the lock at pushing, for example, using this code:

(do 
  (future (run! (fn [item]
                  (locking subject
                    (rx/push! subject item)))
                (range 1 1000)))
  (future (run! (fn [item]
                  (locking subject
                    (rx/push! subject item)))
                (repeat 1000 :foo))))