Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Repeat Operator #699

Merged
merged 4 commits into from
Dec 27, 2013
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions rxjava-core/src/main/java/rx/Observable.java
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
import rx.operators.OperationOnExceptionResumeNextViaObservable;
import rx.operators.OperationParallel;
import rx.operators.OperationParallelMerge;
import rx.operators.OperationRepeat;
import rx.operators.OperationReplay;
import rx.operators.OperationRetry;
import rx.operators.OperationSample;
Expand Down Expand Up @@ -1097,6 +1098,28 @@ public static Observable<Integer> range(int start, int count, Scheduler schedule
return from(Range.createWithCount(start, count), scheduler);
}

/**
* Repeats the observable sequence indefinitely.
* <p>
*
* @return The observable sequence producing the elements of the given sequence repeatedly and sequentially.
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229428(v=vs.103).aspx">MSDN: Observable.Repeat</a>
*/
public Observable<T> repeat() {
return this.repeat(Schedulers.currentThread());
}

/**
* Repeats the observable sequence indefinitely.
* <p>
* @param scheduler the scheduler to send the values on.
* @return The observable sequence producing the elements of the given sequence repeatedly and sequentially.
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229428(v=vs.103).aspx">MSDN: Observable.Repeat</a>
*/
public Observable<T> repeat(Scheduler scheduler) {
return create(OperationRepeat.repeat(this, scheduler));
}

/**
* Returns an Observable that calls an Observable factory to create its
* Observable for each new Observer that subscribes. That is, for each
Expand Down
71 changes: 71 additions & 0 deletions rxjava-core/src/main/java/rx/operators/OperationRepeat.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
/**
* Copyright 2013 Netflix, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package rx.operators;

import rx.Observable;
import rx.Observer;
import rx.Scheduler;
import rx.Subscription;
import rx.subscriptions.CompositeSubscription;
import rx.subscriptions.MultipleAssignmentSubscription;
import rx.util.functions.Action0;
import rx.util.functions.Action1;

public class OperationRepeat<T> implements Observable.OnSubscribeFunc<T> {

private final Observable<T> source;
private final Scheduler scheduler;

public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) {
return new OperationRepeat<T>(source, scheduler);
}

private OperationRepeat(Observable<T> source, Scheduler scheduler) {
this.source = source;
this.scheduler = scheduler;
}

@Override
public Subscription onSubscribe(final Observer<? super T> observer) {
final CompositeSubscription compositeSubscription = new CompositeSubscription();
final MultipleAssignmentSubscription innerSubscription = new MultipleAssignmentSubscription();
compositeSubscription.add(innerSubscription);
compositeSubscription.add(scheduler.schedule(new Action1<Action0>() {
@Override
public void call(final Action0 self) {
innerSubscription.set(source.subscribe(new Observer<T>() {

@Override
public void onCompleted() {
self.call();
}

@Override
public void onError(Throwable error) {
observer.onError(error);
}

@Override
public void onNext(T value) {
observer.onNext(value);
}
}));
}
}));
return compositeSubscription;
}
}
4 changes: 2 additions & 2 deletions rxjava-core/src/test/java/rx/ObservableTests.java
Original file line number Diff line number Diff line change
Expand Up @@ -37,11 +37,11 @@
import org.mockito.stubbing.Answer;

import rx.Observable.OnSubscribeFunc;
import rx.schedulers.TestScheduler;
import rx.observables.ConnectableObservable;
import rx.schedulers.Schedulers;
import rx.schedulers.TestScheduler;
import rx.subscriptions.BooleanSubscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
import rx.util.functions.Action1;
import rx.util.functions.Func0;
import rx.util.functions.Func1;
Expand Down
62 changes: 62 additions & 0 deletions rxjava-core/src/test/java/rx/operators/OperationRepeatTest.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
/**
* Copyright 2013 Netflix, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rx.operators;

import static org.junit.Assert.*;

import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import rx.Observable;
import rx.Observable.OnSubscribeFunc;
import rx.Observer;
import rx.Subscription;
import rx.schedulers.Schedulers;
import rx.subscriptions.Subscriptions;

public class OperationRepeatTest {

@Test
public void testRepetition() {
int NUM = 10;
final AtomicInteger count = new AtomicInteger();
int value = Observable.create(new OnSubscribeFunc<Integer>() {

@Override
public Subscription onSubscribe(Observer<? super Integer> o) {
o.onNext(count.incrementAndGet());
o.onCompleted();
return Subscriptions.empty();
}
}).repeat(Schedulers.threadPoolForComputation()).take(NUM).toBlockingObservable().last();

assertEquals(NUM, value);
}

@Test
public void testRepeatTake() {
Observable<Integer> xs = Observable.from(1, 2);
Object[] ys = xs.repeat(Schedulers.newThread()).take(4).toList().toBlockingObservable().last().toArray();
assertArrayEquals(new Object[] { 1, 2, 1, 2 }, ys);
}

@Test
public void testNoStackOverFlow() {
Observable.from(1).repeat(Schedulers.newThread()).take(100000).toBlockingObservable().last();
}

}