        public void execute(CancellationToken cancelToken) {
            assert eventScheduler.isExecutingInThis();
            OrderedData<DataType> data = pollData();
            if (data == null) {
                return;
            }
            if (forwardedData && data.getIndex() <= lastSentIndex) {
                return;
            }
            if (done) {
                return;
            }
            wrappedListener.onDataArrive(data.getRawData());
        }

        public void execute(CancellationToken cancelToken) {
            assert eventScheduler.isExecutingInThis();
            OrderedData<DataType> data = pollData();
            if (data == null) {
                return;
            }
            if (forwardedData && data.getIndex() <= lastSentIndex) {
                return;
            }
            if (done) {
                return;
            }
            lastSentIndex = data.getIndex();
            forwardedData = true;
            wrappedListener.onDataArrive(data.getRawData());
        }
