@@ -0,0 +1,2419 @@
+/**%0A * Copyright 2013 Netflix, Inc.%0A * %0A * Licensed under the Apache License, Version 2.0 (the %22License%22);%0A * you may not use this file except in compliance with the License.%0A * You may obtain a copy of the License at%0A * %0A * http://www.apache.org/licenses/LICENSE-2.0%0A * %0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage rx.operators;%0A%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport rx.Observable.OnSubscribeFunc;%0Aimport rx.Observer;%0Aimport rx.Scheduler;%0Aimport rx.Subscription;%0Aimport rx.concurrency.Schedulers;%0Aimport rx.subscriptions.Subscriptions;%0Aimport rx.util.functions.Action0;%0A%0Apublic final class OperationTimer %7B%0A%0A%09public static OnSubscribeFunc%3CVoid%3E timer(long interval, TimeUnit unit) %7B%0A        return timer(interval, unit, Schedulers.threadPoolForComputation());%0A    %7D%0A%0A    public static OnSubscribeFunc%3CVoid%3E timer(final long delay, final TimeUnit unit, final Scheduler scheduler) %7B%0A        return new OnSubscribeFunc%3CVoid%3E() %7B%0A            @Override%0A            public Subscription onSubscribe(Observer%3C? super Void%3E observer) %7B%0A                return new Timer(delay, unit, scheduler, observer).start();%0A            %7D%0A        %7D;%0A    %7D%0A%09%0A%09private static class Timer %7B%0A        private final long period;%0A        private final TimeUnit unit;%0A        private final Scheduler scheduler;%0A        private final Observer%3C? super Void%3E observer;%0A        %0A        private Timer(long period, TimeUnit unit, Scheduler scheduler, Observer%3C? super Void%3E observer) %7B%0A            this.period = period;%0A            this.unit = unit;%0A            this.scheduler = scheduler;%0A            this.observer = observer;%0A        %7D%0A%0A        public Subscription start() %7B%0A            final Subscription s = scheduler.schedule(new Action0() %7B%0A                @Override%0A                public void call() %7B%0A                    observer.onNext(null);%0A                    observer.onCompleted();%0A                %7D%0A            %7D, period, unit);%0A%0A            return Subscriptions.create(new Action0() %7B%0A                @Override%0A                public void call() %7B%0A                    s.unsubscribe();%0A                %7D%0A            %7D);%0A        %7D%0A    %7D%0A%0A%7D%0A

add OperationTimer