[0.16.x] Possible bug in Sample(Observable) operator

[amazari]

The Observable.sample(Observable) operator behaves non-intuitively (at least to me):
Here is the javadoc description:
"Return an Observable that emits the results of sampling the items emitted by the source Observable whenever the specified sampler Observable emits an item or completes."

Which I understand as:
The resulting Observable emits the current value of src for each value emitted by sampler, repeating src last value if required (in case sampler frequency > src frequency).

But the current behavior as of 0.16.1 is:
When sampling a source Observable src using a sampler Observable, the resulting Observable res always emits only once the last value of src, even if sampler's frequency is superior to src's .

Code example:

val src = Observable.interval(1, TimeUnit.SECONDS);
val sampler = Observable.interval(1, TimeUnit.MILLISECONDS);
val res = src.sample(sampler);

res.subscribe(aLong -> { System.out.println("aLong = " + aLong)  });

Actual output:

aLong = 0
aLong = 1
aLong = 2
aLong = 3
aLong = 4
...

Expected output:

aLong = 0 (thousand times)
aLong = 1 (thousand times)
aLong = 2 (thousand times)
aLong = 3 (thousand times)
aLong = 4 (thousand times)

Is my understanding of the javadoc wrong orcan we concider the current behavior as a bug ? Was it already reported and fixed in 0.17 ?

Thanks for your hard work!

[rvanheest]

Yes, this is a bug. Together with @headinthebox I also just found this out.

We found that the following Java code prints out infinite amounts of 3's.

Observable.concat(Observable.from(1, 2, 3), Observable.never()).sample(1L, TimeUnit.SECONDS).subscribe(System.out::println);

The .NET variant works correctly:

using System;
using System.Reactive.Linq;

namespace Playground
{
    class MainClass
    {
        public static void Main (string[] args)
        {  
            var xs = new[]{ 1, 2, 3 }.ToObservable ().
            Concat (Observable.Never<int> ());

            var ys = xs.Sample (TimeSpan.FromMilliseconds (10));

            ys.Subscribe (
                Console.WriteLine,
                e => Console.WriteLine (e.Message),
                () => Console.WriteLine ("|")
            );

            Console.ReadLine ();

        }
    }

}

and only prints 3 once!

[amazari]

The marble diagram describes correctly the current behavior, but seems to contradict the text description stating that a sample is taken "whenever the specified sampler Observable emits an item or completes."

My current use case requires the behavior as defined in letter and is more intuitive to me. If the current one is the expected one, I guess the javadoc should be altered and a new operator/override could be provided.

I haven't checked the rxnet behavior and the introtorx sample [0] only shows the case where sampler frequency < src frequency.

[0] http://www.introtorx.com/content/v1.0.10621.0/13_TimeShiftedSequences.html#Sample

[amazari]

@rvanheest Good to know you are looking into it guys !

[amazari]

If this is is indeed considered as a bug, I thing the culprit lays at https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/operators/OperationSample.java#L197

For some reason the code especially guards against sending the same source value twice using the valueTaken boolean.

[akarnokd]

It was a bit odd at the time, but I think L197 does this value-once thing correctly. L85 does not clear the value flag hence it will repeat the last value indefinitely.

[amazari]

@akarnokd @DavidMGross Thanks for your quick answers and reaction.
The javadoc patch does indeed describe accurately the current behavior.

Still, the emit-value-from-source-whenever-sampler-emits behavior could also prove useful, would such a variant be considerable for inclusion ?

[benjchristensen]

We can get this fixed fairly easily in the next release, it should not be emitting anything (repeating the last value) if nothing has been emitted in a given time window from the source.

[benjchristensen]

This should be fixed as of 0.18.2 based on some basic testing I did. Can someone confirm?

[rvanheest]

Yes, I can confirm that this bug is fixed now. When running the same program as shown above, I now get 3 only printed once, instead of over and over again. Also while running other programs that used this method, I noticed that the behavior is now correct. Thanks for fixing it!

public class RxSampleTest {

    public static void main(String[] args) throws IOException {
        Observable.concat(Observable.from(1, 2, 3), Observable.never()).sample(1L, TimeUnit.SECONDS).subscribe(System.out::println);
        System.in.read();
    }
}

[benjchristensen]

Thanks for confirming @rvanheest