Full chapter 7

Author: nurkiewicz

build.gradle

@@ -21,6 +21,7 @@ dependencies {
 
     testCompile 'ch.qos.logback:logback-classic:1.1.7'
     testCompile 'org.slf4j:slf4j-api:1.7.21'
+    testCompile 'io.dropwizard:dropwizard-metrics:1.0.3'
 
     testCompile 'org.springframework:spring-context:4.3.4.RELEASE'
     testCompile 'org.springframework:spring-jms:4.3.4.RELEASE'
@@ -38,6 +39,8 @@ dependencies {
 //    testCompile 'com.squareup.retrofit2:converter-jackson:2.0.1'
 
     testCompile 'junit:junit:4.12'
+    testCompile 'org.assertj:assertj-core:3.5.2'
+    testCompile 'org.mockito:mockito-core:2.2.15'
 }
 
 task wrapper(type: Wrapper) {

src/test/java/com/oreilly/rxjava/ch7/Chapter7.java

@@ -5,8 +5,17 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import rx.Observable;
+import rx.schedulers.TimeInterval;
 
 import java.math.BigInteger;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static java.time.Month.*;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 @Ignore
 public class Chapter7 {
@@ -157,4 +166,132 @@ private Observable<Income> guessIncome(Person person) {
 		return Observable.just(new Income(1));
 	}
 
+	@Test
+	public void sample_161() throws Exception {
+		Observable<Person> person = Observable.just(new Person());
+
+		Observable<Income> income = person
+				.flatMap(this::determineIncome)
+				.flatMap(
+						Observable::just,
+						th -> Observable.empty(),
+						Observable::empty)
+				.concatWith(person.flatMap(this::guessIncome))
+				.first();
+	}
+
+	@Test
+	public void sample_175() throws Exception {
+		Observable<Person> person = Observable.just(new Person());
+
+		Observable<Income> income = person
+				.flatMap(this::determineIncome)
+				.flatMap(
+						Observable::just,
+						th -> person.flatMap(this::guessIncome),
+						Observable::empty);
+	}
+
+	@Test
+	public void sample_187() throws Exception {
+		Observable<Person> person = Observable.just(new Person());
+
+		Observable<Income> income = person
+				.flatMap(this::determineIncome)
+				.onErrorResumeNext(th -> {
+					if (th instanceof NullPointerException) {
+						return Observable.error(th);
+					} else {
+						return person.flatMap(this::guessIncome);
+					}
+				});
+	}
+
+	Observable<Confirmation> confirmation() {
+		Observable<Confirmation> delayBeforeCompletion =
+				Observable
+						.<Confirmation>empty()
+						.delay(200, MILLISECONDS);
+		return Observable
+				.just(new Confirmation())
+				.delay(100, MILLISECONDS)
+				.concatWith(delayBeforeCompletion);
+	}
+
+	@Test
+	public void sample_215() throws Exception {
+		confirmation()
+				.timeout(210, MILLISECONDS)
+				.forEach(
+						System.out::println,
+						th -> {
+							if ((th instanceof TimeoutException)) {
+								System.out.println("Too long");
+							} else {
+								th.printStackTrace();
+							}
+						}
+				);
+	}
+
+	Observable<LocalDate> nextSolarEclipse(LocalDate after) {
+		return Observable
+				.just(
+						LocalDate.of(2016, MARCH, 9),
+						LocalDate.of(2016, SEPTEMBER, 1),
+						LocalDate.of(2017, FEBRUARY, 26),
+						LocalDate.of(2017, AUGUST, 21),
+						LocalDate.of(2018, FEBRUARY, 15),
+						LocalDate.of(2018, JULY, 13),
+						LocalDate.of(2018, AUGUST, 11),
+						LocalDate.of(2019, JANUARY, 6),
+						LocalDate.of(2019, JULY, 2),
+						LocalDate.of(2019, DECEMBER, 26))
+				.skipWhile(date -> !date.isAfter(after))
+				.zipWith(
+						Observable.interval(500, 50, MILLISECONDS),
+						(date, x) -> date);
+	}
+
+	@Test
+	public void sample_253() throws Exception {
+		nextSolarEclipse(LocalDate.of(2016, SEPTEMBER, 1))
+				.timeout(
+						() -> Observable.timer(1000, TimeUnit.MILLISECONDS),
+						date -> Observable.timer(100, MILLISECONDS));
+	}
+
+	@Test
+	public void sample_262() throws Exception {
+		Observable<TimeInterval<LocalDate>> intervals =
+				nextSolarEclipse(LocalDate.of(2016, JANUARY, 1))
+						.timeInterval();
+	}
+
+	@Test
+	public void sample_271() throws Exception {
+		Observable<Instant> timestamps = Observable
+				.fromCallable(() -> dbQuery())
+				.doOnSubscribe(() -> log.info("subscribe()"))
+				.doOnRequest(c -> log.info("Requested {}", c))
+				.doOnNext(instant -> log.info("Got: {}", instant));
+
+		timestamps
+				.zipWith(timestamps.skip(1), Duration::between)
+				.map(Object::toString)
+				.subscribe(log::info);
+	}
+
+	private Instant dbQuery() {
+		return Instant.now();
+	}
+
+	@Test
+	public void sample_291() throws Exception {
+		Observable<String> obs = Observable
+				.<String>error(new RuntimeException("Swallowed"))
+				.doOnError(th -> log.warn("onError", th))
+				.onErrorReturn(th -> "Fallback");
+	}
+
 }

src/test/java/com/oreilly/rxjava/ch7/Confirmation.java

@@ -0,0 +1,4 @@
+package com.oreilly.rxjava.ch7;
+
+class Confirmation {
+}

src/test/java/com/oreilly/rxjava/ch7/Monitoring.java

@@ -0,0 +1,88 @@
+package com.oreilly.rxjava.ch7;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Slf4jReporter;
+import com.codahale.metrics.Timer;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.LoggerFactory;
+import rx.Observable;
+
+import java.util.concurrent.TimeUnit;
+
+@Ignore
+public class Monitoring {
+
+	private MetricRegistry metricRegistry;
+
+	@Test
+	public void sample_9() throws Exception {
+		metricRegistry = new MetricRegistry();
+		Slf4jReporter reporter = Slf4jReporter
+				.forRegistry(metricRegistry)
+				.outputTo(LoggerFactory.getLogger(Monitoring.class))
+				.build();
+		reporter.start(1, TimeUnit.SECONDS);
+	}
+
+	@Test
+	public void sample_26() throws Exception {
+		final Observable<Integer> observable = Observable.range(1, 100);
+		final Counter items = metricRegistry.counter("items");
+		observable
+				.doOnNext(x -> items.inc())
+				.subscribe(/* ... */);
+	}
+
+	Observable<Long> makeNetworkCall(long x) {
+		//...
+		return Observable.just(10L);
+	}
+
+	@Test
+	public void sample_38() throws Exception {
+		final Observable<Integer> observable = Observable.range(1, 100);
+
+		Counter counter = metricRegistry.counter("counter");
+		observable
+				.doOnNext(x -> counter.inc())
+				.flatMap(this::makeNetworkCall)
+				.doOnNext(x -> counter.dec())
+				.subscribe(/* ... */);
+	}
+
+	@Test
+	public void sample_55() throws Exception {
+		final Observable<Integer> observable = Observable.range(1, 100);
+		Counter counter = metricRegistry.counter("counter");
+
+		observable
+				.flatMap(x ->
+						makeNetworkCall(x)
+								.doOnSubscribe(counter::inc)
+								.doOnTerminate(counter::dec)
+				)
+				.subscribe(/* ... */);
+	}
+
+	@Test
+	public void sample_69() throws Exception {
+		Timer timer = metricRegistry.timer("timer");
+		Timer.Context ctx = timer.time();
+		//some lengthy operation...
+		ctx.stop();
+	}
+
+	@Test
+	public void sample_78() throws Exception {
+		Observable<Long> external = Observable.just(42L);
+
+		Timer timer = metricRegistry.timer("timer");
+
+		Observable<Long> externalWithTimer = Observable
+				.defer(() -> Observable.just(timer.time()))
+				.flatMap(timerCtx ->
+						external.doOnCompleted(timerCtx::stop));
+	}
+}

src/test/java/com/oreilly/rxjava/ch7/MyService.java

@@ -0,0 +1,9 @@
+package com.oreilly.rxjava.ch7;
+
+import rx.Observable;
+
+import java.time.LocalDate;
+
+interface MyService {
+    Observable<LocalDate> externalCall();
+}

src/test/java/com/oreilly/rxjava/ch7/MyServiceWithTimeout.java

@@ -0,0 +1,27 @@
+package com.oreilly.rxjava.ch7;
+
+import rx.Observable;
+import rx.Scheduler;
+
+import java.time.LocalDate;
+import java.util.concurrent.TimeUnit;
+
+class MyServiceWithTimeout implements MyService {
+
+    private final MyService delegate;
+    private final Scheduler scheduler;
+
+    MyServiceWithTimeout(MyService d, Scheduler s) {
+        this.delegate = d;
+        this.scheduler = s;
+    }
+
+    @Override
+    public Observable<LocalDate> externalCall() {
+        return delegate
+                .externalCall()
+                .timeout(1, TimeUnit.SECONDS,
+                     Observable.empty(),
+                     scheduler);
+    }
+}