fail on an exception and test it better

Author: AndreKoepke

.idea/modules.xml

@@ -37,8 +37,13 @@ dependencies {
     testCompileOnly 'org.projectlombok:lombok:1.18.36'
     testAnnotationProcessor 'org.projectlombok:lombok:1.18.36'
 
-    testImplementation 'org.junit.jupiter:junit-jupiter-api:5.8.2'
-    testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.8.2'
+    testImplementation 'org.junit.jupiter:junit-jupiter-api:5.7.0'
+    testImplementation 'org.junit.jupiter:junit-jupiter-params:5.7.0'
+    testImplementation 'org.assertj:assertj-core:3.17.2'
+    testImplementation 'org.mockito:mockito-core:3.5.13'
+
+    testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.7.0'
+
     testImplementation 'org.slf4j:slf4j-api:1.7.36'
     testImplementation 'org.slf4j:slf4j-simple:1.7.36'
 
@@ -88,21 +93,3 @@ publishing {
         }
     }
 }
-
-jreleaser {
-    signing {
-        active = 'ALWAYS'
-        armored = true
-    }
-    deploy {
-        maven {
-            mavenCentral {
-                sonatype {
-                    active = 'ALWAYS'
-                    url = 'https://central.sonatype.com/api/v1/publisher'
-                    stagingRepository('build/staging-deploy')
-                }
-            }
-        }
-    }
-}

build.gradle

@@ -18,6 +18,7 @@
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
 import java.net.URI;
@@ -30,11 +31,11 @@
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Use this scraper to fetch data from Weathercloud.
  */
+@Slf4j
 public class Scraper {
 
     private static final String BASE_URL = "https://app.weathercloud.net/device/stats?code=";
@@ -102,24 +103,19 @@ private String doRequestAndGetBody(String deviceId) throws URISyntaxException, I
 
     /**
      * Pull weather-data at a constant rate. It will fetch the first dataset immediately.
+     * On an exception, the Observable will be stopped. The caller needs a logic to recover from a bad state.
      *
      * @param deviceId numeric deviceId, like 1234567890
      * @param rate     Duration, how often data should be fetched.
      * @return A flowable, which will emit on each new value (same values will not be emitted twice)
      */
-    public Flowable<Weather> scrape$(String deviceId, Duration rate) {
-        // because of backoff-logic, we need to prevent that the timer-emits will stack up
-        AtomicBoolean isWorking = new AtomicBoolean(false);
+    public static Flowable<Weather> scrape$(String deviceId, Duration rate, Scraper scraper) {
         return Observable.interval(0, rate.toMillis(), TimeUnit.MILLISECONDS, Schedulers.io())
-                .filter(aLong -> !isWorking.get())
-                .flatMap(ignored -> {
-                    isWorking.set(true);
-
-                    return Observable.fromSupplier(() -> this.scrape(deviceId))
-                            .doOnComplete(() -> isWorking.set(false))
-                            .subscribeOn(Schedulers.computation());
-                })
+                .concatMap(ignored -> Observable.fromSupplier(() -> scraper.scrape(deviceId))
+                        .subscribeOn(Schedulers.computation())
+                )
+                .onErrorComplete()
                 .distinct(Weather::getRecordedAt)
-                .toFlowable(BackpressureStrategy.DROP);
+                .toFlowable(BackpressureStrategy.BUFFER);
     }
 }

src/main/java/ch/akop/weathercloud/scraper/weathercloud/Scraper.java

@@ -1,33 +1,82 @@
 package ch.akop.weathercloud.scraper.weathercloud;
 
+import ch.akop.weathercloud.Weather;
+import ch.akop.weathercloud.light.Light;
+import ch.akop.weathercloud.light.LightUnit;
+import ch.akop.weathercloud.rain.Rain;
+import ch.akop.weathercloud.rain.RainUnit;
+import ch.akop.weathercloud.temperature.Temperature;
+import ch.akop.weathercloud.temperature.TemperatureUnit;
+import ch.akop.weathercloud.wind.Wind;
+import ch.akop.weathercloud.wind.WindSpeedUnit;
+import io.reactivex.rxjava3.observers.TestObserver;
 import lombok.extern.slf4j.Slf4j;
-import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.math.BigDecimal;
 import java.time.Duration;
+import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 
 @Slf4j
 class ScraperTest {
 
+    Scraper testee;
+
+    @BeforeEach
+    void setup() {
+        testee = spy(new Scraper());
+    }
+
     @Test
     void manual_test() {
-        var testee = new Scraper();
         var result = testee.scrape("7003523537");
         assertNotNull(result);
     }
 
     @Test
-    @Disabled("Run manually")
-    void manual_scheduler_test() throws InterruptedException {
-        var testee = new Scraper();
+    public void manual_scheduler_test() throws InterruptedException {
+        var value1 = new Weather()
+                .setRecordedAt(ZonedDateTime.now())
+                .setLight(Light.fromUnit(BigDecimal.ZERO, LightUnit.KILO_LUX))
+                .setOuterTemperatur(Temperature.fromUnit(BigDecimal.ZERO, TemperatureUnit.DEGREE))
+                .setWind(Wind.fromUnit(BigDecimal.ZERO, WindSpeedUnit.KILOMETERS_PER_SECOND))
+                .setRain(Rain.fromUnit(BigDecimal.ZERO, RainUnit.MILLIMETER_PER_HOUR));
+
+        var value2 = new Weather()
+                .setRecordedAt(ZonedDateTime.now().plusHours(12))
+                .setLight(Light.fromUnit(BigDecimal.ONE, LightUnit.KILO_LUX))
+                .setOuterTemperatur(Temperature.fromUnit(BigDecimal.ONE, TemperatureUnit.DEGREE))
+                .setWind(Wind.fromUnit(BigDecimal.ONE, WindSpeedUnit.KILOMETERS_PER_SECOND))
+                .setRain(Rain.fromUnit(BigDecimal.ONE, RainUnit.MILLIMETER_PER_HOUR));
+
+
+        doReturn(value1)
+                .doThrow(new RuntimeException("blub"))
+                .doReturn(value1)
+                .doReturn(value2)
+                .when(testee)
+                .scrape(anyString());
+
+
+        var tester = new TestObserver<Weather>();
+        Scraper.scrape$("7003523537", Duration.of(1, ChronoUnit.SECONDS), testee)
+                .doOnError(e -> log.warn("Hi"))
+                .toObservable()
+                .subscribe(tester);
 
-        testee.scrape$("7003523537", Duration.of(15, ChronoUnit.SECONDS))
-                .subscribe(weather -> log.info("In Oensingen, we had {}", weather.getOuterTemperatur()));
+        Thread.sleep(100);
+        tester.assertValuesOnly(value1);
+        Thread.sleep(1500);
 
-        Thread.sleep(600000);
+        tester.assertComplete();
     }
 
 }