JAVA-11499 move rxjava* modules to new rxjava-modules (#12342)

* JAVA-11499 move rxjava* modules to new rxjava-modules

* JAVA-11499 addressed PR comments
This commit is contained in:
Keerthi
2022-06-19 01:51:25 +10:00
committed by GitHub
parent 467d2d5b3a
commit eb99a50559
64 changed files with 79 additions and 111 deletions

62
rxjava-modules/pom.xml Normal file
View File

@@ -0,0 +1,62 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.baeldung.rxjava-modules</groupId>
<artifactId>rxjava-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>rxjava-modules</name>
<packaging>pom</packaging>
<parent>
<groupId>com.baeldung</groupId>
<artifactId>parent-java</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../parent-java</relativePath>
</parent>
<modules>
<module>rxjava-core</module>
<module>rxjava-libraries</module>
<module>rxjava-observables</module>
<module>rxjava-operators</module>
</modules>
<dependencies>
<dependency>
<groupId>io.reactivex</groupId>
<artifactId>rxjava</artifactId>
<version>${rx.java.version}</version>
</dependency>
<dependency>
<groupId>io.reactivex.rxjava2</groupId>
<artifactId>rxjava</artifactId>
<version>${rx.java2.version}</version>
</dependency>
<dependency>
<groupId>io.reactivex</groupId>
<artifactId>rxjava-string</artifactId>
<version>${rx.java.string.version}</version>
</dependency>
<dependency>
<groupId>io.reactivex</groupId>
<artifactId>rxjava-math</artifactId>
<version>${rx.java.math.version}</version>
</dependency>
<dependency>
<groupId>com.jayway.awaitility</groupId>
<artifactId>awaitility</artifactId>
<version>${awaitility.version}</version>
</dependency>
</dependencies>
<properties>
<rx.java.version>1.2.5</rx.java.version>
<awaitility.version>1.7.0</awaitility.version>
<rx.java.string.version>1.1.1</rx.java.string.version>
<rx.java.math.version>1.0.0</rx.java.math.version>
<rx.java2.version>2.2.2</rx.java2.version>
</properties>
</project>

View File

@@ -0,0 +1,16 @@
## RxJava
This module contains articles about RxJava.
### Relevant articles:
- [Dealing with Backpressure with RxJava](https://www.baeldung.com/rxjava-backpressure)
- [How to Test RxJava?](https://www.baeldung.com/rxjava-testing)
- [Introduction to RxJava](https://www.baeldung.com/rx-java)
- [Schedulers in RxJava](https://www.baeldung.com/rxjava-schedulers)
- [Difference Between Flatmap and Switchmap in RxJava](https://www.baeldung.com/rxjava-flatmap-switchmap)
- [RxJava and Error Handling](https://www.baeldung.com/rxjava-error-handling)
- [RxJava Maybe](https://www.baeldung.com/rxjava-maybe)
- [Combining RxJava Completables](https://www.baeldung.com/rxjava-completable)
- [RxJava Hooks](https://www.baeldung.com/rxjava-hooks)
- More articles: [[next -->]](/rxjava-2)

View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>rxjava-core</artifactId>
<version>1.0-SNAPSHOT</version>
<name>rxjava-core</name>
<parent>
<groupId>com.baeldung.rxjava-modules</groupId>
<artifactId>rxjava-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
</project>

View File

@@ -0,0 +1,43 @@
package com.baeldung.rxjava;
import rx.Observable;
import java.util.List;
public class ComputeFunction {
public static void compute(Integer v) {
try {
System.out.println("compute integer v: " + v);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void compute(List<Integer> v) {
try {
System.out.println("compute integer v: " + v);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void compute(Observable<Integer> v) {
try {
v.forEach(System.out::println);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void compute(Long v) {
try {
System.out.println("compute integer v: " + v);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

View File

@@ -0,0 +1,22 @@
package com.baeldung.rxjava;
import rx.Observable;
import rx.observables.ConnectableObservable;
import java.util.concurrent.TimeUnit;
public class ConnectableObservableImpl {
public static void main(String[] args) throws InterruptedException {
ConnectableObservable<Long> connectable
= Observable.interval(200, TimeUnit.MILLISECONDS).publish();
connectable.subscribe(System.out::println);
System.out.println("Connect");
connectable.connect();
Thread.sleep(500);
System.out.println("Sleep");
}
}

View File

@@ -0,0 +1,81 @@
package com.baeldung.rxjava;
import rx.Observable;
import rx.observables.BlockingObservable;
import java.util.Arrays;
import java.util.List;
public class ObservableImpl {
private static Integer[] numbers = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
private static String[] letters = {"a", "b", "c", "d", "e", "f", "g", "h", "i"};
private static String[] titles = {"title"};
private static List<String> titleList = Arrays.asList(titles);
static Observable<String> getTitle() {
return Observable.from(titleList);
}
public static void main(String[] args) {
System.out.println("-------Just-----------");
Observable<String> observable = Observable.just("Hello");
observable.subscribe(
System.out::println, //onNext
Throwable::printStackTrace, //onError
() -> System.out.println("onCompleted") //onCompleted
);
BlockingObservable<String> blockingObservable = observable.toBlocking();
System.out.println();
System.out.println("-------Map-----------");
Observable.from(letters)
.map(String::toUpperCase)
.subscribe(System.out::print);
System.out.println();
System.out.println("-------FlatMap-----------");
Observable.just("book1", "book2")
.flatMap(s -> getTitle())
.subscribe(System.out::print);
System.out.println();
System.out.println("--------Scan----------");
Observable.from(letters)
.scan(new StringBuilder(), StringBuilder::append)
.subscribe(System.out::println);
System.out.println();
System.out.println("------GroubBy------------");
Observable.from(numbers)
.groupBy(i -> 0 == (i % 2) ? "EVEN" : "ODD")
.subscribe((group) -> group.subscribe((number) -> {
System.out.println(group.getKey() + " : " + number);
}));
System.out.println();
System.out.println("-------Filter-----------");
Observable.from(numbers)
.filter(i -> (i % 2 == 1))
.subscribe(System.out::println);
System.out.println("------DefaultIfEmpty------------");
Observable.empty()
.defaultIfEmpty("Observable is empty")
.subscribe(System.out::println);
System.out.println("------DefaultIfEmpty-2-----------");
Observable.from(letters)
.defaultIfEmpty("Observable is empty")
.first()
.subscribe(System.out::println);
System.out.println("-------TakeWhile-----------");
Observable.from(numbers)
.takeWhile(i -> i < 5)
.subscribe(System.out::println);
}
}

View File

@@ -0,0 +1,30 @@
package com.baeldung.rxjava;
import rx.Observable;
public class ResourceManagement {
public static void main(String[] args) {
Observable<Character> values = Observable.using(
() -> {
String resource = "MyResource";
System.out.println("Leased: " + resource);
return resource;
},
r -> Observable.create(o -> {
for (Character c : r.toCharArray()) {
o.onNext(c);
}
o.onCompleted();
}),
r -> System.out.println("Disposed: " + r)
);
values.subscribe(
System.out::println,
System.out::println
);
}
}

View File

@@ -0,0 +1,18 @@
package com.baeldung.rxjava;
import rx.Observable;
import rx.Single;
public class SingleImpl {
public static void main(String[] args) {
Single<String> single = Observable.just("Hello")
.toSingle()
.doOnSuccess(System.out::print)
.doOnError(e -> {
throw new RuntimeException(e.getMessage());
});
single.subscribe();
}
}

View File

@@ -0,0 +1,73 @@
package com.baeldung.rxjava;
import rx.Observer;
import rx.subjects.PublishSubject;
public class SubjectImpl {
static Integer subscriber1 = 0;
static Integer subscriber2 = 0;
private static Integer subjectMethod() {
PublishSubject<Integer> subject = PublishSubject.create();
subject.subscribe(getFirstObserver());
subject.onNext(1);
subject.onNext(2);
subject.onNext(3);
subject.subscribe(getSecondObserver());
subject.onNext(4);
subject.onCompleted();
return subscriber1 + subscriber2;
}
static Observer<Integer> getFirstObserver() {
return new Observer<Integer>() {
@Override
public void onNext(Integer value) {
subscriber1 += value;
System.out.println("Subscriber1: " + value);
}
@Override
public void onError(Throwable e) {
System.out.println("error");
}
@Override
public void onCompleted() {
System.out.println("Subscriber1 completed");
}
};
}
static Observer<Integer> getSecondObserver() {
return new Observer<Integer>() {
@Override
public void onNext(Integer value) {
subscriber2 += value;
System.out.println("Subscriber2: " + value);
}
@Override
public void onError(Throwable e) {
System.out.println("error");
}
@Override
public void onCompleted() {
System.out.println("Subscriber2 completed");
}
};
}
public static void main(String[] args) throws InterruptedException {
System.out.println(subjectMethod());
}
}

View File

@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n
</pattern>
</encoder>
</appender>
<root level="INFO">
<appender-ref ref="STDOUT" />
</root>
</configuration>

View File

@@ -0,0 +1,112 @@
package com.baeldung.rxjava;
import io.reactivex.Completable;
import io.reactivex.Flowable;
import io.reactivex.Single;
import io.reactivex.observers.DisposableCompletableObserver;
import org.junit.Before;
import org.junit.Test;
public class CompletableUnitTest {
Completable first;
Completable second;
Completable error;
Throwable throwable = new RuntimeException();
@Before
public void setUpCompletables() {
first = Completable.fromSingle(Single.just(1));
second = Completable.fromRunnable(() -> {});
error = Single.error(throwable)
.ignoreElement();
}
@Test
public void whenCompletableConstructed_thenCompletedSuccessfully() {
Completable completed = Completable.complete();
completed.subscribe(new DisposableCompletableObserver() {
@Override
public void onComplete() {
System.out.println("Completed!");
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
});
Flowable<String> flowable = Flowable.just("request received", "user logged in");
Completable flowableCompletable = Completable.fromPublisher(flowable);
Completable singleCompletable = Single.just(1)
.ignoreElement();
completed.andThen(flowableCompletable)
.andThen(singleCompletable)
.test()
.assertComplete();
}
@Test
public void whenCombiningCompletables_thenCompletedSuccessfully() {
first.andThen(second)
.test()
.assertComplete();
}
@Test
public void whenCombinedWithError_thenCompletedWithError() {
first.andThen(second)
.andThen(error)
.test()
.assertError(throwable);
}
@Test
public void whenCombinedWithNever_thenDoesNotComplete() {
first.andThen(second)
.andThen(Completable.never())
.test()
.assertNotComplete();
}
@Test
public void whenMergedCompletables_thenCompletedSuccessfully() {
Completable.mergeArray(first, second)
.test()
.assertComplete();
}
@Test
public void whenMergedWithError_thenCompletedWithError() {
Completable.mergeArray(first, second, error)
.test()
.assertError(throwable);
}
@Test
public void whenFlatMaped_thenCompletedSuccessfully() {
Completable allElementsCompletable = Flowable.just("request received", "user logged in")
.flatMapCompletable(message -> Completable
.fromRunnable(() -> System.out.println(message))
);
allElementsCompletable
.test()
.assertComplete();
}
@Test
public void whenAmbWithNever_thenCompletedSuccessfully() {
Completable.ambArray(first, Completable.never(), second)
.test()
.assertComplete();
}
@Test
public void whenAmbWithError_thenCompletedWithError() {
Completable.ambArray(error, first, second)
.test()
.assertError(throwable);
}
}

View File

@@ -0,0 +1,27 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.Observable;
import rx.observables.ConnectableObservable;
import java.util.concurrent.TimeUnit;
import static com.jayway.awaitility.Awaitility.await;
import static junit.framework.Assert.assertFalse;
import static junit.framework.Assert.assertTrue;
public class ConnectableObservableIntegrationTest {
@Test
public void givenConnectableObservable_whenConnect_thenGetMessage() throws InterruptedException {
String[] result = {""};
ConnectableObservable<Long> connectable
= Observable.interval(500, TimeUnit.MILLISECONDS).publish();
connectable.subscribe(i -> result[0] += i);
assertFalse(result[0].equals("01"));
connectable.connect();
await()
.until(() -> assertTrue(result[0].equals("01")));
}
}

View File

@@ -0,0 +1,40 @@
package com.baeldung.rxjava;
import org.junit.Test;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
public class MaybeUnitTest {
@Test
public void whenEmitsSingleValue_thenItIsObserved() {
Maybe<Integer> maybe = Flowable.just(1, 2, 3, 4, 5)
.firstElement();
maybe.map(x -> x + 7)
.filter(x -> x > 0)
.test()
.assertResult(8)
.assertComplete();
}
@Test
public void whenEmitsNoValue_thenSignalsCompletionAndNoValueObserved() {
Maybe<Integer> maybe = Flowable.just(1, 2, 3, 4, 5)
.skip(5)
.firstElement();
maybe.test()
.assertComplete()
.assertNoValues();
}
@Test
public void whenThrowsError_thenErrorIsRaised() {
Maybe<Integer> maybe = Flowable.<Integer> error(new Exception("msg"))
.firstElement();
maybe.test()
.assertErrorMessage("msg");
}
}

View File

@@ -0,0 +1,144 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.Observable;
import static com.baeldung.rxjava.ObservableImpl.getTitle;
import static junit.framework.Assert.assertTrue;
public class ObservableUnitTest {
private String result = "";
@Test
public void givenString_whenJustAndSubscribe_thenEmitsSingleItem() {
Observable<String> observable = Observable.just("Hello");
observable.subscribe(s -> result = s);
assertTrue(result.equals("Hello"));
}
@Test
public void givenArray_whenFromAndSubscribe_thenEmitsItems() {
String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
Observable<String> observable = Observable.from(letters);
observable.subscribe(
i -> result += i,
Throwable::printStackTrace,
() -> result += "_Complete"
);
assertTrue(result.equals("abcdefg_Complete"));
}
@Test
public void givenArray_whenConvertsObservabletoBlockingObservable_thenReturnFirstElement() {
String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
Observable<String> observable = Observable.from(letters);
String blockingObservable = observable.toBlocking().first();
observable.subscribe(
i -> result += i,
Throwable::printStackTrace,
() -> result += "_Completed"
);
assertTrue(String.valueOf(result.charAt(0)).equals(blockingObservable));
}
@Test
public void givenArray_whenMapAndSubscribe_thenReturnCapitalLetters() {
String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
Observable.from(letters)
.map(String::toUpperCase)
.subscribe(letter -> result += letter);
assertTrue(result.equals("ABCDEFG"));
}
@Test
public void givenArray_whenFlatMapAndSubscribe_thenReturnUpperAndLowerCaseLetters() {
Observable.just("book1", "book2")
.flatMap(s -> getTitle())
.subscribe(l -> result += l);
assertTrue(result.equals("titletitle"));
}
@Test
public void givenArray_whenScanAndSubscribe_thenReturnTheSumOfAllLetters() {
String[] letters = {"a", "b", "c"};
Observable.from(letters)
.scan(new StringBuilder(), StringBuilder::append)
.subscribe(total -> result += total.toString());
assertTrue(result.equals("aababc"));
}
@Test
public void givenArrayOfNumbers_whenGroupBy_thenCreateTwoGroupsBasedOnParity() {
Integer[] numbers = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
String[] EVEN = {""};
String[] ODD = {""};
Observable.from(numbers)
.groupBy(i -> 0 == (i % 2) ? "EVEN" : "ODD")
.subscribe(group ->
group.subscribe((number) -> {
if (group.getKey().equals("EVEN")) {
EVEN[0] += number;
} else {
ODD[0] += number;
}
})
);
assertTrue(EVEN[0].equals("0246810"));
assertTrue(ODD[0].equals("13579"));
}
@Test
public void givenArrayOfNumbers_whenFilter_thenGetAllOddNumbers() {
Integer[] numbers = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
Observable.from(numbers)
.filter(i -> (i % 2 == 1))
.subscribe(i -> result += i);
assertTrue(result.equals("13579"));
}
@Test
public void givenEmptyObservable_whenDefaultIfEmpty_thenGetDefaultMessage() {
Observable.empty()
.defaultIfEmpty("Observable is empty")
.subscribe(s -> result += s);
assertTrue(result.equals("Observable is empty"));
}
@Test
public void givenObservableFromArray_whenDefaultIfEmptyAndFirst_thenGetFirstLetterFromArray() {
String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
Observable.from(letters)
.defaultIfEmpty("Observable is empty")
.first()
.subscribe(s -> result += s);
assertTrue(result.equals("a"));
}
@Test
public void givenObservableFromArray_whenTakeWhile_thenGetSumOfNumbersFromCondition() {
Integer[] numbers = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
final Integer[] sum = {0};
Observable.from(numbers)
.takeWhile(i -> i < 5)
.subscribe(s -> sum[0] += s);
assertTrue(sum[0] == 10);
}
}

View File

@@ -0,0 +1,30 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.Observable;
import static junit.framework.Assert.assertTrue;
public class ResourceManagementUnitTest {
@Test
public void givenResource_whenUsingOberservable_thenCreatePrintDisposeResource() throws InterruptedException {
String[] result = {""};
Observable<Character> values = Observable.using(
() -> "MyResource",
r -> Observable.create(o -> {
for (Character c : r.toCharArray())
o.onNext(c);
o.onCompleted();
}),
r -> System.out.println("Disposed: " + r)
);
values.subscribe(
v -> result[0] += v,
e -> result[0] += e
);
assertTrue(result[0].equals("MyResource"));
}
}

View File

@@ -0,0 +1,125 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.BackpressureOverflow;
import rx.Observable;
import rx.exceptions.MissingBackpressureException;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import static org.junit.Assert.assertTrue;
public class RxJavaBackpressureLongRunningUnitTest {
@Test
public void givenColdObservable_shouldNotThrowException() {
// given
TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
// when
Observable.range(1, 1_000_000).observeOn(Schedulers.computation()).subscribe(testSubscriber);
// then
testSubscriber.awaitTerminalEvent();
assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
}
@Test
public void givenHotObservable_whenBackpressureNotDefined_shouldTrowException() {
// given
TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
PublishSubject<Integer> source = PublishSubject.create();
source.observeOn(Schedulers.computation()).subscribe(testSubscriber);
// when
IntStream.range(0, 1_000_000).forEach(source::onNext);
// then
testSubscriber.awaitTerminalEvent();
testSubscriber.assertError(MissingBackpressureException.class);
}
@Test
public void givenHotObservable_whenWindowIsDefined_shouldNotThrowException() {
// given
TestSubscriber<Observable<Integer>> testSubscriber = new TestSubscriber<>();
PublishSubject<Integer> source = PublishSubject.create();
// when
source.window(500).observeOn(Schedulers.computation()).subscribe(testSubscriber);
IntStream.range(0, 1_000).forEach(source::onNext);
// then
testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
}
@Test
public void givenHotObservable_whenBufferIsDefined_shouldNotThrowException() {
// given
TestSubscriber<List<Integer>> testSubscriber = new TestSubscriber<>();
PublishSubject<Integer> source = PublishSubject.create();
// when
source.buffer(1024).observeOn(Schedulers.computation()).subscribe(testSubscriber);
IntStream.range(0, 1_000).forEach(source::onNext);
// then
testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
}
@Test
public void givenHotObservable_whenSkippingOperationIsDefined_shouldNotThrowException() {
// given
TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
PublishSubject<Integer> source = PublishSubject.create();
// when
source.sample(100, TimeUnit.MILLISECONDS)
// .throttleFirst(100, TimeUnit.MILLISECONDS)
.observeOn(Schedulers.computation()).subscribe(testSubscriber);
IntStream.range(0, 1_000).forEach(source::onNext);
// then
testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
}
@Test
public void givenHotObservable_whenOnBackpressureBufferDefined_shouldNotThrowException() {
// given
TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
// when
Observable.range(1, 1_000_000).onBackpressureBuffer(16, () -> {
}, BackpressureOverflow.ON_OVERFLOW_DROP_OLDEST).observeOn(Schedulers.computation()).subscribe(testSubscriber);
// then
testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
}
@Test
public void givenHotObservable_whenOnBackpressureDropDefined_shouldNotThrowException() {
// given
TestSubscriber<Integer> testSubscriber = new TestSubscriber<>();
// when
Observable.range(1, 1_000_000).onBackpressureDrop().observeOn(Schedulers.computation())
.subscribe(testSubscriber);
// then
testSubscriber.awaitTerminalEvent(2, TimeUnit.SECONDS);
assertTrue(testSubscriber.getOnErrorEvents().size() == 0);
}
}

View File

@@ -0,0 +1,84 @@
package com.baeldung.rxjava;
import static org.junit.Assert.assertTrue;
import org.junit.After;
import org.junit.Test;
import io.reactivex.Observable;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;
public class RxJavaHooksManualTest {
private boolean initHookCalled = false;
private boolean hookCalled = false;
@Test
public void givenIOScheduler_whenCalled_shouldExecuteTheHooks() {
RxJavaPlugins.setInitIoSchedulerHandler((scheduler) -> {
initHookCalled = true;
return scheduler.call();
});
RxJavaPlugins.setIoSchedulerHandler((scheduler) -> {
hookCalled = true;
return scheduler;
});
Observable.range(1, 10)
.map(v -> v * 2)
.subscribeOn(Schedulers.io())
.test();
assertTrue(hookCalled && initHookCalled);
}
@Test
public void givenNewThreadScheduler_whenCalled_shouldExecuteTheHook() {
RxJavaPlugins.setInitNewThreadSchedulerHandler((scheduler) -> {
initHookCalled = true;
return scheduler.call();
});
RxJavaPlugins.setNewThreadSchedulerHandler((scheduler) -> {
hookCalled = true;
return scheduler;
});
Observable.range(1, 15)
.map(v -> v * 2)
.subscribeOn(Schedulers.newThread())
.test();
assertTrue(hookCalled && initHookCalled);
}
@Test
public void givenSingleScheduler_whenCalled_shouldExecuteTheHooks() {
RxJavaPlugins.setInitSingleSchedulerHandler((scheduler) -> {
initHookCalled = true;
return scheduler.call();
});
RxJavaPlugins.setSingleSchedulerHandler((scheduler) -> {
hookCalled = true;
return scheduler;
});
Observable.range(1, 10)
.map(v -> v * 2)
.subscribeOn(Schedulers.single())
.test();
assertTrue(hookCalled && initHookCalled);
}
@After
public void reset() {
hookCalled = false;
initHookCalled = false;
RxJavaPlugins.reset();
}
}

View File

@@ -0,0 +1,244 @@
package com.baeldung.rxjava;
import static org.junit.Assert.assertTrue;
import org.junit.After;
import org.junit.Test;
import io.reactivex.Completable;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.reactivex.flowables.ConnectableFlowable;
import io.reactivex.observables.ConnectableObservable;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;
public class RxJavaHooksUnitTest {
private boolean initHookCalled = false;
private boolean hookCalled = false;
@Test
public void givenObservable_whenError_shouldExecuteTheHook() {
RxJavaPlugins.setErrorHandler(throwable -> {
hookCalled = true;
});
Observable.error(new IllegalStateException())
.subscribe();
assertTrue(hookCalled);
}
@Test
public void givenCompletable_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnCompletableAssembly(completable -> {
hookCalled = true;
return completable;
});
Completable.fromSingle(Single.just(1));
assertTrue(hookCalled);
}
@Test
public void givenCompletable_whenSubscribed_shouldExecuteTheHook() {
RxJavaPlugins.setOnCompletableSubscribe((completable, observer) -> {
hookCalled = true;
return observer;
});
Completable.fromSingle(Single.just(1))
.test();
assertTrue(hookCalled);
}
@Test
public void givenObservable_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnObservableAssembly(observable -> {
hookCalled = true;
return observable;
});
Observable.range(1, 10);
assertTrue(hookCalled);
}
@Test
public void givenObservable_whenSubscribed_shouldExecuteTheHook() {
RxJavaPlugins.setOnObservableSubscribe((observable, observer) -> {
hookCalled = true;
return observer;
});
Observable.range(1, 10)
.test();
assertTrue(hookCalled);
}
@Test
public void givenConnectableObservable_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnConnectableObservableAssembly(connectableObservable -> {
hookCalled = true;
return connectableObservable;
});
ConnectableObservable.range(1, 10)
.publish()
.connect();
assertTrue(hookCalled);
}
@Test
public void givenFlowable_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnFlowableAssembly(flowable -> {
hookCalled = true;
return flowable;
});
Flowable.range(1, 10);
assertTrue(hookCalled);
}
@Test
public void givenFlowable_whenSubscribed_shouldExecuteTheHook() {
RxJavaPlugins.setOnFlowableSubscribe((flowable, observer) -> {
hookCalled = true;
return observer;
});
Flowable.range(1, 10)
.test();
assertTrue(hookCalled);
}
@Test
public void givenConnectableFlowable_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnConnectableFlowableAssembly(connectableFlowable -> {
hookCalled = true;
return connectableFlowable;
});
ConnectableFlowable.range(1, 10)
.publish()
.connect();
assertTrue(hookCalled);
}
@Test
public void givenParallel_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnParallelAssembly(parallelFlowable -> {
hookCalled = true;
return parallelFlowable;
});
Flowable.range(1, 10)
.parallel();
assertTrue(hookCalled);
}
@Test
public void givenMaybe_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnMaybeAssembly(maybe -> {
hookCalled = true;
return maybe;
});
Maybe.just(1);
assertTrue(hookCalled);
}
@Test
public void givenMaybe_whenSubscribed_shouldExecuteTheHook() {
RxJavaPlugins.setOnMaybeSubscribe((maybe, observer) -> {
hookCalled = true;
return observer;
});
Maybe.just(1)
.test();
assertTrue(hookCalled);
}
@Test
public void givenSingle_whenAssembled_shouldExecuteTheHook() {
RxJavaPlugins.setOnSingleAssembly(single -> {
hookCalled = true;
return single;
});
Single.just(1);
assertTrue(hookCalled);
}
@Test
public void givenSingle_whenSubscribed_shouldExecuteTheHook() {
RxJavaPlugins.setOnSingleSubscribe((single, observer) -> {
hookCalled = true;
return observer;
});
Single.just(1)
.test();
assertTrue(hookCalled);
}
@Test
public void givenAnyScheduler_whenCalled_shouldExecuteTheHook() {
RxJavaPlugins.setScheduleHandler((runnable) -> {
hookCalled = true;
return runnable;
});
Observable.range(1, 10)
.map(v -> v * 2)
.subscribeOn(Schedulers.single())
.test();
hookCalled = false;
Observable.range(1, 10)
.map(v -> v * 2)
.subscribeOn(Schedulers.computation())
.test();
assertTrue(hookCalled);
}
@Test
public void givenComputationScheduler_whenCalled_shouldExecuteTheHooks() {
RxJavaPlugins.setInitComputationSchedulerHandler((scheduler) -> {
initHookCalled = true;
return scheduler.call();
});
RxJavaPlugins.setComputationSchedulerHandler((scheduler) -> {
hookCalled = true;
return scheduler;
});
Observable.range(1, 10)
.map(v -> v * 2)
.subscribeOn(Schedulers.computation())
.test();
assertTrue(hookCalled && initHookCalled);
}
@After
public void reset() {
initHookCalled = false;
hookCalled = false;
RxJavaPlugins.reset();
}
}

View File

@@ -0,0 +1,104 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
import rx.schedulers.TestScheduler;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.hamcrest.Matchers.hasItems;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertThat;
public class RxJavaUnitTest {
@Test
public void givenObservable_whenZip_shouldAssertBlockingInASameThread() {
// given
List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
List<String> results = new ArrayList<>();
Observable<String> observable = Observable.from(letters)
.zipWith(Observable.range(1, Integer.MAX_VALUE), (string, index) -> index + "-" + string);
// when
observable.subscribe(results::add);
// then
assertThat(results, notNullValue());
assertThat(results, hasSize(5));
assertThat(results, hasItems("1-A", "2-B", "3-C", "4-D", "5-E"));
}
@Test
public void givenObservable_whenZip_shouldAssertOnTestSubscriber() {
// given
List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
TestSubscriber<String> subscriber = new TestSubscriber<>();
Observable<String> observable = Observable.from(letters)
.zipWith(Observable.range(1, Integer.MAX_VALUE), ((string, index) -> index + "-" + string));
// when
observable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(5);
assertThat(subscriber.getOnNextEvents(), hasItems("1-A", "2-B", "3-C", "4-D", "5-E"));
}
@Test
public void givenTestObserver_whenExceptionWasThrowsOnObservable_observerShouldGetError() {
// given
List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
TestSubscriber<String> subscriber = new TestSubscriber<>();
Observable<String> observable = Observable.from(letters)
.zipWith(Observable.range(1, Integer.MAX_VALUE), ((string, index) -> index + "-" + string))
.concatWith(Observable.error(new RuntimeException("error in Observable")));
// when
observable.subscribe(subscriber);
// then
subscriber.assertError(RuntimeException.class);
subscriber.assertNotCompleted();
}
@Test
public void givenObservableThatEmitsEventPerSecond_whenUseAdvanceByTime_shouldEmitEventPerSecond() {
// given
List<String> letters = Arrays.asList("A", "B", "C", "D", "E");
TestScheduler scheduler = new TestScheduler();
TestSubscriber<String> subscriber = new TestSubscriber<>();
Observable<Long> tick = Observable.interval(1, TimeUnit.SECONDS, scheduler);
Observable<String> observable = Observable.from(letters).zipWith(tick, (string, index) -> index + "-" + string);
observable.subscribeOn(scheduler).subscribe(subscriber);
// expect
subscriber.assertNoValues();
subscriber.assertNotCompleted();
// when
scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
// then
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValues("0-A");
// when
scheduler.advanceTimeTo(6, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(5);
assertThat(subscriber.getOnNextEvents(), hasItems("0-A", "1-B", "2-C", "3-D", "4-E"));
}
}

View File

@@ -0,0 +1,247 @@
package com.baeldung.rxjava;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.junit.Ignore;
import org.junit.Test;
import rx.Observable;
import rx.Scheduler;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;
import rx.schedulers.TestScheduler;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import static com.jayway.awaitility.Awaitility.await;
import static java.util.concurrent.Executors.newFixedThreadPool;
import static org.hamcrest.Matchers.hasItems;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
public class SchedulersLiveTest {
private String result = "";
private String result1 = "";
private String result2 = "";
@Test
public void givenScheduledWorker_whenScheduleAnAction_thenResultAction() throws InterruptedException {
System.out.println("scheduling");
Scheduler scheduler = Schedulers.immediate();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> result += "action");
assertTrue(result.equals("action"));
}
@Test
public void givenScheduledWorker_whenUnsubscribeOnWorker_thenResultFirstAction() throws InterruptedException {
System.out.println("canceling");
Scheduler scheduler = Schedulers.newThread();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
result += "First_Action";
worker.unsubscribe();
});
worker.schedule(() -> result += "Second_Action");
await()
.until(() -> assertTrue(result.equals("First_Action")));
}
@Ignore //it's not safe, not every time is running correctly
@Test
public void givenWorker_whenScheduledOnNewThread_thenResultIsBoundToNewThread() throws InterruptedException {
System.out.println("newThread_1");
Scheduler scheduler = Schedulers.newThread();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
result += Thread.currentThread().getName() + "_Start";
worker.schedule(() -> result += "_worker_");
result += "_End";
});
await()
.until(() -> assertTrue(result.equals("RxNewThreadScheduler-1_Start_End_worker_")));
}
@Test
public void givenObservable_whenObserveOnNewThread_thenRunOnDifferentThreadEachTime() throws InterruptedException {
System.out.println("newThread_2");
Observable.just("Hello")
.observeOn(Schedulers.newThread())
.doOnNext(s ->
result2 += Thread.currentThread().getName()
)
.observeOn(Schedulers.newThread())
.subscribe(s ->
result1 += Thread.currentThread().getName()
);
await()
.until(() -> {
assertTrue(result1.equals("RxNewThreadScheduler-1"));
assertTrue(result2.equals("RxNewThreadScheduler-2"));
});
}
@Test
public void givenWorker_whenScheduledOnImmediate_thenResultIsBoundToThread() throws InterruptedException {
System.out.println("immediate_1");
Scheduler scheduler = Schedulers.immediate();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
result += Thread.currentThread().getName() + "_Start";
worker.schedule(() -> result += "_worker_");
result += "_End";
});
await()
.until(() -> assertTrue(result.equals("main_Start_worker__End")));
}
@Test
public void givenObservable_whenImmediateScheduled_thenExecuteOnMainThread() throws InterruptedException {
System.out.println("immediate_2");
Observable.just("Hello")
.subscribeOn(Schedulers.immediate())
.subscribe(s ->
result += Thread.currentThread().getName()
);
await()
.until(() -> assertTrue(result.equals("main")));
}
@Test
public void givenObservable_whenTrampolineScheduled_thenExecuteOnMainThread() throws InterruptedException {
System.out.println("trampoline_1");
Observable.just(2, 4, 6, 8)
.subscribeOn(Schedulers.trampoline())
.subscribe(i -> result += "" + i);
Observable.just(1, 3, 5, 7, 9)
.subscribeOn(Schedulers.trampoline())
.subscribe(i -> result += "" + i);
await()
.until(() -> assertTrue(result.equals("246813579")));
}
@Test
public void givenWorker_whenScheduledOnTrampoline_thenComposeResultAsBlocking() throws InterruptedException {
System.out.println("trampoline_2");
Scheduler scheduler = Schedulers.trampoline();
Scheduler.Worker worker = scheduler.createWorker();
worker.schedule(() -> {
result += Thread.currentThread().getName() + "Start";
worker.schedule(() -> {
result += "_middleStart";
worker.schedule(() ->
result += "_worker_"
);
result += "_middleEnd";
});
result += "_mainEnd";
});
await()
.until(() -> assertTrue(result.equals("mainStart_mainEnd_middleStart_middleEnd_worker_")));
}
private ThreadFactory threadFactory(String pattern) {
return new ThreadFactoryBuilder()
.setNameFormat(pattern)
.build();
}
@Test
public void givenExecutors_whenSchedulerFromCreatedExecutors_thenReturnElementsOnEacheThread() throws InterruptedException {
System.out.println("from");
ExecutorService poolA = newFixedThreadPool(10, threadFactory("Sched-A-%d"));
Scheduler schedulerA = Schedulers.from(poolA);
ExecutorService poolB = newFixedThreadPool(10, threadFactory("Sched-B-%d"));
Scheduler schedulerB = Schedulers.from(poolB);
Observable<String> observable = Observable.create(subscriber -> {
subscriber.onNext("Alfa");
subscriber.onNext("Beta");
subscriber.onCompleted();
});
observable
.subscribeOn(schedulerA)
.subscribeOn(schedulerB)
.subscribe(
x -> result += Thread.currentThread().getName() + x + "_",
Throwable::printStackTrace,
() -> result += "_Completed"
);
await()
.until(() -> assertTrue(result.equals("Sched-A-0Alfa_Sched-A-0Beta__Completed")));
}
@Test
public void givenObservable_whenIoScheduling_thenReturnThreadName() throws InterruptedException {
System.out.println("io");
Observable.just("io")
.subscribeOn(Schedulers.io())
.subscribe(i -> result += Thread.currentThread().getName());
await()
.until(() -> assertTrue(result.equals("RxIoScheduler-2")));
}
@Test
@Ignore
public void givenObservable_whenComputationScheduling_thenReturnThreadName() throws InterruptedException {
System.out.println("computation");
Observable.just("computation")
.subscribeOn(Schedulers.computation())
.subscribe(i -> result += Thread.currentThread().getName());
await()
.until(() -> assertTrue(result.equals("RxComputationScheduler-1")));
}
@Test
public void givenLetters_whenTestScheduling_thenReturnValuesControllingAdvanceTime() throws InterruptedException {
List<String> letters = Arrays.asList("A", "B", "C");
TestScheduler scheduler = Schedulers.test();
TestSubscriber<String> subscriber = new TestSubscriber<>();
Observable<Long> tick = Observable.interval(1, TimeUnit.SECONDS, scheduler);
Observable.from(letters)
.zipWith(tick, (string, index) -> index + "-" + string)
.subscribeOn(scheduler)
.subscribe(subscriber);
subscriber.assertNoValues();
subscriber.assertNotCompleted();
scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValues("0-A");
scheduler.advanceTimeTo(3, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(3);
assertThat(subscriber.getOnNextEvents(), hasItems("0-A", "1-B", "2-C"));
}
@Test
public void givenLetters_whenDelay_thenReturne() throws InterruptedException {
ExecutorService poolA = newFixedThreadPool(10, threadFactory("Sched1-"));
Scheduler schedulerA = Schedulers.from(poolA);
Observable.just('A', 'B')
.delay(1, TimeUnit.SECONDS, schedulerA)
.subscribe(i -> result += Thread.currentThread().getName() + i + " ");
await()
.until(() -> assertTrue(result.equals("Sched1-A Sched1-B ")));
}
}

View File

@@ -0,0 +1,23 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.Observable;
import rx.Single;
import static junit.framework.Assert.assertTrue;
public class SingleUnitTest {
@Test
public void givenSingleObservable_whenSuccess_thenGetMessage() throws InterruptedException {
String[] result = {""};
Single<String> single = Observable.just("Hello")
.toSingle()
.doOnSuccess(i -> result[0] += i)
.doOnError(error -> {
throw new RuntimeException(error.getMessage());
});
single.subscribe();
assertTrue(result[0].equals("Hello"));
}
}

View File

@@ -0,0 +1,25 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.subjects.PublishSubject;
import static junit.framework.Assert.assertTrue;
public class SubjectUnitTest {
@Test
public void givenSubjectAndTwoSubscribers_whenSubscribeOnSubject_thenSubscriberBeginsToAdd() {
PublishSubject<Integer> subject = PublishSubject.create();
subject.subscribe(SubjectImpl.getFirstObserver());
subject.onNext(1);
subject.onNext(2);
subject.onNext(3);
subject.subscribe(SubjectImpl.getSecondObserver());
subject.onNext(4);
subject.onCompleted();
assertTrue(SubjectImpl.subscriber1 + SubjectImpl.subscriber2 == 14);
}
}

View File

@@ -0,0 +1,138 @@
package com.baeldung.rxjava.onerror;
import io.reactivex.Observable;
import io.reactivex.exceptions.CompositeException;
import io.reactivex.observers.TestObserver;
import org.junit.Test;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.Assert.assertTrue;
public class ExceptionHandlingIntegrationTest {
private Error UNKNOWN_ERROR = new Error("unknown error");
private Exception UNKNOWN_EXCEPTION = new Exception("unknown exception");
@Test
public void givenSubscriberAndError_whenHandleOnErrorReturn_thenResumed() {
TestObserver<String> testObserver = new TestObserver<>();
Observable
.<String>error(UNKNOWN_ERROR)
.onErrorReturn(Throwable::getMessage)
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(1);
testObserver.assertValue("unknown error");
}
@Test
public void givenSubscriberAndError_whenHandleOnErrorResume_thenResumed() {
TestObserver<String> testObserver = new TestObserver<>();
Observable
.<String>error(UNKNOWN_ERROR)
.onErrorResumeNext(Observable.just("one", "two"))
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(2);
testObserver.assertValues("one", "two");
}
@Test
public void givenSubscriberAndError_whenHandleOnErrorResumeItem_thenResumed() {
TestObserver<String> testObserver = new TestObserver<>();
Observable
.<String>error(UNKNOWN_ERROR)
.onErrorReturnItem("singleValue")
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(1);
testObserver.assertValue("singleValue");
}
@Test
public void givenSubscriberAndError_whenHandleOnErrorResumeFunc_thenResumed() {
TestObserver<String> testObserver = new TestObserver<>();
Observable
.<String>error(UNKNOWN_ERROR)
.onErrorResumeNext(throwable -> {
return Observable.just(throwable.getMessage(), "nextValue");
})
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(2);
testObserver.assertValues("unknown error", "nextValue");
}
@Test
public void givenSubscriberAndError_whenChangeStateOnError_thenErrorThrown() {
TestObserver<String> testObserver = new TestObserver<>();
final AtomicBoolean state = new AtomicBoolean(false);
Observable
.<String>error(UNKNOWN_ERROR)
.doOnError(throwable -> state.set(true))
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("state should be changed", state.get());
}
@Test
public void givenSubscriberAndError_whenExceptionOccurOnError_thenCompositeExceptionThrown() {
TestObserver<String> testObserver = new TestObserver<>();
Observable
.<String>error(UNKNOWN_ERROR)
.doOnError(throwable -> {
throw new RuntimeException("unexcepted");
})
.subscribe(testObserver);
testObserver.assertError(CompositeException.class);
testObserver.assertNotComplete();
testObserver.assertNoValues();
}
@Test
public void givenSubscriberAndException_whenHandleOnException_thenResumed() {
TestObserver<String> testObserver = new TestObserver<>();
Observable
.<String>error(UNKNOWN_EXCEPTION)
.onExceptionResumeNext(Observable.just("exceptionResumed"))
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertValueCount(1);
testObserver.assertValue("exceptionResumed");
}
@Test
public void givenSubscriberAndError_whenHandleOnException_thenNotResumed() {
TestObserver<String> testObserver = new TestObserver<>();
Observable
.<String>error(UNKNOWN_ERROR)
.onExceptionResumeNext(Observable.just("exceptionResumed"))
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
}
}

View File

@@ -0,0 +1,146 @@
package com.baeldung.rxjava.onerror;
import io.reactivex.Observable;
import io.reactivex.observers.TestObserver;
import org.junit.Test;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.assertTrue;
public class OnErrorRetryIntegrationTest {
private Error UNKNOWN_ERROR = new Error("unknown error");
@Test
public void givenSubscriberAndError_whenRetryOnError_thenRetryConfirmed() {
TestObserver<String> testObserver = new TestObserver<>();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.<String>error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retry(1)
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("should call twice", atomicCounter.get() == 2);
}
@Test
public void givenSubscriberAndError_whenRetryConditionallyOnError_thenRetryConfirmed() {
TestObserver<String> testObserver = new TestObserver<>();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.<String>error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retry((integer, throwable) -> integer < 4)
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("should call 4 times", atomicCounter.get() == 4);
}
@Test
public void givenSubscriberAndError_whenRetryUntilOnError_thenRetryConfirmed() {
TestObserver<String> testObserver = new TestObserver<>();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.<String>error(UNKNOWN_ERROR)
.retryUntil(() -> atomicCounter.incrementAndGet() > 3)
.subscribe(testObserver);
testObserver.assertError(UNKNOWN_ERROR);
testObserver.assertNotComplete();
testObserver.assertNoValues();
assertTrue("should call 4 times", atomicCounter.get() == 4);
}
@Test
public void givenSubscriberAndError_whenRetryWhenOnError_thenRetryConfirmed() {
TestObserver<String> testObserver = new TestObserver<>();
Exception noretryException = new Exception("don't retry");
Observable
.<String>error(UNKNOWN_ERROR)
.retryWhen(throwableObservable -> Observable.<String>error(noretryException))
.subscribe(testObserver);
testObserver.assertError(noretryException);
testObserver.assertNotComplete();
testObserver.assertNoValues();
}
@Test
public void givenSubscriberAndError_whenRetryWhenOnError_thenCompleted() {
TestObserver<String> testObserver = new TestObserver<>();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.<String>error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retryWhen(throwableObservable -> Observable.empty())
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertNoValues();
assertTrue("should not retry", atomicCounter.get() == 0);
}
@Test
public void givenSubscriberAndError_whenRetryWhenOnError_thenResubscribed() {
TestObserver<String> testObserver = new TestObserver<>();
AtomicInteger atomicCounter = new AtomicInteger(0);
Observable
.<String>error(() -> {
atomicCounter.incrementAndGet();
return UNKNOWN_ERROR;
})
.retryWhen(throwableObservable -> Observable.just("anything"))
.subscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertNoValues();
assertTrue("should retry once", atomicCounter.get() == 1);
}
@Test
public void givenSubscriberAndError_whenRetryWhenForMultipleTimesOnError_thenResumed() {
TestObserver<String> testObserver = new TestObserver<>();
long before = System.currentTimeMillis();
Observable
.<String>error(UNKNOWN_ERROR)
.retryWhen(throwableObservable -> throwableObservable
.zipWith(Observable.range(1, 3), (throwable, integer) -> integer)
.flatMap(integer -> {
System.out.println("retried " + integer + " times");
return Observable.timer(integer, TimeUnit.SECONDS);
}))
.blockingSubscribe(testObserver);
testObserver.assertNoErrors();
testObserver.assertComplete();
testObserver.assertNoValues();
long secondsElapsed = (System.currentTimeMillis() - before) / 1000;
assertTrue("6 seconds should elapse", secondsElapsed == 6);
}
}

View File

@@ -0,0 +1,10 @@
## RxJava Libraries
This module contains articles about RxJava libraries
### Related Articles:
- [RxJava 2 Flowable](https://www.baeldung.com/rxjava-2-flowable)
- [Introduction to RxRelay for RxJava](https://www.baeldung.com/rx-relay)
- [Introduction to rxjava-jdbc](https://www.baeldung.com/rxjava-jdbc)

View File

@@ -0,0 +1,40 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>rxjava-libraries</artifactId>
<version>1.0-SNAPSHOT</version>
<name>rxjava-libraries</name>
<parent>
<groupId>com.baeldung.rxjava-modules</groupId>
<artifactId>rxjava-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
<groupId>com.jakewharton.rxrelay2</groupId>
<artifactId>rxrelay</artifactId>
<version>${rxrelay.version}</version>
</dependency>
<dependency>
<groupId>com.github.davidmoten</groupId>
<artifactId>rxjava-jdbc</artifactId>
<version>${rx.java.jdbc.version}</version>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<version>${h2.version}</version>
<scope>runtime</scope>
</dependency>
</dependencies>
<properties>
<rx.java.jdbc.version>0.7.11</rx.java.jdbc.version>
<rxrelay.version>2.0.0</rxrelay.version>
</properties>
</project>

View File

@@ -0,0 +1,32 @@
package com.baeldung.rxjava;
import com.jakewharton.rxrelay2.Relay;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposables;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
public class RandomRelay extends Relay<Integer> {
Random random = new Random();
List<Observer<? super Integer>> observers = new ArrayList<>();
@Override
public void accept(Integer integer) {
int observerIndex = random.nextInt(observers.size()) & Integer.MAX_VALUE;
observers.get(observerIndex).onNext(integer);
}
@Override
public boolean hasObservers() {
return observers.isEmpty();
}
@Override
protected void subscribeActual(Observer<? super Integer> observer) {
observers.add(observer);
observer.onSubscribe(Disposables.fromRunnable(() -> System.out.println("Disposed")));
}
}

View File

@@ -0,0 +1,13 @@
package com.baeldung.rxjava.jdbc;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.ConnectionProviderFromUrl;
class Connector {
private static final String DB_CONNECTION = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1";
private static final String DB_USER = "";
private static final String DB_PASSWORD = "";
static final ConnectionProvider connectionProvider = new ConnectionProviderFromUrl(DB_CONNECTION, DB_USER, DB_PASSWORD);
}

View File

@@ -0,0 +1,13 @@
package com.baeldung.rxjava.jdbc;
import com.github.davidmoten.rx.jdbc.annotations.Column;
public interface Employee {
@Column("id")
int id();
@Column("name")
String name();
}

View File

@@ -0,0 +1,28 @@
package com.baeldung.rxjava.jdbc;
public class Manager {
private int id;
private String name;
public Manager(int id, String name) {
this.id = id;
this.name = name;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}

View File

@@ -0,0 +1,16 @@
package com.baeldung.rxjava.jdbc;
import java.io.IOException;
import java.io.InputStream;
import java.io.StringWriter;
import org.apache.commons.io.IOUtils;
class Utils {
static String getStringFromInputStream(InputStream input) throws IOException {
StringWriter writer = new StringWriter();
IOUtils.copy(input, writer, "UTF-8");
return writer.toString();
}
}

View File

@@ -0,0 +1,93 @@
package com.baeldung.rxjava;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.exceptions.MissingBackpressureException;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subscribers.TestSubscriber;
import org.junit.Test;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
public class FlowableIntegrationTest {
@Test public void whenFlowableIsCreated_thenItIsProperlyInitialized() {
Flowable<Integer> integerFlowable = Flowable.just(1, 2, 3, 4);
assertNotNull(integerFlowable);
}
@Test public void whenFlowableIsCreatedFromObservable_thenItIsProperlyInitialized() throws InterruptedException {
Observable<Integer> integerObservable = Observable.just(1, 2, 3);
Flowable<Integer> integerFlowable = integerObservable.toFlowable(BackpressureStrategy.BUFFER);
assertNotNull(integerFlowable);
}
@Test public void whenFlowableIsCreatedFromFlowableOnSubscribe_thenItIsProperlyInitialized() throws InterruptedException {
FlowableOnSubscribe<Integer> flowableOnSubscribe = flowableEmitter -> flowableEmitter.onNext(1);
Flowable<Integer> integerFlowable = Flowable.create(flowableOnSubscribe, BackpressureStrategy.BUFFER);
assertNotNull(integerFlowable);
}
@Test public void thenAllValuesAreBufferedAndReceived() {
List testList = IntStream.range(0, 100000).boxed().collect(Collectors.toList());
Observable observable = Observable.fromIterable(testList);
TestSubscriber<Integer> testSubscriber = observable.toFlowable(BackpressureStrategy.BUFFER).observeOn(Schedulers.computation()).test();
testSubscriber.awaitTerminalEvent();
List<Integer> receivedInts = testSubscriber.getEvents().get(0).stream().mapToInt(object -> (int) object).boxed().collect(Collectors.toList());
assertEquals(testList, receivedInts);
}
@Test public void whenDropStrategyUsed_thenOnBackpressureDropped() {
List testList = IntStream.range(0, 100000).boxed().collect(Collectors.toList());
Observable observable = Observable.fromIterable(testList);
TestSubscriber<Integer> testSubscriber = observable.toFlowable(BackpressureStrategy.DROP).observeOn(Schedulers.computation()).test();
testSubscriber.awaitTerminalEvent();
List<Integer> receivedInts = testSubscriber.getEvents().get(0).stream().mapToInt(object -> (int) object).boxed().collect(Collectors.toList());
assertThat(receivedInts.size() < testList.size());
assertThat(!receivedInts.contains(100000));
}
@Test public void whenMissingStrategyUsed_thenException() {
Observable observable = Observable.range(1, 100000);
TestSubscriber subscriber = observable.toFlowable(BackpressureStrategy.MISSING).observeOn(Schedulers.computation()).test();
subscriber.awaitTerminalEvent();
subscriber.assertError(MissingBackpressureException.class);
}
@Test public void whenErrorStrategyUsed_thenExceptionIsThrown() {
Observable observable = Observable.range(1, 100000);
TestSubscriber subscriber = observable.toFlowable(BackpressureStrategy.ERROR).observeOn(Schedulers.computation()).test();
subscriber.awaitTerminalEvent();
subscriber.assertError(MissingBackpressureException.class);
}
@Test public void whenLatestStrategyUsed_thenTheLastElementReceived() {
List testList = IntStream.range(0, 100000).boxed().collect(Collectors.toList());
Observable observable = Observable.fromIterable(testList);
TestSubscriber<Integer> testSubscriber = observable.toFlowable(BackpressureStrategy.LATEST).observeOn(Schedulers.computation()).test();
testSubscriber.awaitTerminalEvent();
List<Integer> receivedInts = testSubscriber.getEvents().get(0).stream().mapToInt(object -> (int) object).boxed().collect(Collectors.toList());
assertThat(receivedInts.size() < testList.size());
assertThat(receivedInts.contains(100000));
}
}

View File

@@ -0,0 +1,120 @@
package com.baeldung.rxjava;
import com.jakewharton.rxrelay2.BehaviorRelay;
import com.jakewharton.rxrelay2.PublishRelay;
import com.jakewharton.rxrelay2.ReplayRelay;
import io.reactivex.internal.schedulers.SingleScheduler;
import io.reactivex.observers.TestObserver;
import org.junit.Test;
import java.util.concurrent.TimeUnit;
public class RxRelayIntegrationTest {
@Test
public void whenObserverSubscribedToPublishRelay_thenItReceivesEmittedEvents () {
PublishRelay<Integer> publishRelay = PublishRelay.create();
TestObserver<Integer> firstObserver = TestObserver.create();
TestObserver<Integer> secondObserver = TestObserver.create();
publishRelay.subscribe(firstObserver);
firstObserver.assertSubscribed();
publishRelay.accept(5);
publishRelay.accept(10);
publishRelay.subscribe(secondObserver);
secondObserver.assertSubscribed();
publishRelay.accept(15);
//First Observer will receive all events
firstObserver.assertValues(5, 10, 15);
//Second Observer will receive only last event
secondObserver.assertValue(15);
}
@Test
public void whenObserverSubscribedToBehaviorRelayWithoutDefaultValue_thenItIsEmpty() {
BehaviorRelay<Integer> behaviorRelay = BehaviorRelay.create();
TestObserver<Integer> firstObserver = new TestObserver<>();
behaviorRelay.subscribe(firstObserver);
firstObserver.assertEmpty();
}
@Test
public void whenObserverSubscribedToBehaviorRelay_thenItReceivesDefaultValue() {
BehaviorRelay<Integer> behaviorRelay = BehaviorRelay.createDefault(1);
TestObserver<Integer> firstObserver = new TestObserver<>();
behaviorRelay.subscribe(firstObserver);
firstObserver.assertValue(1);
}
@Test
public void whenObserverSubscribedToBehaviorRelay_thenItReceivesEmittedEvents () {
BehaviorRelay<Integer> behaviorRelay = BehaviorRelay.create();
TestObserver<Integer> firstObserver = TestObserver.create();
TestObserver<Integer> secondObserver = TestObserver.create();
behaviorRelay.accept(5);
behaviorRelay.subscribe(firstObserver);
behaviorRelay.accept(10);
behaviorRelay.subscribe(secondObserver);
behaviorRelay.accept(15);
firstObserver.assertValues(5, 10, 15);
secondObserver.assertValues(10, 15);
}
@Test
public void whenObserverSubscribedToReplayRelay_thenItReceivesEmittedEvents () {
ReplayRelay<Integer> replayRelay = ReplayRelay.create();
TestObserver<Integer> firstObserver = TestObserver.create();
TestObserver<Integer> secondObserver = TestObserver.create();
replayRelay.subscribe(firstObserver);
replayRelay.accept(5);
replayRelay.accept(10);
replayRelay.accept(15);
replayRelay.subscribe(secondObserver);
firstObserver.assertValues(5, 10, 15);
secondObserver.assertValues(5, 10, 15);
}
@Test
public void whenObserverSubscribedToReplayRelayWithLimitedSize_thenItReceivesEmittedEvents () {
ReplayRelay<Integer> replayRelay = ReplayRelay.createWithSize(2);
TestObserver<Integer> firstObserver = TestObserver.create();
replayRelay.accept(5);
replayRelay.accept(10);
replayRelay.accept(15);
replayRelay.accept(20);
replayRelay.subscribe(firstObserver);
firstObserver.assertValues(15, 20);
}
@Test
public void whenObserverSubscribedToReplayRelayWithMaxAge_thenItReceivesEmittedEvents () throws InterruptedException {
ReplayRelay<Integer> replayRelay = ReplayRelay.createWithTime(2000, TimeUnit.MILLISECONDS, new SingleScheduler());
TestObserver<Integer> firstObserver = TestObserver.create();
replayRelay.accept(5);
replayRelay.accept(10);
replayRelay.accept(15);
replayRelay.accept(20);
Thread.sleep(3000);
replayRelay.subscribe(firstObserver);
firstObserver.assertEmpty();
}
@Test
public void whenTwoObserversSubscribedToRandomRelay_thenOnlyOneReceivesEvent () {
RandomRelay randomRelay = new RandomRelay();
TestObserver<Integer> firstObserver = TestObserver.create();
TestObserver<Integer> secondObserver = TestObserver.create();
randomRelay.subscribe(firstObserver);
randomRelay.subscribe(secondObserver);
randomRelay.accept(5);
if(firstObserver.values().isEmpty()) {
secondObserver.assertValue(5);
} else {
firstObserver.assertValue(5);
secondObserver.assertEmpty();
}
}
}

View File

@@ -0,0 +1,63 @@
package com.baeldung.rxjava.jdbc;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.List;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
public class AutomapClassIntegrationTest {
private ConnectionProvider connectionProvider = Connector.connectionProvider;
private Database db = Database.from(connectionProvider);
private Observable<Integer> create = null;
private Observable<Integer> insert1, insert2 = null;
@Before
public void setup() {
create = db.update("CREATE TABLE IF NOT EXISTS MANAGER(id int primary key, name varchar(255))")
.count();
insert1 = db.update("INSERT INTO MANAGER(id, name) VALUES(1, 'Alan')")
.dependsOn(create)
.count();
insert2 = db.update("INSERT INTO MANAGER(id, name) VALUES(2, 'Sarah')")
.dependsOn(create)
.count();
}
@Test
public void whenSelectManagersAndAutomap_thenCorrect() {
List<Manager> managers = db.select("select id, name from MANAGER")
.dependsOn(create)
.dependsOn(insert1)
.dependsOn(insert2)
.autoMap(Manager.class)
.toList()
.toBlocking()
.single();
assertThat(managers.get(0)
.getId()).isEqualTo(1);
assertThat(managers.get(0)
.getName()).isEqualTo("Alan");
assertThat(managers.get(1)
.getId()).isEqualTo(2);
assertThat(managers.get(1)
.getName()).isEqualTo("Sarah");
}
@After
public void close() {
db.update("DROP TABLE MANAGER")
.dependsOn(create);
connectionProvider.close();
}
}

View File

@@ -0,0 +1,64 @@
package com.baeldung.rxjava.jdbc;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.List;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
public class AutomapInterfaceIntegrationTest {
private ConnectionProvider connectionProvider = Connector.connectionProvider;
private Database db = Database.from(connectionProvider);
private Observable<Integer> truncate = null;
private Observable<Integer> insert1, insert2 = null;
@Before
public void setup() {
Observable<Integer> create = db.update("CREATE TABLE IF NOT EXISTS EMPLOYEE(id int primary key, name varchar(255))")
.count();
truncate = db.update("TRUNCATE TABLE EMPLOYEE")
.dependsOn(create)
.count();
insert1 = db.update("INSERT INTO EMPLOYEE(id, name) VALUES(1, 'Alan')")
.dependsOn(truncate)
.count();
insert2 = db.update("INSERT INTO EMPLOYEE(id, name) VALUES(2, 'Sarah')")
.dependsOn(insert1)
.count();
}
@Test
public void whenSelectFromTableAndAutomap_thenCorrect() {
List<Employee> employees = db.select("select id, name from EMPLOYEE")
.dependsOn(insert2)
.autoMap(Employee.class)
.toList()
.toBlocking()
.single();
assertThat(employees.get(0)
.id()).isEqualTo(1);
assertThat(employees.get(0)
.name()).isEqualTo("Alan");
assertThat(employees.get(1)
.id()).isEqualTo(2);
assertThat(employees.get(1)
.name()).isEqualTo("Sarah");
}
@After
public void close() {
db.update("DROP TABLE EMPLOYEE")
.dependsOn(truncate);
connectionProvider.close();
}
}

View File

@@ -0,0 +1,64 @@
package com.baeldung.rxjava.jdbc;
import static org.junit.Assert.assertEquals;
import java.util.Arrays;
import java.util.List;
import org.junit.After;
import org.junit.Test;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
public class BasicQueryTypesIntegrationTest {
private ConnectionProvider connectionProvider = Connector.connectionProvider;
private Database db = Database.from(connectionProvider);
private Observable<Integer> create;
@Test
public void whenCreateTableAndInsertRecords_thenCorrect() {
create = db.update("CREATE TABLE IF NOT EXISTS EMPLOYEE_TABLE(id int primary key, name varchar(255))")
.count();
Observable<Integer> insert1 = db.update("INSERT INTO EMPLOYEE_TABLE(id, name) VALUES(1, 'John')")
.dependsOn(create)
.count();
Observable<Integer> update = db.update("UPDATE EMPLOYEE_TABLE SET name = 'Alan' WHERE id = 1")
.dependsOn(create)
.count();
Observable<Integer> insert2 = db.update("INSERT INTO EMPLOYEE_TABLE(id, name) VALUES(2, 'Sarah')")
.dependsOn(create)
.count();
Observable<Integer> insert3 = db.update("INSERT INTO EMPLOYEE_TABLE(id, name) VALUES(3, 'Mike')")
.dependsOn(create)
.count();
Observable<Integer> delete = db.update("DELETE FROM EMPLOYEE_TABLE WHERE id = 2")
.dependsOn(create)
.count();
List<String> names = db.select("select name from EMPLOYEE_TABLE where id < ?")
.parameter(3)
.dependsOn(create)
.dependsOn(insert1)
.dependsOn(insert2)
.dependsOn(insert3)
.dependsOn(update)
.dependsOn(delete)
.getAs(String.class)
.toList()
.toBlocking()
.single();
assertEquals(Arrays.asList("Alan"), names);
}
@After
public void close() {
db.update("DROP TABLE EMPLOYEE_TABLE")
.dependsOn(create);
connectionProvider.close();
}
}

View File

@@ -0,0 +1,65 @@
package com.baeldung.rxjava.jdbc;
import static org.junit.Assert.assertEquals;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
public class InsertBlobIntegrationTest {
private ConnectionProvider connectionProvider = Connector.connectionProvider;
private Database db = Database.from(connectionProvider);
private String expectedDocument = null;
private String actualDocument = null;
private Observable<Integer> create, insert = null;
@Before
public void setup() throws IOException {
create = db.update("CREATE TABLE IF NOT EXISTS SERVERLOG (id int primary key, document BLOB)")
.count();
InputStream actualInputStream = new FileInputStream("src/test/resources/actual_clob");
this.actualDocument = Utils.getStringFromInputStream(actualInputStream);
byte[] bytes = this.actualDocument.getBytes(StandardCharsets.UTF_8);
InputStream expectedInputStream = new FileInputStream("src/test/resources/expected_clob");
this.expectedDocument = Utils.getStringFromInputStream(expectedInputStream);
this.insert = db.update("insert into SERVERLOG(id,document) values(?,?)")
.parameter(1)
.parameter(Database.toSentinelIfNull(bytes))
.dependsOn(create)
.count();
}
@Test
public void whenInsertBLOB_thenCorrect() throws IOException {
db.select("select document from SERVERLOG where id = 1")
.dependsOn(create)
.dependsOn(insert)
.getAs(String.class)
.toList()
.toBlocking()
.single();
assertEquals(expectedDocument, actualDocument);
}
@After
public void close() {
db.update("DROP TABLE SERVERLOG")
.dependsOn(create);
connectionProvider.close();
}
}

View File

@@ -0,0 +1,63 @@
package com.baeldung.rxjava.jdbc;
import static org.junit.Assert.assertEquals;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
public class InsertClobIntegrationTest {
private ConnectionProvider connectionProvider = Connector.connectionProvider;
private Database db = Database.from(connectionProvider);
private String expectedDocument = null;
private String actualDocument = null;
private Observable<Integer> create, insert = null;
@Before
public void setup() throws IOException {
create = db.update("CREATE TABLE IF NOT EXISTS SERVERLOG_TABLE (id int primary key, document CLOB)")
.count();
InputStream actualInputStream = new FileInputStream("src/test/resources/actual_clob");
this.actualDocument = Utils.getStringFromInputStream(actualInputStream);
InputStream expectedInputStream = new FileInputStream("src/test/resources/expected_clob");
this.expectedDocument = Utils.getStringFromInputStream(expectedInputStream);
this.insert = db.update("insert into SERVERLOG_TABLE(id,document) values(?,?)")
.parameter(1)
.parameter(Database.toSentinelIfNull(actualDocument))
.dependsOn(create)
.count();
}
@Test
public void whenSelectCLOB_thenCorrect() throws IOException {
db.select("select document from SERVERLOG_TABLE where id = 1")
.dependsOn(create)
.dependsOn(insert)
.getAs(String.class)
.toList()
.toBlocking()
.single();
assertEquals(expectedDocument, actualDocument);
}
@After
public void close() {
db.update("DROP TABLE SERVERLOG_TABLE")
.dependsOn(create);
connectionProvider.close();
}
}

View File

@@ -0,0 +1,48 @@
package com.baeldung.rxjava.jdbc;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
public class ReturnKeysIntegrationTest {
private Observable<Boolean> begin = null;
private Observable<Integer> createStatement = null;
private ConnectionProvider connectionProvider = Connector.connectionProvider;
private Database db = Database.from(connectionProvider);
@Before
public void setup() {
begin = db.beginTransaction();
createStatement = db.update("CREATE TABLE IF NOT EXISTS EMPLOYEE_SAMPLE(id int auto_increment primary key, name varchar(255))")
.dependsOn(begin)
.count();
}
@Test
public void whenInsertAndReturnGeneratedKey_thenCorrect() {
Integer key = db.update("INSERT INTO EMPLOYEE_SAMPLE(name) VALUES('John')")
.dependsOn(createStatement)
.returnGeneratedKeys()
.getAs(Integer.class)
.count()
.toBlocking()
.single();
assertThat(key).isEqualTo(1);
}
@After
public void close() {
db.update("DROP TABLE EMPLOYEE_SAMPLE")
.dependsOn(createStatement);
connectionProvider.close();
}
}

View File

@@ -0,0 +1,52 @@
package com.baeldung.rxjava.jdbc;
import static org.junit.Assert.assertEquals;
import org.junit.After;
import org.junit.Test;
import com.github.davidmoten.rx.jdbc.ConnectionProvider;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
public class TransactionIntegrationTest {
private Observable<Integer> createStatement = null;
private ConnectionProvider connectionProvider = Connector.connectionProvider;
private Database db = Database.from(connectionProvider);
@Test
public void whenCommitTransaction_thenRecordUpdated() {
Observable<Boolean> begin = db.beginTransaction();
Observable<Integer> createStatement = db
.update("CREATE TABLE IF NOT EXISTS EMPLOYEE(id int primary key, name varchar(255))")
.dependsOn(begin)
.count();
Observable<Integer> truncateStatement = db.update("TRUNCATE TABLE EMPLOYEE")
.dependsOn(createStatement)
.count();
Observable<Integer> insertStatement = db.update("INSERT INTO EMPLOYEE(id, name) VALUES(1, 'John')")
.dependsOn(truncateStatement)
.count();
Observable<Integer> updateStatement = db.update("UPDATE EMPLOYEE SET name = 'Tom' WHERE id = 1")
.dependsOn(insertStatement)
.count();
Observable<Boolean> commit = db.commit(updateStatement);
String name = db.select("select name from EMPLOYEE WHERE id = 1")
.dependsOn(commit)
.getAs(String.class)
.toBlocking()
.single();
assertEquals("Tom", name);
}
@After
public void close() {
db.update("DROP TABLE EMPLOYEE")
.dependsOn(createStatement);
connectionProvider.close();
}
}

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,11 @@
## RxJava Observables
This module contains articles about RxJava Observables
### Related Articles:
- [Combining Observables in RxJava](https://www.baeldung.com/rxjava-combine-observables)
- [RxJava One Observable, Multiple Subscribers](https://www.baeldung.com/rxjava-multiple-subscribers-observable)
- [RxJava StringObservable](https://www.baeldung.com/rxjava-string)
- [Filtering Observables in RxJava](https://www.baeldung.com/rxjava-filtering)

View File

@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>rxjava-observables</artifactId>
<version>1.0-SNAPSHOT</version>
<name>rxjava-observables</name>
<parent>
<groupId>com.baeldung.rxjava-modules</groupId>
<artifactId>rxjava-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
</project>

View File

@@ -0,0 +1,59 @@
package com.baeldung.rxjava;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.Subscription;
import rx.observables.ConnectableObservable;
import rx.subscriptions.Subscriptions;
public class MultipleSubscribersColdObs {
private static final Logger LOGGER = LoggerFactory.getLogger(MultipleSubscribersColdObs.class);
public static void main(String[] args) throws InterruptedException {
defaultBehaviour();
// subscribeBeforeConnect();
}
private static void defaultBehaviour() {
Observable obs = getObservable();
LOGGER.info("Subscribing");
Subscription s1 = obs.subscribe(i -> LOGGER.info("subscriber#1 is printing " + i));
Subscription s2 = obs.subscribe(i -> LOGGER.info("subscriber#2 is printing " + i));
s1.unsubscribe();
s2.unsubscribe();
}
private static void subscribeBeforeConnect() throws InterruptedException {
ConnectableObservable obs = getObservable().publish();
LOGGER.info("Subscribing");
obs.subscribe(i -> LOGGER.info("subscriber #1 is printing " + i));
obs.subscribe(i -> LOGGER.info("subscriber #2 is printing " + i));
Thread.sleep(1000);
LOGGER.info("Connecting");
Subscription s = obs.connect();
s.unsubscribe();
}
private static Observable getObservable() {
return Observable.create(subscriber -> {
subscriber.onNext(gettingValue(1));
subscriber.onNext(gettingValue(2));
subscriber.add(Subscriptions.create(() -> {
LOGGER.info("Clear resources");
}));
});
}
private static Integer gettingValue(int i) {
LOGGER.info("Getting " + i);
return i;
}
}

View File

@@ -0,0 +1,152 @@
package com.baeldung.rxjava;
import java.awt.Color;
import java.awt.Dimension;
import java.awt.event.MouseAdapter;
import java.awt.event.MouseEvent;
import java.awt.event.MouseListener;
import java.lang.reflect.InvocationTargetException;
import javax.swing.JFrame;
import rx.Observable;
import rx.Subscription;
import rx.observables.ConnectableObservable;
import rx.subscriptions.Subscriptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MultipleSubscribersHotObs {
private static final Logger LOGGER = LoggerFactory.getLogger(MultipleSubscribersHotObs.class);
private static JFrame frame;
public static void main(String[] args) throws InterruptedException, InvocationTargetException {
javax.swing.SwingUtilities.invokeAndWait(new Runnable() {
public void run() {
createAndShowGUI();
}
});
defaultBehaviour();
// subscribeBeforeConnect();
// connectBeforeSubscribe();
// autoConnectAndSubscribe();
// refCountAndSubscribe();
}
private static void createAndShowGUI() {
frame = new JFrame("Hot Observable Demo");
frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
frame.getContentPane().setBackground(Color.GRAY);
frame.setPreferredSize(new Dimension(500, 500));
frame.pack();
frame.setVisible(true);
}
public static void defaultBehaviour() throws InterruptedException {
Observable obs = getObservable();
LOGGER.info("subscribing #1");
Subscription subscription1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription subscription2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("unsubscribe#1");
subscription1.unsubscribe();
Thread.sleep(1000);
LOGGER.info("unsubscribe#2");
subscription2.unsubscribe();
}
public static void subscribeBeforeConnect() throws InterruptedException {
ConnectableObservable obs = getObservable().publish();
LOGGER.info("subscribing #1");
Subscription subscription1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription subscription2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("connecting:");
Subscription s = obs.connect();
Thread.sleep(1000);
LOGGER.info("unsubscribe connected");
s.unsubscribe();
}
public static void connectBeforeSubscribe() throws InterruptedException {
ConnectableObservable obs = getObservable().doOnNext(x -> LOGGER.info("saving " + x)).publish();
LOGGER.info("connecting:");
Subscription s = obs.connect();
Thread.sleep(1000);
LOGGER.info("subscribing #1");
obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
s.unsubscribe();
}
public static void autoConnectAndSubscribe() throws InterruptedException {
Observable obs = getObservable().doOnNext(x -> LOGGER.info("saving " + x)).publish().autoConnect();
LOGGER.info("autoconnect()");
Thread.sleep(1000);
LOGGER.info("subscribing #1");
Subscription s1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription s2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("unsubscribe 1");
s1.unsubscribe();
Thread.sleep(1000);
LOGGER.info("unsubscribe 2");
s2.unsubscribe();
}
public static void refCountAndSubscribe() throws InterruptedException {
Observable obs = getObservable().doOnNext(x -> LOGGER.info("saving " + x)).publish().refCount();
LOGGER.info("refcount()");
Thread.sleep(1000);
LOGGER.info("subscribing #1");
Subscription subscription1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("subscribing #2");
Subscription subscription2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
Thread.sleep(1000);
LOGGER.info("unsubscribe#1");
subscription1.unsubscribe();
Thread.sleep(1000);
LOGGER.info("unsubscribe#2");
subscription2.unsubscribe();
}
private static Observable getObservable() {
return Observable.create(subscriber -> {
frame.addMouseListener(new MouseAdapter() {
@Override
public void mouseClicked(MouseEvent e) {
subscriber.onNext(e.getX());
}
});
subscriber.add(Subscriptions.create(() -> {
LOGGER.info("Clear resources");
for (MouseListener listener : frame.getListeners(MouseListener.class)) {
frame.removeMouseListener(listener);
}
}));
});
}
}

View File

@@ -0,0 +1,67 @@
package com.baeldung.rxjava.combine;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
import java.util.concurrent.TimeUnit;
import static java.util.Arrays.asList;
public class ObservableCombineUnitTest {
@Test
public void givenTwoObservables_whenMerged_shouldEmitCombinedResults() {
TestSubscriber<String> testSubscriber = new TestSubscriber<>();
Observable.merge(
Observable.from(asList("Hello", "World")),
Observable.from(asList("I love", "RxJava"))
).subscribe(testSubscriber);
testSubscriber.assertValues("Hello", "World", "I love", "RxJava");
}
@Test
public void givenTwoObservables_whenZipped_thenReturnCombinedResults() {
TestSubscriber<String> testSubscriber = new TestSubscriber<>();
Observable.zip(
Observable.from(asList("Simple", "Moderate", "Complex")),
Observable.from(asList("Solutions", "Success", "Hierarchy")),
(str1, str2) -> String.format("%s %s", str1, str2))
.subscribe(testSubscriber);
testSubscriber.assertValues("Simple Solutions", "Moderate Success", "Complex Hierarchy");
}
@Test
public void givenMutipleObservablesOneThrows_whenMerged_thenCombineBeforePropagatingError() {
TestSubscriber<String> testSubscriber = new TestSubscriber<>();
Observable.mergeDelayError(
Observable.from(asList("hello", "world")),
Observable.error(new RuntimeException("Some exception")),
Observable.from(asList("rxjava"))
).subscribe(testSubscriber);
testSubscriber.assertValues("hello", "world", "rxjava");
testSubscriber.assertError(RuntimeException.class);
}
@Test
public void givenAStream_whenZippedWithInterval_shouldDelayStreamEmmission() {
TestSubscriber<String> testSubscriber = new TestSubscriber<>();
Observable<String> data = Observable.just("one", "two", "three", "four", "five");
Observable<Long> interval = Observable.interval(1L, TimeUnit.SECONDS);
Observable
.zip(data, interval, (strData, tick) -> String.format("[%d]=%s", tick, strData))
.toBlocking().subscribe(testSubscriber);
testSubscriber.assertCompleted();
testSubscriber.assertValueCount(5);
testSubscriber.assertValues("[0]=one", "[1]=two", "[2]=three", "[3]=four", "[4]=five");
}
}

View File

@@ -0,0 +1,202 @@
package com.baeldung.rxjava.filters;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
public class RxJavaFilterOperatorsIntegrationTest {
@Test
public void givenRangeObservable_whenFilteringItems_thenOddItemsAreFiltered() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.filter(i -> i % 2 != 0);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(5);
subscriber.assertValues(1, 3, 5, 7, 9);
}
@Test
public void givenRangeObservable_whenFilteringWithTake_thenOnlyFirstThreeItemsAreEmitted() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.take(3);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(3);
subscriber.assertValues(1, 2, 3);
}
@Test
public void givenObservable_whenFilteringWithTakeWhile_thenItemsEmittedUntilConditionIsVerified() {
Observable<Integer> sourceObservable = Observable.just(1, 2, 3, 4, 3, 2, 1);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.takeWhile(i -> i < 4);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(3);
subscriber.assertValues(1, 2, 3);
}
@Test
public void givenRangeObservable_whenFilteringWithTakeFirst_thenOnlyFirstItemIsEmitted() {
Observable<Integer> sourceObservable = Observable.just(1, 2, 3, 4, 5, 7, 6);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.takeFirst(x -> x > 5);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(7);
}
@Test
public void givenRangeObservable_whenFilteringWithFirst_thenOnlyFirstThreeItemsAreEmitted() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.first();
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(1);
}
@Test
public void givenEmptyObservable_whenFilteringWithFirstOrDefault_thenDefaultValue() {
Observable<Integer> sourceObservable = Observable.empty();
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.firstOrDefault(-1);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(-1);
}
@Test
public void givenRangeObservable_whenFilteringWithTakeLast_thenLastThreeItemAreEmitted() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.takeLast(3);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(3);
subscriber.assertValues(8, 9, 10);
}
@Test
public void givenRangeObservable_whenFilteringWithLast_thenOnlyLastItemIsEmitted() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.last(i -> i % 2 != 0);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(9);
}
@Test
public void givenRangeObservable_whenFilteringWithLastAndDefault_thenOnlyDefaultIsEmitted() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.lastOrDefault(-1, i -> i > 10);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(-1);
}
@Test
public void givenObservable_whenTakingElementAt_thenItemAtSpecifiedIndexIsEmitted() {
Observable<Integer> sourceObservable = Observable.just(1, 2, 3, 5, 7, 11);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.elementAt(4);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(7);
}
@Test
public void givenObservable_whenTakingElementAtOrDefault_thenDefaultIsReturned() {
Observable<Integer> sourceObservable = Observable.just(1, 2, 3, 5, 7, 11);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.elementAtOrDefault(7, -1);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(-1);
}
@Test
public void givenMixedTypeObservable_whenFilteringByType_thenOnlyNumbersAreEmitted() {
Observable sourceObservable = Observable.just(1, "two", 3, "five", 7, 11);
TestSubscriber subscriber = new TestSubscriber();
Observable filteredObservable = sourceObservable.ofType(String.class);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(2);
subscriber.assertValues("two", "five");
}
}

View File

@@ -0,0 +1,105 @@
package com.baeldung.rxjava.filters;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
public class RxJavaSkipOperatorsIntegrationTest {
@Test
public void givenRangeObservable_whenSkipping_thenFirstFourItemsAreSkipped() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.skip(4);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(6);
subscriber.assertValues(5, 6, 7, 8, 9, 10);
}
@Test
public void givenObservable_whenSkippingWhile_thenFirstItemsAreSkipped() {
Observable<Integer> sourceObservable = Observable.just(1, 2, 3, 4, 5, 4, 3, 2, 1);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.skipWhile(i -> i < 4);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(6);
subscriber.assertValues(4, 5, 4, 3, 2, 1);
}
@Test
public void givenRangeObservable_whenSkippingLast_thenLastFiveItemsAreSkipped() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = sourceObservable.skipLast(5);
filteredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(5);
subscriber.assertValues(1, 2, 3, 4, 5);
}
@Test
public void givenObservable_whenFilteringDistinct_thenOnlyDistinctValuesAreEmitted() {
Observable<Integer> sourceObservable = Observable.just(1, 1, 2, 2, 1, 3, 3, 1);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> distinctObservable = sourceObservable.distinct();
distinctObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(3);
subscriber.assertValues(1, 2, 3);
}
@Test
public void givenObservable_whenFilteringDistinctUntilChanged_thenOnlyDistinctConsecutiveItemsAreEmitted() {
Observable<Integer> sourceObservable = Observable.just(1, 1, 2, 2, 1, 3, 3, 1);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> distinctObservable = sourceObservable.distinctUntilChanged();
distinctObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(5);
subscriber.assertValues(1, 2, 1, 3, 1);
}
@Test
public void givenRangeObservable_whenIgnoringElements_thenOnlyDistinctConsecutiveItemsAreEmitted() {
Observable<Integer> sourceObservable = Observable.range(1, 10);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> ignoredObservable = sourceObservable.ignoreElements();
ignoredObservable.subscribe(subscriber);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(0);
subscriber.assertNoValues();
}
}

View File

@@ -0,0 +1,221 @@
package com.baeldung.rxjava.filters;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
import rx.schedulers.TestScheduler;
public class RxJavaTimeFilteringOperatorsIntegrationTest {
@Test
public void givenTimedObservable_whenSampling_thenOnlySampleItemsAreEmitted() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> sampledObservable =
timedObservable.sample(2500L, TimeUnit.MILLISECONDS, testScheduler);
sampledObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValues(3, 5, 6);
}
@Test
public void givenTimedObservable_whenThrottlingLast_thenThrottleLastItemsAreEmitted() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = timedObservable.throttleLast(3100L, TimeUnit.MILLISECONDS, testScheduler);
filteredObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValues(4, 6);
}
@Test
public void givenRangeObservable_whenThrottlingFirst_thenThrottledFirstItemsAreEmitted() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable =
timedObservable.throttleFirst(4100L, TimeUnit.MILLISECONDS, testScheduler);
filteredObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValues(1, 6);
}
@Test
public void givenTimedObservable_whenThrottlingWithTimeout_thenLastItemIsEmitted() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = timedObservable.throttleWithTimeout(2000L, TimeUnit.MILLISECONDS, testScheduler);
filteredObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValue(6);
}
@Test
public void givenTimedObservable_whenDebounceOperatorIsApplied_thenLastItemIsEmitted() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = timedObservable.debounce(2000L, TimeUnit.MILLISECONDS, testScheduler);
filteredObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValue(6);
}
@Test
public void givenTimedObservable_whenUsingTimeout_thenTimeOutException() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = timedObservable.timeout(500L, TimeUnit.MILLISECONDS, testScheduler);
filteredObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertError(TimeoutException.class);
subscriber.assertValues(1);
}
@Test
public void givenObservable_whenSkippingUntil_thenItemsAreSkippedUntilSecondObservableEmitsItems() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
Observable<Integer> delayedObservable = Observable.just(1)
.delay(3000, TimeUnit.MILLISECONDS, testScheduler);
TestSubscriber<Integer> subscriber = new TestSubscriber();
Observable<Integer> filteredObservable = timedObservable.skipUntil(delayedObservable);
filteredObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValues(4, 5, 6);
}
@Test
public void givenObservable_whenSkippingWhile_thenItemsAreEmittedUntilSecondObservableEmitsItems() {
TestScheduler testScheduler = new TestScheduler();
Observable<Integer> timedObservable =
Observable.just(1, 2, 3, 4, 5, 6)
.zipWith(
Observable.interval(0, 1, TimeUnit.SECONDS, testScheduler),
(item, time) -> item
);
TestSubscriber subscriber = new TestSubscriber();
Observable<Integer> delayedObservable = Observable.just(1)
.delay(3000, TimeUnit.MILLISECONDS, testScheduler);
Observable<Integer> filteredObservable = timedObservable.takeUntil(delayedObservable);
filteredObservable.subscribe(subscriber);
testScheduler.advanceTimeBy(7, TimeUnit.SECONDS);
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValues(1, 2, 3);
}
}

View File

@@ -0,0 +1,145 @@
package com.baeldung.rxjava.operators;
import java.io.ByteArrayInputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import org.junit.Assert;
import org.junit.Test;
import rx.Observable;
import rx.observables.StringObservable;
import rx.observers.TestSubscriber;
public class RxStringOperatorsUnitTest
{
@Test
public void givenStringObservable_whenFromInputStream_ThenSuccessfull()
{
//given
ByteArrayInputStream is = new ByteArrayInputStream("Lorem ipsum loream, Lorem ipsum lore".getBytes(StandardCharsets.UTF_8));
TestSubscriber<String> subscriber = TestSubscriber.create();
// when
StringObservable.decode(StringObservable.from(is), StandardCharsets.UTF_8)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValues("Lorem ipsum loream, Lorem ipsum lore");
}
@Test
public void givenStringObservable_whenEncodingString_ThenSuccessfullObtainingByteStream()
{
//given
Observable<String> sourceObservable = Observable.just("Lorem ipsum loream");
TestSubscriber<byte[]> subscriber = TestSubscriber.create();
// when
StringObservable.encode(sourceObservable, StandardCharsets.UTF_8)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.getOnNextEvents()
.stream()
.forEach(bytes -> Assert.assertTrue(Arrays.equals(bytes, "Lorem ipsum loream".getBytes(StandardCharsets.UTF_8))));
}
@Test
public void givenStringObservable_whenConcatenatingStrings_ThenSuccessfullObtainingSingleString()
{
//given
Observable<String> sourceObservable = Observable.just("Lorem ipsum loream","Lorem ipsum lore");
TestSubscriber<String> subscriber = TestSubscriber.create();
// when
StringObservable.stringConcat(sourceObservable)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValues("Lorem ipsum loreamLorem ipsum lore");
}
@Test
public void givenStringObservable_whenDecodingByteArray_ThenSuccessfullObtainingStringStream()
{
//given
Observable<byte[]> sourceObservable = Observable.just("Lorem ipsum loream".getBytes(StandardCharsets.UTF_8));
TestSubscriber<String> subscriber = TestSubscriber.create();
// when
StringObservable.decode(sourceObservable, StandardCharsets.UTF_8)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValues("Lorem ipsum loream");
}
@Test
public void givenStringObservable_whenStringSplitted_ThenSuccessfullObtainingStringsStream()
{
//given
Observable<String> sourceObservable = Observable.just("Lorem ipsum loream,Lorem ipsum lore");
TestSubscriber<String> subscriber = TestSubscriber.create();
// when
StringObservable.split(sourceObservable,",")
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(2);
subscriber.assertValues("Lorem ipsum loream", "Lorem ipsum lore");
}
@Test
public void givenStringObservable_whenSplittingByLine_ThenSuccessfullObtainingStringsStream() {
//given
Observable<String> sourceObservable = Observable.just("Lorem ipsum loream\nLorem ipsum lore");
TestSubscriber<String> subscriber = TestSubscriber.create();
// when
StringObservable.byLine(sourceObservable)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(2);
subscriber.assertValues("Lorem ipsum loream", "Lorem ipsum lore");
}
@Test
public void givenStringObservable_whenJoiningStrings_ThenSuccessfullObtainingSingleString() {
//given
Observable<String> sourceObservable = Observable.just("Lorem ipsum loream","Lorem ipsum lore");
TestSubscriber<String> subscriber = TestSubscriber.create();
// when
StringObservable.join(sourceObservable,",")
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValues("Lorem ipsum loream,Lorem ipsum lore");
}
}

View File

@@ -0,0 +1,12 @@
## RxJava Operators
This module contains articles about RxJava Operators
### Related Articles:
- [Mathematical and Aggregate Operators in RxJava](https://www.baeldung.com/rxjava-math)
- [Observable Utility Operators in RxJava](https://www.baeldung.com/rxjava-observable-operators)
- [Implementing Custom Operators in RxJava](https://www.baeldung.com/rxjava-custom-operators)
- [Converting Synchronous and Asynchronous APIs to Observables using RxJava2](https://www.baeldung.com/rxjava-apis-to-observables)

View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>rxjava-operators</artifactId>
<version>1.0-SNAPSHOT</version>
<name>rxjava-operators</name>
<parent>
<groupId>com.baeldung.rxjava-modules</groupId>
<artifactId>rxjava-modules</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<dependencies>
<!-- https://mvnrepository.com/artifact/com.github.akarnokd/rxjava2-extensions -->
<dependency>
<groupId>com.github.akarnokd</groupId>
<artifactId>rxjava2-extensions</artifactId>
<version>${rxjava2.ext.version}</version>
</dependency>
</dependencies>
<properties>
<rxjava2.ext.version>0.20.4</rxjava2.ext.version>
</properties>
</project>

View File

@@ -0,0 +1,42 @@
package com.baeldung.rxjava.operator;
import rx.Observable.Operator;
import rx.Subscriber;
public class ToCleanString implements Operator<String, String> {
public static ToCleanString toCleanString() {
return new ToCleanString();
}
private ToCleanString() {
super();
}
@Override
public Subscriber<? super String> call(final Subscriber<? super String> subscriber) {
return new Subscriber<String>(subscriber) {
@Override
public void onCompleted() {
if (!subscriber.isUnsubscribed()) {
subscriber.onCompleted();
}
}
@Override
public void onError(Throwable t) {
if (!subscriber.isUnsubscribed()) {
subscriber.onError(t);
}
}
@Override
public void onNext(String item) {
if (!subscriber.isUnsubscribed()) {
final String result = item.replaceAll("[^A-Za-z0-9]", "");
subscriber.onNext(result);
}
}
};
}
}

View File

@@ -0,0 +1,20 @@
package com.baeldung.rxjava.operator;
import rx.Observable;
import rx.Observable.Transformer;
public class ToLength implements Transformer<String, Integer> {
public static ToLength toLength() {
return new ToLength();
}
private ToLength() {
super();
}
@Override
public Observable<Integer> call(Observable<String> source) {
return source.map(String::length);
}
}

View File

@@ -0,0 +1,107 @@
package com.baeldung.rxjava;
import static org.junit.Assert.assertEquals;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import hu.akarnokd.rxjava2.async.AsyncObservable;
import io.reactivex.Observable;
public class AsyncAndSyncToObservableIntegrationTest {
AtomicInteger counter = new AtomicInteger();
Callable<Integer> callable = () -> counter.incrementAndGet();
/* Method will execute every time it gets subscribed*/
@Test
public void givenSyncMethod_whenConvertedWithFromCallable_thenReturnObservable() {
Observable<Integer> source = Observable.fromCallable(callable);
for (int i = 1; i < 5; i++) {
source.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(i);
assertEquals(i, counter.get());
}
}
/* Method will execute only once and cache its result.*/
@Test
public void givenSyncMethod_whenConvertedWithStart_thenReturnObservable() {
Observable<Integer> source = AsyncObservable.start(callable);
for (int i = 1; i < 5; i++) {
source.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1);
assertEquals(1, counter.get());
}
}
/* Method will execute only once and cache its result.*/
@Test
public void givenAsyncMethod_whenConvertedWithFromFuture_thenRetrunObservble() {
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<Integer> future = executor.submit(callable);
Observable<Integer> source = Observable.fromFuture(future);
for (int i = 1; i < 5; i++) {
source.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1);
assertEquals(1, counter.get());
}
executor.shutdown();
}
/* Method will execute every time it gets subscribed*/
@Test
public void givenAsyncMethod_whenConvertedWithStartFuture_thenRetrunObservble() {
ExecutorService executor = Executors.newSingleThreadExecutor();
Observable<Integer> source = AsyncObservable.startFuture(() -> executor.submit(callable));
for (int i = 1; i < 5; i++) {
source.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(i);
assertEquals(i, counter.get());
}
executor.shutdown();
}
/*Method will execute only once and cache its result.*/
@Test
public void givenAsyncMethod_whenConvertedWithDeferFuture_thenRetrunObservble() {
List<Integer> list = Arrays.asList(new Integer[] { counter.incrementAndGet(), counter.incrementAndGet(), counter.incrementAndGet() });
ExecutorService exec = Executors.newSingleThreadExecutor();
Callable<Observable<Integer>> callable = () -> Observable.fromIterable(list);
Observable<Integer> source = AsyncObservable.deferFuture(() -> exec.submit(callable));
for (int i = 1; i < 4; i++) {
source.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3);
}
exec.shutdown();
}
}

View File

@@ -0,0 +1,105 @@
package com.baeldung.rxjava;
import org.junit.Test;
import rx.Observable;
import rx.Observable.Operator;
import rx.Observable.Transformer;
import rx.Subscriber;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import static com.baeldung.rxjava.operator.ToCleanString.toCleanString;
import static com.baeldung.rxjava.operator.ToLength.toLength;
import static org.hamcrest.Matchers.hasItems;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertThat;
public class RxJavaCustomOperatorUnitTest {
@Test
public void whenUseCleanStringOperator_thenSuccess() {
final List<String> list = Arrays.asList("john_1", "tom-3");
final List<String> results = new ArrayList<>();
final Observable<String> observable = Observable.from(list)
.lift(toCleanString());
// when
observable.subscribe(results::add);
// then
assertThat(results, notNullValue());
assertThat(results, hasSize(2));
assertThat(results, hasItems("john1", "tom3"));
}
@Test
public void whenUseToLengthOperator_thenSuccess() {
final List<String> list = Arrays.asList("john", "tom");
final List<Integer> results = new ArrayList<>();
final Observable<Integer> observable = Observable.from(list)
.compose(toLength());
// when
observable.subscribe(results::add);
// then
assertThat(results, notNullValue());
assertThat(results, hasSize(2));
assertThat(results, hasItems(4, 3));
}
@Test
public void whenUseFunctionOperator_thenSuccess() {
final Operator<String, String> cleanStringFn = subscriber -> new Subscriber<String>(subscriber) {
@Override
public void onCompleted() {
if (!subscriber.isUnsubscribed()) {
subscriber.onCompleted();
}
}
@Override
public void onError(Throwable t) {
if (!subscriber.isUnsubscribed()) {
subscriber.onError(t);
}
}
@Override
public void onNext(String str) {
if (!subscriber.isUnsubscribed()) {
final String result = str.replaceAll("[^A-Za-z0-9]", "");
subscriber.onNext(result);
}
}
};
final List<String> results = new ArrayList<>();
Observable.from(Arrays.asList("ap_p-l@e", "or-an?ge"))
.lift(cleanStringFn)
.subscribe(results::add);
assertThat(results, notNullValue());
assertThat(results, hasSize(2));
assertThat(results, hasItems("apple", "orange"));
}
@Test
public void whenUseFunctionTransformer_thenSuccess() {
final Transformer<String, Integer> toLengthFn = source -> source.map(String::length);
final List<Integer> results = new ArrayList<>();
Observable.from(Arrays.asList("apple", "orange"))
.compose(toLengthFn)
.subscribe(results::add);
assertThat(results, notNullValue());
assertThat(results, hasSize(2));
assertThat(results, hasItems(5, 6));
}
}

View File

@@ -0,0 +1,270 @@
package com.baeldung.rxjava;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import rx.Observable;
import rx.Observer;
import rx.exceptions.OnErrorNotImplementedException;
import rx.schedulers.Schedulers;
import rx.schedulers.Timestamped;
import java.util.concurrent.TimeUnit;
import static com.jayway.awaitility.Awaitility.await;
import static org.junit.Assert.assertTrue;
public class UtilityOperatorsIntegrationTest {
private int emittedTotal = 0;
private int receivedTotal = 0;
private String result = "";
@Rule
public ExpectedException thrown = ExpectedException.none();
@Test
public void givenObservable_whenObserveOnAfterOnNext_thenEmitsEventsOnComputeScheduler() throws InterruptedException {
Observable.range(1, 5)
.map(i -> i * 100)
.doOnNext(i -> {
emittedTotal += i;
System.out.println("Emitting " + i
+ " on thread " + Thread.currentThread().getName());
})
.observeOn(Schedulers.computation())
.map(i -> i * 10)
.subscribe(i -> {
receivedTotal += i;
System.out.println("Received " + i + " on thread "
+ Thread.currentThread().getName());
});
await().until(() -> {
assertTrue(emittedTotal == 1500);
assertTrue(receivedTotal == 15000);
}
);
}
@Test
public void givenObservable_whenObserveOnBeforeOnNext_thenEmitsEventsOnComputeScheduler() throws InterruptedException {
Observable.range(1, 5)
.map(i -> i * 100)
.observeOn(Schedulers.computation())
.doOnNext(i -> {
emittedTotal += i;
System.out.println("Emitting " + i
+ " on thread " + Thread.currentThread().getName());
})
.map(i -> i * 10)
.subscribe(i -> {
receivedTotal += i;
System.out.println("Received " + i + " on thread "
+ Thread.currentThread().getName());
});
await().until(() -> {
assertTrue(emittedTotal == 1500);
assertTrue(receivedTotal == 15000);
});
}
@Test
public void givenObservable_whenSubscribeOn_thenEmitsEventsOnComputeScheduler() throws InterruptedException {
Observable.range(1, 5)
.map(i -> i * 100)
.doOnNext(i -> {
emittedTotal += i;
System.out.println("Emitting " + i
+ " on thread " + Thread.currentThread().getName());
})
.subscribeOn(Schedulers.computation())
.map(i -> i * 10)
.subscribe(i -> {
receivedTotal += i;
System.out.println("Received " + i + " on thread "
+ Thread.currentThread().getName());
});
await().until(() -> {
assertTrue(emittedTotal == 1500);
assertTrue(receivedTotal == 15000);
});
}
@Test
public void givenObservableWithOneEvent_whenSingle_thenEmitEvent() {
Observable.range(1, 1)
.single()
.subscribe(i -> receivedTotal += i);
assertTrue(receivedTotal == 1);
}
@Test
public void givenObservableWithNoEvents_whenSingle_thenThrowException() {
Observable.range(1, 3)
.single()
.onErrorReturn(e -> receivedTotal += 10)
.subscribe();
assertTrue(receivedTotal == 10);
}
@Test
public void givenObservableWihNoEvents_whenSingleOrDefault_thenDefaultMessage() {
Observable.empty()
.singleOrDefault("Default")
.subscribe(i -> result += i);
assertTrue(result.equals("Default"));
}
@Test
public void givenObservableWithManyEvents_whenSingleOrDefault_thenThrowException() {
Observable.range(1, 3)
.singleOrDefault(5)
.onErrorReturn(e -> receivedTotal += 10)
.subscribe();
assertTrue(receivedTotal == 10);
}
@Test
public void givenObservable_whenDoOnNextAndDoOnCompleted_thenSumAllEventsAndShowMessage() {
Observable.range(1, 10)
.doOnNext(r -> receivedTotal += r)
.doOnCompleted(() -> result = "Completed")
.subscribe();
assertTrue(receivedTotal == 55);
assertTrue(result.equals("Completed"));
}
@Test
public void givenObservable_whenDoOnEachAndDoOnSubscribe_thenSumAllValuesAndShowMessage() {
Observable.range(1, 10)
.doOnEach(new Observer<Integer>() {
@Override
public void onCompleted() {
System.out.println("Complete");
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onNext(Integer value) {
receivedTotal += value;
}
})
.doOnSubscribe(() -> result = "Subscribed")
.subscribe();
assertTrue(receivedTotal == 55);
assertTrue(result.equals("Subscribed"));
}
@Test
public void givenObservable_whenDoOnErrorDoOnTerminateAndDoAfterTerminate_thenShowErrorTerminateAndAfterTerminateMessages() {
thrown.expect(OnErrorNotImplementedException.class);
Observable.empty()
.single()
.doOnError(throwable -> {
throw new RuntimeException("error");
})
.doOnTerminate(() -> result += "doOnTerminate")
.doAfterTerminate(() -> result += "_doAfterTerminate")
.subscribe();
assertTrue(result.equals("doOnTerminate_doAfterTerminate"));
}
@Test
public void givenObservable_whenTimestamp_thenEventsShouldAppearTimestamped() {
Observable.range(1, 10)
.timestamp()
.map(o -> result = o.getClass().toString())
.last()
.subscribe();
assertTrue(result.equals("class rx.schedulers.Timestamped"));
}
@Test
public void givenObservables_whenDelay_thenEventsStartAppearAfterATime() throws InterruptedException {
Observable<Timestamped<Long>> source = Observable.interval(1, TimeUnit.SECONDS)
.take(5)
.timestamp();
Observable<Timestamped<Long>> delay = source.delaySubscription(2, TimeUnit.SECONDS);
source.<Long>subscribe(
value -> System.out.println("source :" + value),
t -> System.out.println("source error"),
() -> System.out.println("source completed"));
delay.subscribe(
value -> System.out.println("delay : " + value),
t -> System.out.println("delay error"),
() -> System.out.println("delay completed"));
//Thread.sleep(8000);
}
@Test
public void givenObservable_whenRepeat_thenSumNumbersThreeTimes() {
Observable.range(1, 3)
.repeat(3)
.subscribe(i -> receivedTotal += i);
assertTrue(receivedTotal == 18);
}
@Test
public void givenObservable_whenUsing_thenReturnCreatedResource() {
Observable<Character> values = Observable.using(
() -> "resource",
r -> Observable.create(o -> {
for (Character c : r.toCharArray()) {
o.onNext(c);
}
o.onCompleted();
}),
r -> System.out.println("Disposed: " + r)
);
values.subscribe(
v -> result += v,
e -> result += e
);
assertTrue(result.equals("resource"));
}
@Test
public void givenObservableCached_whenSubscribesWith2Actions_thenEmitsCachedValues() {
Observable<Integer> source =
Observable.<Integer>create(subscriber -> {
System.out.println("Create");
subscriber.onNext(receivedTotal += 5);
subscriber.onCompleted();
}
).cache();
source.subscribe(i -> {
System.out.println("element 1");
receivedTotal += 1;
});
source.subscribe(i -> {
System.out.println("element 2");
receivedTotal += 2;
});
assertTrue(receivedTotal == 8);
}
}

View File

@@ -0,0 +1,221 @@
package com.baeldung.rxjava.operators;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
public class RxAggregateOperatorsUnitTest {
@Test
public void givenTwoObservable_whenConcatenatingThem_thenSuccessfull() {
// given
List<Integer> listOne = Arrays.asList(1, 2, 3, 4);
Observable<Integer> observableOne = Observable.from(listOne);
List<Integer> listTwo = Arrays.asList(5, 6, 7, 8);
Observable<Integer> observableTwo = Observable.from(listTwo);
TestSubscriber<Integer> subscriber = TestSubscriber.create();
// when
Observable<Integer> concatObservable = observableOne.concatWith(observableTwo);
concatObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(8);
subscriber.assertValues(1, 2, 3, 4, 5, 6, 7, 8);
}
@Test
public void givenObservable_whenCounting_thenObtainingNumberOfElements() {
// given
List<String> lettersList = Arrays.asList("A", "B", "C", "D", "E", "F", "G");
TestSubscriber<Integer> subscriber = TestSubscriber.create();
// when
Observable<Integer> sourceObservable = Observable.from(lettersList)
.count();
sourceObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(7);
}
@Test
public void givenObservable_whenReducing_thenObtainingInvertedConcatenatedString() {
// given
List<String> list = Arrays.asList("A", "B", "C", "D", "E", "F", "G");
TestSubscriber<String> subscriber = TestSubscriber.create();
// when
Observable<String> reduceObservable = Observable.from(list)
.reduce((letter1, letter2) -> letter2 + letter1);
reduceObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue("GFEDCBA");
}
@Test
public void givenObservable_whenCollecting_thenObtainingASet() {
// given
List<String> list = Arrays.asList("A", "B", "C", "B", "B", "A", "D");
TestSubscriber<HashSet> subscriber = TestSubscriber.create();
// when
Observable<HashSet<String>> reduceListObservable = Observable.from(list)
.collect(HashSet::new, HashSet::add);
reduceListObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValues(new HashSet<>(list));
}
@Test
public void givenObservable_whenUsingToList_thenObtainedAList() {
// given
Observable<Integer> sourceObservable = Observable.range(1, 5);
TestSubscriber<List> subscriber = TestSubscriber.create();
// when
Observable<List<Integer>> listObservable = sourceObservable.toList();
listObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(Arrays.asList(1, 2, 3, 4, 5));
}
@Test
public void givenObservable_whenUsingToSortedList_thenObtainedASortedList() {
// given
Observable<Integer> sourceObservable = Observable.range(10, 5);
TestSubscriber<List> subscriber = TestSubscriber.create();
// when
Observable<List<Integer>> listObservable = sourceObservable.toSortedList();
listObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(Arrays.asList(10, 11, 12, 13, 14));
}
@Test
public void givenObservable_whenUsingToSortedListWithComparator_thenObtainedAnInverseSortedList() {
// given
Observable<Integer> sourceObservable = Observable.range(10, 5);
TestSubscriber<List> subscriber = TestSubscriber.create();
// when
Observable<List<Integer>> listObservable = sourceObservable.toSortedList((int1, int2) -> int2 - int1);
listObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(Arrays.asList(14, 13, 12, 11, 10));
}
@Test
public void givenObservable_whenUsingToMap_thenObtainedAMap() {
// given
Observable<Book> bookObservable = Observable
.just(
new Book("The North Water", 2016),
new Book("Origin", 2017),
new Book("Sleeping Beauties", 2017));
TestSubscriber<Map> subscriber = TestSubscriber.create();
// when
Observable<Map<String, Integer>> mapObservable = bookObservable
.toMap(Book::getTitle, Book::getYear, HashMap::new);
mapObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(new HashMap() {
{
put("The North Water", 2016);
put("Origin", 2017);
put("Sleeping Beauties", 2017);
}
});
}
@Test
public void givenObservable_whenUsingToMultiMap_thenObtainedAMultiMap() {
// given
Observable<Book> bookObservable = Observable
.just(
new Book("The North Water", 2016),
new Book("Origin", 2017),
new Book("Sleeping Beauties", 2017));
TestSubscriber<Map> subscriber = TestSubscriber.create();
// when
Observable multiMapObservable = bookObservable
.toMultimap(Book::getYear, Book::getTitle, () -> new HashMap<>(), (key) -> new ArrayList<>());
multiMapObservable.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(new HashMap() {
{
put(2016, Arrays.asList("The North Water"));
put(2017, Arrays.asList("Origin", "Sleeping Beauties"));
}
});
}
class Book {
private String title;
private Integer year;
public Book(String title, Integer year) {
this.title = title;
this.year = year;
}
public String getTitle() {
return title;
}
public Integer getYear() {
return year;
}
}
}

View File

@@ -0,0 +1,65 @@
package com.baeldung.rxjava.operators;
import io.reactivex.Observable;
import io.reactivex.schedulers.TestScheduler;
import org.junit.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.hamcrest.CoreMatchers.hasItems;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
public class RxFlatmapAndSwitchmapUnitTest {
@Test
public void givenObservable_whenFlatmap_shouldAssertAllItemsReturned() {
//given
List<String> actualOutput = new ArrayList<>();
final TestScheduler scheduler = new TestScheduler();
final List<String> keywordToSearch = Arrays.asList("b", "bo", "boo", "book", "books");
//when
Observable.fromIterable(keywordToSearch)
.flatMap(s -> Observable
.just(s + " FirstResult", s + " SecondResult")
.delay(10, TimeUnit.SECONDS, scheduler))
.toList()
.doOnSuccess(s -> actualOutput.addAll(s))
.subscribe();
scheduler.advanceTimeBy(1, TimeUnit.MINUTES);
//then
assertThat(actualOutput, hasItems("b FirstResult", "b SecondResult",
"boo FirstResult", "boo SecondResult",
"bo FirstResult", "bo SecondResult",
"book FirstResult", "book SecondResult",
"books FirstResult", "books SecondResult"));
}
@Test
public void givenObservable_whenSwitchmap_shouldAssertLatestItemReturned() {
//given
List<String> actualOutput = new ArrayList<>();
final TestScheduler scheduler = new TestScheduler();
final List<String> keywordToSearch = Arrays.asList("b", "bo", "boo", "book", "books");
//when
Observable.fromIterable(keywordToSearch)
.switchMap(s -> Observable
.just(s + " FirstResult", s + " SecondResult")
.delay(10, TimeUnit.SECONDS, scheduler))
.toList()
.doOnSuccess(s -> actualOutput.addAll(s))
.subscribe();
scheduler.advanceTimeBy(1, TimeUnit.MINUTES);
//then
assertEquals(2, actualOutput.size());
assertThat(actualOutput, hasItems("books FirstResult", "books SecondResult"));
}
}

View File

@@ -0,0 +1,139 @@
package com.baeldung.rxjava.operators;
import org.junit.Test;
import rx.Observable;
import rx.observables.MathObservable;
import rx.observers.TestSubscriber;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
public class RxMathematicalOperatorsUnitTest {
@Test
public void givenRangeNumericObservable_whenCalculatingAverage_ThenSuccessfull() {
// given
Observable<Integer> sourceObservable = Observable.range(1, 20);
TestSubscriber<Integer> subscriber = TestSubscriber.create();
// when
MathObservable.averageInteger(sourceObservable)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(10);
}
@Test
public void givenRangeNumericObservable_whenCalculatingSum_ThenSuccessfull() {
// given
Observable<Integer> sourceObservable = Observable.range(1, 20);
TestSubscriber<Integer> subscriber = TestSubscriber.create();
// when
MathObservable.sumInteger(sourceObservable)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(210);
}
@Test
public void givenRangeNumericObservable_whenCalculatingMax_ThenSuccessfullObtainingMaxValue() {
// given
Observable<Integer> sourceObservable = Observable.range(1, 20);
TestSubscriber<Integer> subscriber = TestSubscriber.create();
// when
MathObservable.max(sourceObservable)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(20);
}
@Test
public void givenRangeNumericObservable_whenCalculatingMin_ThenSuccessfullObtainingMinValue() {
// given
Observable<Integer> sourceObservable = Observable.range(1, 20);
TestSubscriber<Integer> subscriber = TestSubscriber.create();
// when
MathObservable.min(sourceObservable)
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(1);
}
@Test
public void givenItemObservable_whenCalculatingMaxWithComparator_ThenSuccessfullObtainingMaxItem() {
// given
Item five = new Item(5);
List<Item> list = Arrays.asList(new Item(1), new Item(2), new Item(3), new Item(4), five);
Observable<Item> itemObservable = Observable.from(list);
TestSubscriber<Item> subscriber = TestSubscriber.create();
// when
MathObservable.from(itemObservable)
.max(Comparator.comparing(Item::getId))
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(five);
}
@Test
public void givenItemObservable_whenCalculatingMinWithComparator_ThenSuccessfullObtainingMinItem() {
// given
Item one = new Item(1);
List<Item> list = Arrays.asList(one, new Item(2), new Item(3), new Item(4), new Item(5));
TestSubscriber<Item> subscriber = TestSubscriber.create();
Observable<Item> itemObservable = Observable.from(list);
// when
MathObservable.from(itemObservable)
.min(Comparator.comparing(Item::getId))
.subscribe(subscriber);
// then
subscriber.assertCompleted();
subscriber.assertNoErrors();
subscriber.assertValueCount(1);
subscriber.assertValue(one);
}
class Item {
private Integer id;
public Item(Integer id) {
this.id = id;
}
public Integer getId() {
return id;
}
}
}