Added typedrest-reactive

Author: bastianeicher

README.md

@@ -63,6 +63,10 @@ Artifact group: [`net.typedrest`](https://mvnrepository.com/artifact/net.typedre
 [![typedrest](https://img.shields.io/maven-central/v/net.typedrest/typedrest.svg?label=typedrest)](https://mvnrepository.com/artifact/net.typedrest/typedrest)  
 The main TypedRest library.
 
+[![typedrest-reactive](https://img.shields.io/maven-central/v/net.typedrest/typedrest-reactive.svg?label=typedrest-reactive)](https://mvnrepository.com/artifact/net.typedrest/typedrest-reactive)  
+Adds support for streaming with [ReactiveX (Rx)](http://reactivex.io/).  
+Create endpoints using the types in the `net.typedrest.endpoints.reactive` package.
+
 [![typedrest-serializers-jackson](https://img.shields.io/maven-central/v/net.typedrest/typedrest-serializers-jackson.svg?label=typedrest-serializers-jackson)](https://mvnrepository.com/artifact/net.typedrest/typedrest-serializers-jackson)  
 Adds support for serializing using [Jackson](https://github.com/FasterXML/jackson) instead of [kotlinx.serialization](https://kotlinlang.org/docs/serialization.html).  
 Pass `new JacksonJsonSerializer()` to the `EntryEndpoint` constructor.

gradle/libs.versions.toml

@@ -1,6 +1,7 @@
 [versions]
 okhttp3 = "4.12.0"
 uri-templates = "2.1.7"
+rxkotlin = "3.0.1"
 kotlinx-serialization = "1.8.0"
 jackson = "2.19.0"
 moshi = "1.15.2"
@@ -9,6 +10,7 @@ moshi = "1.15.2"
 okhttp3 = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp3" }
 okhttp3-mockwebserver = { module = "com.squareup.okhttp3:mockwebserver", version.ref = "okhttp3" }
 uri-templates = { module = "com.damnhandy:handy-uri-templates", version.ref = "uri-templates" }
+rxkotlin = { module = "io.reactivex.rxjava3:rxkotlin", version.ref = "rxkotlin" }
 kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinx-serialization" }
 jackson-databind = { module = "com.fasterxml.jackson.core:jackson-databind", version.ref = "jackson" }
 jackson-kotlin = { module = "com.fasterxml.jackson.module:jackson-module-kotlin", version.ref = "jackson" }

settings.gradle.kts

@@ -1,2 +1,2 @@
 rootProject.name = "typedrest"
-include("typedrest", "typedrest-serializers-jackson", "typedrest-serializers-moshi")
+include("typedrest", "typedrest-reactive", "typedrest-serializers-jackson", "typedrest-serializers-moshi")

typedrest-reactive/build.gradle.kts

@@ -0,0 +1,13 @@
+kotlin.jvmToolchain(21)
+tasks.test { useJUnitPlatform() }
+
+dependencies {
+    api(libs.okhttp3)
+    api(libs.rxkotlin)
+    api(project(":typedrest"))
+    implementation(libs.kotlinx.serialization.json)
+
+    testImplementation(kotlin("test"))
+    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
+    testImplementation(libs.okhttp3.mockwebserver)
+}

typedrest-reactive/src/main/kotlin/net/typedrest/endpoints/reactive/EmitterExtensions.kt

@@ -0,0 +1,18 @@
+package net.typedrest.endpoints.reactive
+
+import io.reactivex.rxjava3.core.ObservableEmitter
+import java.time.Duration
+import kotlin.math.min
+
+/**
+ * Sleep for the specified duration unless the emitter is disposed in the meantime.
+ */
+fun ObservableEmitter<*>.sleep(duration: Duration) {
+    val waitMillis = duration.toMillis()
+    var slept = 0L
+    while (slept < waitMillis && !isDisposed) {
+        val step = min(500L, waitMillis - slept)
+        Thread.sleep(step)
+        slept += step
+    }
+}

typedrest-reactive/src/main/kotlin/net/typedrest/endpoints/reactive/GenericStreamingCollectionEndpoint.kt

@@ -0,0 +1,31 @@
+package net.typedrest.endpoints.reactive
+
+import io.reactivex.rxjava3.core.Observable
+import net.typedrest.endpoints.generic.ElementEndpoint
+import net.typedrest.endpoints.generic.GenericCollectionEndpoint
+import java.time.Duration
+
+/**
+ * Endpoint for a collection of [TEntity]s observable as an append-only stream.
+ *
+ * Use the more constrained [StreamingCollectionEndpoint] when possible.
+ *
+ * @param TEntity The type of individual elements in the collection.
+ * @param TElementEndpoint The type of [ElementEndpoint] to provide for individual [TEntity]s.
+ */
+interface GenericStreamingCollectionEndpoint<TEntity : Any, TElementEndpoint : ElementEndpoint<TEntity>>
+    : GenericCollectionEndpoint<TEntity, TElementEndpoint> {
+    /**
+     * Interval in which requests are sent to the server.
+     * The server may update this interval via the `Retry-After` response header.
+     */
+    var pollingInterval: Duration
+
+    /**
+     * Provides an observable stream of elements.
+     *
+     * @param startIndex The index of the first element to return in the stream. Use negative values to start counting from the end of the stream.
+     * @return A cold observable. HTTP communication only starts once [Observable.subscribe] is invoked
+     */
+    fun getObservable(startIndex: Long = 0): Observable<TEntity>
+}

typedrest-reactive/src/main/kotlin/net/typedrest/endpoints/reactive/GenericStreamingCollectionEndpointImpl.kt

@@ -0,0 +1,81 @@
+package net.typedrest.endpoints.reactive
+
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.schedulers.Schedulers
+import net.typedrest.endpoints.Endpoint
+import net.typedrest.endpoints.generic.ElementEndpoint
+import net.typedrest.endpoints.generic.GenericCollectionEndpointImpl
+import net.typedrest.errors.ConflictException
+import net.typedrest.http.retryAfterDuration
+import okhttp3.Response
+import java.net.URI
+import java.time.Duration
+import java.time.Duration.ofSeconds
+
+/**
+ * Endpoint for a collection of [TEntity]s observable as an append-only stream.
+ *
+ * Use the more constrained [StreamingCollectionEndpointImpl] when possible.
+ *
+ * @param referrer The endpoint used to navigate to this one.
+ * @param relativeUri The URI of this endpoint relative to the [referrer]'s.
+ * @param entityType The type of individual elements in the collection.
+ * @param elementEndpointFactory The factory for constructing [TElementEndpoint]s to provide for individual elements.
+ * @param TEntity The type of individual elements in the collection.
+ * @param TElementEndpoint The type of [ElementEndpoint] to provide for individual [TEntity]s.
+ */
+open class GenericStreamingCollectionEndpointImpl<TEntity : Any, TElementEndpoint : ElementEndpoint<TEntity>>(
+    referrer: Endpoint,
+    relativeUri: URI,
+    entityType: Class<TEntity>,
+    elementEndpointFactory: (referrer: Endpoint, relativeUri: URI) -> TElementEndpoint
+) : GenericCollectionEndpointImpl<TEntity, TElementEndpoint>(referrer, relativeUri, entityType, elementEndpointFactory),
+    GenericStreamingCollectionEndpoint<TEntity, TElementEndpoint> {
+    /**
+     * Creates a new streaming collection endpoint.
+     *
+     * @param referrer The endpoint used to navigate to this one.
+     * @param relativeUri The URI of this endpoint relative to the [referrer]'s. Add a `./` prefix here to imply a trailing slash on referrer's URI.
+     * @param entityType The type of individual elements in the collection.
+     * @param elementEndpointFactory The factory for constructing [TElementEndpoint]s to provide for individual elements.
+     */
+    constructor(referrer: Endpoint, relativeUri: String, entityType: Class<TEntity>, elementEndpointFactory: (referrer: Endpoint, relativeUri: URI) -> TElementEndpoint) :
+        this(referrer, URI(relativeUri), entityType, elementEndpointFactory)
+
+    override fun handle(response: Response): Response {
+        val response = super.handle(response)
+        response.retryAfterDuration()?.let { pollingInterval = it }
+        return response
+    }
+
+    override var pollingInterval: Duration = ofSeconds(3)
+
+    override fun getObservable(startIndex: Long): Observable<TEntity> =
+        Observable.create<TEntity> { emitter ->
+            var currentStartIndex: Long = startIndex
+
+            while (!emitter.isDisposed) {
+                val response = try {
+                    if (currentStartIndex >= 0) readRange(from = currentStartIndex, to = null) // Offset
+                    else readRange(from = null, to = -currentStartIndex) // Tail
+                } catch (_: ConflictException) {
+                    // No new data available yet, keep polling
+                    continue
+                }
+
+                response.elements.forEach(emitter::onNext)
+
+                if (response.endReached) {
+                    emitter.onComplete()
+                    return@create
+                }
+
+                // Continue polling for more data
+                response.range?.to?.let { to -> currentStartIndex = to + 1 }
+                    ?: return@create
+
+                emitter.sleep(pollingInterval)
+            }
+        }
+            .subscribeOn(Schedulers.io())
+}

typedrest-reactive/src/main/kotlin/net/typedrest/endpoints/reactive/PollingEndpoint.kt

@@ -0,0 +1,25 @@
+package net.typedrest.endpoints.reactive
+
+import io.reactivex.rxjava3.core.Observable
+import net.typedrest.endpoints.generic.ElementEndpoint
+import java.time.Duration
+
+/**
+ * Endpoint for a resource that can be polled for state changes.
+ *
+ * @param TEntity The type of entity the endpoint represents.
+ */
+interface PollingEndpoint<TEntity : Any> : ElementEndpoint<TEntity> {
+    /**
+     * Interval in which requests are sent to the server.
+     * The server may update this interval via the `Retry-After` response header.
+     */
+    var pollingInterval: Duration
+
+    /**
+     * Returns an [Observable] stream of entity states.
+     * Consecutive items are emitted only when the server-supplied entity
+     * differs from the previously emitted one according to [Any.equals].
+     */
+    fun getObservable(): Observable<TEntity>
+}

typedrest-reactive/src/main/kotlin/net/typedrest/endpoints/reactive/PollingEndpointImpl.kt

@@ -0,0 +1,65 @@
+package net.typedrest.endpoints.reactive
+
+import io.reactivex.rxjava3.core.Observable
+import io.reactivex.rxjava3.schedulers.Schedulers
+import net.typedrest.endpoints.Endpoint
+import net.typedrest.endpoints.generic.ElementEndpointImpl
+import net.typedrest.http.retryAfterDuration
+import okhttp3.Response
+import java.net.URI
+import java.time.Duration
+import java.time.Duration.ofSeconds
+
+/**
+ * Endpoint for a resource that can be polled for state changes.
+ *
+ * @param referrer The endpoint used to navigate to this one.
+ * @param relativeUri The URI of this endpoint relative to the [referrer]'s.
+ * @param entityType The type of entity the endpoint represents.
+ * @param endCondition A check to determine whether the entity has reached its final state and no further polling is required.
+ * @param TEntity The type of entity the endpoint represents.
+ */
+class PollingEndpointImpl<TEntity : Any>(
+    referrer: Endpoint,
+    relativeUri: URI,
+    entityType: Class<TEntity>,
+    private val endCondition: ((TEntity) -> Boolean)? = null
+) : ElementEndpointImpl<TEntity>(referrer, relativeUri, entityType), PollingEndpoint<TEntity> {
+    /**
+     * Creates a new polling endpoint.
+     *
+     * @param referrer The endpoint used to navigate to this one.
+     * @param relativeUri The URI of this endpoint relative to the [referrer]'s. Add a `./` prefix here to imply a trailing slash on referrer's URI.
+     * @param entityType The type of entity the endpoint represents.
+     * @param endCondition A check to determine whether the entity has reached its final state and no further polling is required.
+     */
+    constructor(
+        referrer: Endpoint,
+        relativeUri: String,
+        entityType: Class<TEntity>,
+        endCondition: ((TEntity) -> Boolean)? = null
+    ) : this(referrer, URI(relativeUri), entityType, endCondition)
+
+    override fun handle(response: Response): Response {
+        val response = super.handle(response)
+        response.retryAfterDuration()?.let { pollingInterval = it }
+        return response
+    }
+
+    override var pollingInterval: Duration = ofSeconds(3)
+
+    override fun getObservable(): Observable<TEntity> =
+        Observable.create<TEntity> { emitter ->
+            while (!emitter.isDisposed) {
+                val entity = read()
+                emitter.onNext(entity)
+                if (endCondition?.invoke(entity) == true) break
+
+                emitter.sleep(pollingInterval)
+            }
+
+            emitter.onComplete()
+        }
+            .distinctUntilChanged()
+            .subscribeOn(Schedulers.io())
+}

typedrest-reactive/src/main/kotlin/net/typedrest/endpoints/reactive/StreamingCollectionEndpoint.kt

@@ -0,0 +1,13 @@
+package net.typedrest.endpoints.reactive
+
+import net.typedrest.endpoints.generic.ElementEndpoint
+
+/**
+ * Endpoint for a collection of [TEntity]s observable as an append-only stream.
+ *
+ * Use [GenericStreamingCollectionEndpoint] instead if you wish to customize the element endpoint type.
+ *
+ * @param TEntity The type of entity the endpoint represents.
+ */
+interface StreamingCollectionEndpoint<TEntity : Any>
+    : GenericStreamingCollectionEndpoint<TEntity, ElementEndpoint<TEntity>>