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

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

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.

@akarnokd

@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

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---|

@akarnokd

@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.

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

Okay,Thanks!

Navigation Menu

Repository navigation

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

Code

Result

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

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

Description

Code

Result

Activity

akarnokd commented on Sep 11, 2019

FlyingPigQAQ commented on Sep 16, 2019

akarnokd commented on Sep 16, 2019

FlyingPigQAQ commented on Sep 16, 2019

akarnokd commented on Sep 16, 2019

FlyingPigQAQ commented on Sep 16, 2019

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions