iSharkFly-Docs
/
java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Reformat Rxjava examples

This commit is contained in:
pivovarit 2017-02-05 08:30:32 +01:00
parent 10eb6bfcc0
commit 6d0dcd52fa
6 changed files with 36 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
rxjava/src/main/java/com/baelding/rxjava/ColdObservableBackpressure.java
View File

@ -5,7 +5,10 @@ import rx.schedulers.Schedulers;
public class ColdObservableBackpressure { public class ColdObservableBackpressure {
public static void main(String[] args) throws InterruptedException { public static void main(String[] args) throws InterruptedException {
Observable.range(1, 1_000_000).observeOn(Schedulers.computation()).subscribe(v -> ComputeFunction.compute(v), Throwable::printStackTrace); Observable
.range(1, 1_000_000)
.observeOn(Schedulers.computation())
.subscribe(ComputeFunction::compute, Throwable::printStackTrace);
Thread.sleep(10_000); Thread.sleep(10_000);

7
rxjava/src/main/java/com/baelding/rxjava/HotObservableBackpressureBatching.java
View File

@ -5,9 +5,12 @@ import rx.subjects.PublishSubject;
public class HotObservableBackpressureBatching { public class HotObservableBackpressureBatching {
public static void main(String[] args) throws InterruptedException { public static void main(String[] args) throws InterruptedException {
PublishSubject<Integer> source = PublishSubject.<Integer>create(); PublishSubject<Integer> source = PublishSubject.<Integer> create();
source.window(500).observeOn(Schedulers.computation()).subscribe(ComputeFunction::compute, Throwable::printStackTrace); source
.window(500)
.observeOn(Schedulers.computation())
.subscribe(ComputeFunction::compute, Throwable::printStackTrace);
for (int i = 0; i < 1_000_000; i++) { for (int i = 0; i < 1_000_000; i++) {
source.onNext(i); source.onNext(i);

7
rxjava/src/main/java/com/baelding/rxjava/HotObservableBackpressureBuffering.java
View File

@ -5,9 +5,12 @@ import rx.subjects.PublishSubject;
public class HotObservableBackpressureBuffering { public class HotObservableBackpressureBuffering {
public static void main(String[] args) throws InterruptedException { public static void main(String[] args) throws InterruptedException {
PublishSubject<Integer> source = PublishSubject.<Integer>create(); PublishSubject<Integer> source = PublishSubject.<Integer> create();
source.buffer(1024).observeOn(Schedulers.computation()).subscribe(ComputeFunction::compute, Throwable::printStackTrace); source
.buffer(1024)
.observeOn(Schedulers.computation())
.subscribe(ComputeFunction::compute, Throwable::printStackTrace);
for (int i = 0; i < 1_000_000; i++) { for (int i = 0; i < 1_000_000; i++) {
source.onNext(i); source.onNext(i);

7
rxjava/src/main/java/com/baelding/rxjava/HotObservableBackpressureSkipping.java
View File

@ -7,11 +7,12 @@ import java.util.concurrent.TimeUnit;
public class HotObservableBackpressureSkipping { public class HotObservableBackpressureSkipping {
public static void main(String[] args) throws InterruptedException { public static void main(String[] args) throws InterruptedException {
PublishSubject<Integer> source = PublishSubject.<Integer>create(); PublishSubject<Integer> source = PublishSubject.<Integer> create();
source.sample(100, TimeUnit.MILLISECONDS) source.sample(100, TimeUnit.MILLISECONDS)
// .throttleFirst(100, TimeUnit.MILLISECONDS) //.throttleFirst(100, TimeUnit.MILLISECONDS)
.observeOn(Schedulers.computation()).subscribe(ComputeFunction::compute, Throwable::printStackTrace); .observeOn(Schedulers.computation())
.subscribe(ComputeFunction::compute, Throwable::printStackTrace);
for (int i = 0; i < 1_000_000; i++) { for (int i = 0; i < 1_000_000; i++) {
source.onNext(i); source.onNext(i);

19
rxjava/src/main/java/com/baelding/rxjava/HotObservableOnBackpressure.java
View File

@ -6,12 +6,21 @@ import rx.schedulers.Schedulers;
public class HotObservableOnBackpressure { public class HotObservableOnBackpressure {
public static void main(String[] args) throws InterruptedException { public static void main(String[] args) throws InterruptedException {
Observable.range(1, 1_000_000).onBackpressureBuffer(16, () -> { Observable
}, BackpressureOverflow.ON_OVERFLOW_DROP_OLDEST).observeOn(Schedulers.computation()).subscribe(e -> { .range(1, 1_000_000)
}, Throwable::printStackTrace); .onBackpressureBuffer(16, () -> {
}, BackpressureOverflow.ON_OVERFLOW_DROP_OLDEST)
.observeOn(Schedulers.computation())
.subscribe(e -> {
}, Throwable::printStackTrace);
Observable.range(1, 1_000_000).onBackpressureDrop().observeOn(Schedulers.io()).doOnNext(ComputeFunction::compute).subscribe(v -> { Observable
}, Throwable::printStackTrace); .range(1, 1_000_000)
.onBackpressureDrop()
.observeOn(Schedulers.io())
.doOnNext(ComputeFunction::compute)
.subscribe(v -> {
}, Throwable::printStackTrace);
Thread.sleep(10_000); Thread.sleep(10_000);
} }

9
rxjava/src/main/java/com/baelding/rxjava/HotObservableWithoutBackpressure.java
View File

@ -1,16 +1,15 @@
package com.baelding.rxjava; package com.baelding.rxjava;
import rx.schedulers.Schedulers; import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject; import rx.subjects.PublishSubject;
public class HotObservableWithoutBackpressure { public class HotObservableWithoutBackpressure {
public static void main(String[] args) throws InterruptedException { public static void main(String[] args) throws InterruptedException {
PublishSubject<Integer> source = PublishSubject.<Integer>create(); PublishSubject<Integer> source = PublishSubject.<Integer> create();
source.observeOn(Schedulers.computation())
.subscribe(ComputeFunction::compute, Throwable::printStackTrace);
source
.observeOn(Schedulers.computation())
.subscribe(ComputeFunction::compute, Throwable::printStackTrace);
for (int i = 0; i < 1_000_000; i++) { for (int i = 0; i < 1_000_000; i++) {
source.onNext(i); source.onNext(i);