Skip to content

Implemented Sample Operation #248

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.

Sign up for GitHub

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

Merged
merged 9 commits into from
May 1, 2013
Merged

Implemented Sample Operation #248

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
8ed552b
First attempt at "sample" operator. Should probably use "interval"
Apr 8, 2013
0bfe5a0
Merge branch 'interval' into sample
Apr 11, 2013
74ba4eb
Switched sample from internal clock to using the interval operator
Apr 11, 2013
cd7a56e
Added a unit test against the sample operator
Apr 11, 2013
cdcfb08
Merge branch 'master' into sample
Apr 22, 2013
5924a76
added missing linebreak
Apr 22, 2013
36e460a
minor renaming and removing unnecessary warning suppression again
Apr 27, 2013
143c730
Merge branch 'master' into sample
May 1, 2013
ae2183c
Added sample methods to observable.
May 1, 2013
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
59 changes: 59 additions & 0 deletions rxjava-core/src/main/java/rx/Observable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@
import rx.operators.OperationOnErrorResumeNextViaFunction;
import rx.operators.OperationOnErrorResumeNextViaObservable;
import rx.operators.OperationOnErrorReturn;
import rx.operators.OperationSample;
import rx.operators.OperationScan;
import rx.operators.OperationSkip;
import rx.operators.OperationSubscribeOn;
Expand Down Expand Up @@ -252,13 +253,15 @@ public Subscription subscribe(final Map<String, Object> callbacks) {
*/
return protectivelyWrapAndSubscribe(new Observer() {

@Override
public void onCompleted() {
Object onComplete = callbacks.get("onCompleted");
if (onComplete != null) {
Functions.from(onComplete).call();
}
}

@Override
public void onError(Exception e) {
handleError(e);
Object onError = callbacks.get("onError");
Expand All @@ -267,6 +270,7 @@ public void onError(Exception e) {
}
}

@Override
public void onNext(Object args) {
onNext.call(args);
}
Expand Down Expand Up @@ -298,15 +302,18 @@ public Subscription subscribe(final Object o) {
*/
return protectivelyWrapAndSubscribe(new Observer() {

@Override
public void onCompleted() {
// do nothing
}

@Override
public void onError(Exception e) {
handleError(e);
// no callback defined
}

@Override
public void onNext(Object args) {
onNext.call(args);
}
Expand All @@ -327,15 +334,18 @@ public Subscription subscribe(final Action1<T> onNext) {
*/
return protectivelyWrapAndSubscribe(new Observer<T>() {

@Override
public void onCompleted() {
// do nothing
}

@Override
public void onError(Exception e) {
handleError(e);
// no callback defined
}

@Override
public void onNext(T args) {
if (onNext == null) {
throw new RuntimeException("onNext must be implemented");
Expand Down Expand Up @@ -365,17 +375,20 @@ public Subscription subscribe(final Object onNext, final Object onError) {
*/
return protectivelyWrapAndSubscribe(new Observer() {

@Override
public void onCompleted() {
// do nothing
}

@Override
public void onError(Exception e) {
handleError(e);
if (onError != null) {
Functions.from(onError).call(e);
}
}

@Override
public void onNext(Object args) {
onNextFunction.call(args);
}
Expand All @@ -396,17 +409,20 @@ public Subscription subscribe(final Action1<T> onNext, final Action1<Exception>
*/
return protectivelyWrapAndSubscribe(new Observer<T>() {

@Override
public void onCompleted() {
// do nothing
}

@Override
public void onError(Exception e) {
handleError(e);
if (onError != null) {
onError.call(e);
}
}

@Override
public void onNext(T args) {
if (onNext == null) {
throw new RuntimeException("onNext must be implemented");
Expand Down Expand Up @@ -436,19 +452,22 @@ public Subscription subscribe(final Object onNext, final Object onError, final O
*/
return protectivelyWrapAndSubscribe(new Observer() {

@Override
public void onCompleted() {
if (onComplete != null) {
Functions.from(onComplete).call();
}
}

@Override
public void onError(Exception e) {
handleError(e);
if (onError != null) {
Functions.from(onError).call(e);
}
}

@Override
public void onNext(Object args) {
onNextFunction.call(args);
}
Expand All @@ -469,17 +488,20 @@ public Subscription subscribe(final Action1<T> onNext, final Action1<Exception>
*/
return protectivelyWrapAndSubscribe(new Observer<T>() {

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

@Override
public void onError(Exception e) {
handleError(e);
if (onError != null) {
onError.call(e);
}
}

@Override
public void onNext(T args) {
if (onNext == null) {
throw new RuntimeException("onNext must be implemented");
Expand Down Expand Up @@ -516,10 +538,12 @@ public void forEach(final Action1<T> onNext) {
* See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
*/
protectivelyWrapAndSubscribe(new Observer<T>() {
@Override
public void onCompleted() {
latch.countDown();
}

@Override
public void onError(Exception e) {
/*
* If we receive an onError event we set the reference on the outer thread
Expand All @@ -531,6 +555,7 @@ public void onError(Exception e) {
latch.countDown();
}

@Override
public void onNext(T args) {
onNext.call(args);
}
Expand Down Expand Up @@ -582,6 +607,7 @@ public void forEach(final Object o) {

forEach(new Action1() {

@Override
public void call(Object args) {
onNext.call(args);
}
Expand Down Expand Up @@ -2743,6 +2769,7 @@ public Observable<T> filter(final Object callback) {
final FuncN _f = Functions.from(callback);
return filter(this, new Func1<T, Boolean>() {

@Override
public Boolean call(T t1) {
return (Boolean) _f.call(t1);
}
Expand Down Expand Up @@ -2913,6 +2940,7 @@ public <R> Observable<R> map(final Object callback) {
final FuncN _f = Functions.from(callback);
return map(this, new Func1<T, R>() {

@Override
@SuppressWarnings("unchecked")
public R call(T t1) {
return (R) _f.call(t1);
Expand Down Expand Up @@ -2963,6 +2991,7 @@ public <R> Observable<R> mapMany(final Object callback) {
final FuncN _f = Functions.from(callback);
return mapMany(this, new Func1<T, Observable<R>>() {

@Override
@SuppressWarnings("unchecked")
public Observable<R> call(T t1) {
return (Observable<R>) _f.call(t1);
Expand Down Expand Up @@ -3071,6 +3100,7 @@ public Observable<T> onErrorResumeNext(final Object resumeFunction) {
final FuncN _f = Functions.from(resumeFunction);
return onErrorResumeNext(this, new Func1<Exception, Observable<T>>() {

@Override
@SuppressWarnings("unchecked")
public Observable<T> call(Exception e) {
return (Observable<T>) _f.call(e);
Expand Down Expand Up @@ -3152,6 +3182,7 @@ public Observable<T> onErrorReturn(final Object resumeFunction) {
final FuncN _f = Functions.from(resumeFunction);
return onErrorReturn(this, new Func1<Exception, T>() {

@Override
@SuppressWarnings("unchecked")
public T call(Exception e) {
return (T) _f.call(e);
Expand Down Expand Up @@ -3288,6 +3319,34 @@ public Observable<T> scan(Func2<T, T, T> accumulator) {
return scan(this, accumulator);
}

/**
* Samples the observable sequence at each interval.
*
* @param period
* The period of time that defines the sampling rate.
* @param unit
* The time unit for the sampling rate time period.
* @return An observable sequence whose elements are the results of sampling the current observable sequence.
*/
public Observable<T> sample(long period, TimeUnit unit) {
return create(OperationSample.sample(this, period, unit));
}

/**
* Samples the observable sequence at each interval.
*
* @param period
* The period of time that defines the sampling rate.
* @param unit
* The time unit for the sampling rate time period.
* @param scheduler
* The scheduler to use for sampling.
* @return An observable sequence whose elements are the results of sampling the current observable sequence.
*/
public Observable<T> sample(long period, TimeUnit unit, Scheduler scheduler) {
return create(OperationSample.sample(this, period, unit, scheduler));
}

/**
* Returns an Observable that applies a function of your choosing to the first item emitted by a
* source Observable, then feeds the result of that function along with the second item emitted
Expand Down
Loading