Implement the 'Repeat' operator #498
Conversation
|
RxJava-pull-requests #422 SUCCESS |
| observer.onCompleted(); | ||
| } else { | ||
| ConnectableObservable<T> replayObservable = source | ||
| .replay(); |
There was a problem hiding this comment.
There is an approach with schedulers to do recursive scheduling. You can see examples here:
|
I believe this stack overflows: Observable.from(1).repeat().toBlockingObservable().forEach(new Action1<Integer>() {
@Override
public void call(Integer t1) {
}});Stacktrace: If We need to use the recursive scheduler idiom to implement repeat. |
|
Small detail: in some places, you should replace "The number of times to repeat the element" by "The number of times to repeat the source sequence". |
|
Is there a reason the repeat() uses an external Scheduler and the ReplaySubject? How about a simpler approach: public static <T> OnSubscribeFunc<T> repeat(final Observable<T> source, final int count) {
return new OnSubscribeFunc<T>() {
@Override
public Subscription onSubscribe(final Observer<? super T> t1) {
final SerialSubscription sreg = new SerialSubscription();
final Observable<T> ssource = source.subscribeOn(Schedulers.currentThread());
Observer<T> o = new Observer<T>() {
int remaining = count;
@Override
public void onNext(T args) {
t1.onNext(args);
}
@Override
public void onError(Throwable e) {
try {
t1.onError(e);
} finally {
sreg.unsubscribe();
}
}
@Override
public void onCompleted() {
if (remaining-- > 0) {
sreg.setSubscription(ssource.subscribe(this));
} else {
t1.onCompleted();
sreg.unsubscribe();
}
}
};
sreg.setSubscription(ssource.subscribe(o));
return sreg;
}
};
} |
|
@benjchristensen I think using Schedulers.immediate() causes the stack overflow problem. Here is a test in C#: static void Main(string[] args)
{
IObservable<int> obs = Observable.Create<int>(o => new Foo(o).test(Scheduler.Immediate, 0));
obs.Subscribe(
x => Console.WriteLine("OnNext: " + x)
);
Console.ReadLine();
}
class Foo
{
IObserver<int> _o;
public Foo(IObserver<int> o)
{
_o = o;
}
public IDisposable test(IScheduler s, int i)
{
return s.Schedule(i + 1, (scheduler, x) => {
_o.OnNext(x);
return this.test(scheduler, x);
});
}
}This code does not cause a stack overflow exception. But in RxJava, the following unit test will cause a stack overflow exception: @Test
public void testRecursiveScheduler1() {
Observable<Integer> obs = Observable
.create(new OnSubscribeFunc<Integer>() {
@Override
public Subscription onSubscribe(
final Observer<? super Integer> observer) {
return Schedulers.immediate().schedule(0,
new Func2<Scheduler, Integer, Subscription>() {
@Override
public Subscription call(
Scheduler scheduler, Integer i) {
observer.onNext(i);
return scheduler.schedule(i + 1, this);
}
});
}
});
obs.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onNext(Integer args) {
System.out.println(args);
}
});
}I'm trying to understand the ImmediateScheduler in Rx.Net. Seems that it also uses a queue to save the Actions to avoid the stack overflow exception. |
|
RxJava-pull-requests #444 FAILURE |
|
RxJava-pull-requests #445 FAILURE |
|
RxJava-pull-requests #446 SUCCESS |
|
The current implementation still has some problems which can not be simply handled in If using ImmediateScheduler in If using CurrentThreadScheduler, the following test will not stop. I suppose it should stop when the observer has some error. @Test
public void testRepeatWithInfiniteRepeatCountWithCurrentThread() {
Observable<String> observable = repeat("foo", Schedulers.currentThread());
@SuppressWarnings("unchecked")
Observer<String> observer = (Observer<String>) mock(Observer.class);
doAnswer(new Answer<Void>() {
private int count = 0;
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
count++;
if (count == 100) {
// Only verify if repeating 100 times
// We can not really verify if repeating infinitely.
throw new RuntimeException("some error");
}
return null;
}
}).when(observer).onNext(anyString());
observable.subscribe(observer);
InOrder inOrder = inOrder(observer);
inOrder.verify(observer, times(100)).onNext("foo");
inOrder.verify(observer).onError(isA(RuntimeException.class));
inOrder.verifyNoMoreInteractions();
} |
| && calledTimes.incrementAndGet() == repeatCount) { | ||
| observer.onCompleted(); | ||
| } else { | ||
| subscription.setSubscription(scheduler.schedule(cancel, self)); |
There was a problem hiding this comment.
The problem about CurrentThreadScheduler is here. When we reach the onCompleted, an synchronized Observable has not return the subscription yet. In Observable.java, every subscribe will be wrapped by a SafeObservableSubscription and SafeObserver. If the observer has some error, SafeObserver will unsubscribe the SafeObservableSubscription. But now SafeObservableSubscription has not be set the real Subscription yet. So now when we is in the onCompleted here, we has no way to know if unsubscribe has happens. We will continue to call scheduler.schedule.
Next time, when we enter onCompleted here again, we still do not know if unsubscribe has happens. So we will not be able to terminate it even if the observer has some error.
SafeObservableSubscription subscription = new SafeObservableSubscription();
subscription.wrap(onSubscribeFunction.onSubscribe(new SafeObserver<T>(subscription, observer)));
return hook.onSubscribeReturn(this, subscription);
|
Agree. But now as the current thread has some issue, I use ImmediateScheduler to pass the unit test.
Could you give me an example? How can we know the
I suppose my problem it's about SafeObservableSubscription and SafeObserver. But maybe we are talking about the same issue since SafeObservableSubscription and SafeObserver are used in toBlockingObservable, too. |
|
This hangs in Rx.NET as well (2.1.30214.0) Observable.Return(1).Repeat().Take(100).Subscribe(Console.WriteLine);The issue is in the Observable.Return(1, Scheduler.Default).Repeat().Take(100).Subscribe(Console.WriteLine);as now the Repeat.onCompleted can run in another thread and the Repeat.subscribe() can return. I believe this can be achieved by not subscribing to the raw source in the Repeat.subscribe() but rather using subscribeOn(Schedulers.newThread()) to move the subscription into a parallel thread and let the Repeat.subscribe() return. public static <T> OnSubscribeFunc<T> repeat(final Observable<T> source, final int count) {
return new OnSubscribeFunc<T>() {
@Override
public Subscription onSubscribe(final Observer<? super T> t1) {
final SerialSubscription sreg = new SerialSubscription();
final Observable<T> ssource = source.subscribeOn(Schedulers.newThread());
Observer<T> o = new Observer<T>() {
int remaining = count;
@Override
public void onNext(T args) {
t1.onNext(args);
}
@Override
public void onError(Throwable e) {
try {
t1.onError(e);
} finally {
sreg.unsubscribe();
}
}
@Override
public void onCompleted() {
if (remaining-- > 0) {
sreg.setSubscription(ssource.subscribe(this));
} else {
t1.onCompleted();
sreg.unsubscribe();
}
}
};
sreg.setSubscription(ssource.subscribe(o));
return sreg;
}
};
} |
Thanks, you remind me one thing: There is not an So is it OK that we do not provide this kind of method and warn that |
|
I think Rx.Net started out its Return operator to run on the threadpool. If manually put back there, the example works. Nowadays it runs on the immediate scheduler, causing the problem. Even if you warn the user about the scheduler, there is no way to know if an incoming observable is dangerous or not. This affects other operators such as concat and onerrorresume. |
|
Return always used the immediate scheduler :-) |
|
Same stack overflow issue happens in @Test
public void testIntervalWithImmediateScheduler() {
Observable.interval(1, TimeUnit.MILLISECONDS, Schedulers.immediate())
.subscribe(new Observer<Long>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onNext(Long args) {
System.out.println(args);
}
});
} |
|
CurrentThreadScheduler can work with |
|
I am working with @headinthebox on changes to Schedulers including |
|
Completed in #699 |
Hi, this PR implemented the
Repeatoperator #70. Please take a look. Thanks!