Rxjava: 2.x: confused of method takeLast(long,TimeUnit)

[FlyingPigQAQ]

I'm a little confused about takeLast(long, TimeUnit).
When executing the following code, each time the output order is supposed to be N digits in the last time window, but in fact the output results are inconsistent with the ideal state.

Code

 public static void main(String[] args) {
        ArrayList<Integer> ints = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            ints.add(i);
        }
        Observable.fromIterable(ints)
                .takeLast(1,TimeUnit.NANOSECONDS)
                .subscribe(longs->{
                    System.out.println(longs);
                });
        try {
            Thread.sleep(100000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

Result

[akarnokd]

Hi and thanks for reporting. This is a bug in takeLast because the internal time window calculation is incorrect. I'll post a fix shortly.

[FlyingPigQAQ]

@akarnokd Hello! I would like to know that the takeLast() method is to get the data of the last window that was sent. We already got the data of the last window in the OnNext() method, so why do we have to make another time difference judgment in the onComplete() method

[akarnokd]

I don't understand your question. TakeLast takes those last items that have been received between the onComplete and some time before:

source       --o-o-o------o-o-o-o-o---|
time window                  ^========^
result       -----------------o-o-o---|

[FlyingPigQAQ]

@akarnokd Sorry, I mean the last time window data filtering has been done in the following onNext() code

public void onNext(T t) {
            final SpscLinkedArrayQueue<Object> q = queue;

            long now = scheduler.now(unit);
            long time = this.time;
            long c = count;
            boolean unbounded = c == Long.MAX_VALUE;

            q.offer(now, t);

            while (!q.isEmpty()) {
                long ts = (Long)q.peek();
                if (ts <= now - time || (!unbounded && (q.size() >> 1) > c)) {
                    q.poll();
                    q.poll();
                } else {
                    break;
                }
            }
        }

So why are we even checking onComplete?

                T o = (T)q.poll();

                if ((Long)ts < scheduler.now(unit) - time) {
                    continue;
                }

                a.onNext(o);

What I'm reading is that takeLast() is the last time window of observable's entire life cycle, but I look at the source code calling a.onNext(o) and then calling the Observer's accept () method, so that means takeLast() and observer has a coupling. I don't know if my expression is clear enough and thank you for your patient explanation.

[akarnokd]

Because onComplete can happen much later than the previous onNext, thus that value is now out of the time window.

[FlyingPigQAQ]

Okay,Thanks!