OperatorDelay2.java

public final class OperatorDelay2<T> implements OnSubscribe<T> {
        final Observable<T> source;
        final long delay;
        final TimeUnit unit;
        final Scheduler scheduler;

        public Delay2(Observable<T> source, long delay, TimeUnit unit, Scheduler scheduler) {
            this.source = source;
            this.delay = delay;
            this.unit = unit;
            this.scheduler = scheduler;
        }

        @Override
        public void call(Subscriber<? super T> child) {
            final Worker w = scheduler.createWorker();
            child.add(w);
            
            Observable.concat(source.map(new Func1<T, Observable<T>>() {

                @Override
                public Observable<T> call(final T x) {
                    final AsyncSubject<T> result = AsyncSubject.create();
                    w.schedule(new Action0() {
                        @Override
                        public void call() {
                            result.onNext(x);
                            result.onCompleted();
                        }
                    }, delay, unit);
                    return result;
                }
            })).subscribe(child);
        }
    }