Add Flowable

README.md

@@ -1,6 +1,7 @@
 # Java-Library to fetch Weathercloud.net data
 
-Small and simple. Just include the lib via Maven/Gradle
+Small and simple. Just include the lib via Maven/Gradle 
+(don't forget to add "https://maven.pkg.github.com/AndreKoepke/weathercloud" as repository)
 
 ### maven
 
@@ -9,20 +10,26 @@ Small and simple. Just include the lib via Maven/Gradle
 <dependency>
     <groupId>ch.akop</groupId>
     <artifactId>weathercloud</artifactId>
-    <version>0.0.3</version>
+    <version>0.0.4</version>
 </dependency> 
 ```
 
 ### gradle
 
 ```groovy
-implementation ch.akop:weathercloud:0.0.3 
+implementation 'ch.akop:weathercloud:0.0.4' 
 ```
 
 ## Usage
 
 ```java
-var response=new Scraper().scrape("7003523537");
+// poll data periodically
+new Scraper().scrape$("7003523537", Duration.of(5, ChronoUnit.MINUTES))
+        .subscribe(weather -> log.info("In Oensingen, we had {}.", weather.getOuterTemperature()));
+// example log-output: "In Oensingen, we had 15.4 °C."
+
+// single result
+var result = new Scraper().scrape("7003523537");
 ```
 
 ![Example response](doc/resultImage.png)

build.gradle

@@ -4,7 +4,7 @@ plugins {
 }
 
 group = 'ch.akop'
-version = '0.0.3'
+version = '0.0.4'
 
 java {
     sourceCompatibility = 17
@@ -21,7 +21,6 @@ dependencies {
     implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.13.2'
     implementation 'com.fasterxml.jackson.core:jackson-core:2.13.2'
     implementation 'org.jetbrains:annotations:23.0.0'
-    testImplementation 'org.mockito:mockito-core:4.4.0'
 
     compileOnly 'org.projectlombok:lombok:1.18.22'
     annotationProcessor 'org.projectlombok:lombok:1.18.22'
@@ -30,6 +29,10 @@ dependencies {
 
     testImplementation 'org.junit.jupiter:junit-jupiter-api:5.8.2'
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.8.2'
+    testImplementation 'org.slf4j:slf4j-api:1.7.36'
+    testImplementation 'org.slf4j:slf4j-simple:1.7.36'
+
+    implementation 'io.reactivex.rxjava3:rxjava:3.1.3'
 }
 
 test {

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

@@ -13,15 +13,21 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.module.SimpleModule;
+import io.reactivex.rxjava3.core.BackpressureStrategy;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.schedulers.Schedulers;
 import lombok.SneakyThrows;
 
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
+import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Use this scraper to fetch data from Weathercloud.
@@ -41,7 +47,7 @@ public class Scraper {
      * <p>
      * Each device has an ID, you can see it, when you open the device in a webbrowser and copy the id from the url.
      *
-     * @param deviceId numeric url, like 1234567890
+     * @param deviceId numeric deviceId, like 1234567890
      * @return Parsed weather-data
      */
     @SneakyThrows
@@ -50,6 +56,7 @@ public Weather scrape(String deviceId) {
                 .GET()
                 .uri(new URI(BASE_URL + deviceId))
                 .header("X-Requested-With", "XMLHttpRequest")
+                .timeout(Duration.ofSeconds(30))
                 .build();
 
         var strResponse = client.send(request, HttpResponse.BodyHandlers.ofString()).body();
@@ -63,4 +70,19 @@ public Weather scrape(String deviceId) {
                 .setRain(Rain.fromUnit(response.rainCurrent().value(), RainUnit.MILLIMETER_PER_HOUR));
     }
 
+
+    /**
+     * Pull weather-data at a constant rate. It will fetch the first dataset immediately.
+     *
+     * @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) {
+        return Observable.interval(0, rate.toMillis(), TimeUnit.MILLISECONDS)
+                .subscribeOn(Schedulers.io())
+                .map(ignored -> this.scrape(deviceId))
+                .distinct(Weather::getRecordedAt)
+                .toFlowable(BackpressureStrategy.DROP);
+    }
 }

src/test/java/ch/akop/weathercloud/scraper/weathercloud/ScraperTest.java

@@ -1,17 +1,30 @@
 package ch.akop.weathercloud.scraper.weathercloud;
 
-import org.junit.jupiter.api.Test;
+import lombok.extern.slf4j.Slf4j;
 
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+
+@Slf4j
 class ScraperTest {
 
     // run manually
-    @Test
+    //@Test
     void manual_test() {
         var testee = new Scraper();
 
         var result = testee.scrape("7003523537");
 
+    }
+
+    //@Test
+    void manual_scheduler_test() throws InterruptedException {
+        var testee = new Scraper();
+
+        new Scraper().scrape$("7003523537", Duration.of(5, ChronoUnit.MINUTES))
+                .subscribe(weather -> log.info("In Oensingen, we had {}", weather.getOuterTemperatur()));
 
+        Thread.sleep(60000);
     }
 
 }