Optimize OperatorSkipLastTimed

[zsxwing]

Changed OperatorSkipLastTimed to only cache the latest items in the specified time window.

[cloudbees-pull-request-builder]

RxJava-pull-requests #981 SUCCESS
This pull request looks good

[zsxwing]

I'm not sure if we need to call emitItemsOutOfWindow in onError. Which one is more reasonable?

[akarnokd]

I'm doing buffer now and they all emit buffered data in case of onError.

[zsxwing]

Tested in Rx.Net.

            var o = Observable.Create<int>(observer => {
                observer.OnNext(2);
                Thread.Sleep(2000);
                observer.OnError(new Exception("test"));
                return Disposable.Empty;
            });
            o.SkipLast(TimeSpan.FromMilliseconds(100)).ObserveOn(Scheduler.NewThread).Subscribe(
                next=>Console.WriteLine(next),
                e => Console.WriteLine(e),
                () => Console.WriteLine("onCompleted")
                );
            Console.ReadLine();

The above codes only output

System.Exception: test

So Rx.Net only emits onError and drops the buffer?

[benjchristensen]

When onError occurs it immediately emits and does not work any further work.

We had this discussion a while back when debating delay I think.

Rx Design Guideline 6.6

6.6. OnError messages should have abort semantics

As normal control flow in .NET uses abort semantics for exceptions (the stack is unwound, current code path is interrupted), Rx mimics this behavior. To ensure this behavior, no messages should be sent out by an operator once one of it sources has an error message or an exception is thrown within the operator.

...

In this sample, a buffering operator will abandon the observable sequence as soon as the subscription to source encounters an error. The current buffer is not sent to any subscribers, maintain abort semantics.

[zsxwing]

Thank you for clarification. Then this PR should be ready to merge.