@@ -0,0 +1,3100 @@
+package com.github.davidmoten.rx.operators;%0A%0Aimport rx.Notification;%0Aimport rx.Observable;%0Aimport rx.Observable.Operator;%0Aimport rx.Scheduler.Worker;%0Aimport rx.Subscriber;%0Aimport rx.Subscription;%0Aimport rx.functions.Action0;%0Aimport rx.schedulers.Schedulers;%0A%0Apublic class OperatorRetry%3CT%3E implements Operator%3CT, T%3E %7B%0A%0A    private final Observable%3CT%3E source;%0A%0A    public OperatorRetry(Observable%3CT%3E source) %7B%0A        this.source = source;%0A    %7D%0A%0A    @Override%0A    public Subscriber%3C? super T%3E call(Subscriber%3C? super T%3E child) %7B%0A        ParentSubscriber%3CT%3E parent = new ParentSubscriber%3CT%3E(source, child);%0A        parent.init();%0A        return parent;%0A    %7D%0A%0A    private static class ParentSubscriber%3CT%3E extends Subscriber%3CT%3E %7B%0A%0A        private final Subscriber%3C? super T%3E child;%0A        private final Observable%3CT%3E source;%0A        private final Worker worker;%0A%0A        public ParentSubscriber(Observable%3CT%3E source, Subscriber%3C? super T%3E child) %7B%0A            this.source = source;%0A            this.child = child;%0A            this.worker = Schedulers.trampoline().createWorker();%0A        %7D%0A%0A        void init() %7B%0A            subscribe();%0A        %7D%0A%0A        private void subscribe() %7B%0A            Action0 restart = new Action0() %7B%0A%0A                @Override%0A                public void call() %7B%0A                    Subscription sub = source.materialize().unsafeSubscribe(%0A                            new Subscriber%3CNotification%3CT%3E%3E() %7B%0A%0A                                @Override%0A                                public void onStart() %7B%0A%0A                                %7D%0A%0A                                @Override%0A                                public void onCompleted() %7B%0A                                    // TODO Auto-generated method stub%0A%0A                                %7D%0A%0A                                @Override%0A                                public void onError(Throwable e) %7B%0A                                    // TODO Auto-generated method stub%0A%0A                                %7D%0A%0A                                @Override%0A                                public void onNext(Notification%3CT%3E notification) %7B%0A                                    if (notification.hasValue())%0A                                        child.onNext(notification.getValue());%0A                                    else if (notification.isOnCompleted())%0A                                        child.onCompleted();%0A                                    else %7B%0A                                        // is error we resubscribe%0A                                        subscribe();%0A                                    %7D%0A                                %7D%0A                            %7D);%0A                %7D%0A            %7D;%0A            worker.schedule(restart);%0A        %7D%0A%0A        @Override%0A        public void onCompleted() %7B%0A            // TODO Auto-generated method stub%0A%0A        %7D%0A%0A        @Override%0A        public void onError(Throwable e) %7B%0A            // TODO Auto-generated method stub%0A%0A        %7D%0A%0A        @Override%0A        public void onNext(T t) %7B%0A            // TODO Auto-generated method stub%0A%0A        %7D%0A%0A    %7D%0A%0A%7D%0A

add OperatorRetry