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,12 @@
+kotlin.jvmToolchain(21)
+tasks.test { useJUnitPlatform() }
+
+dependencies {
+    api(libs.okhttp3)
+    api(libs.rxkotlin)
+    api(project(":typedrest"))
+
+    testImplementation(kotlin("test"))
+    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
+    testImplementation(libs.okhttp3.mockwebserver)
+}

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,87 @@
+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 okhttp3.Response
+import java.net.URI
+import java.time.Duration
+import java.time.Duration.ofSeconds
+import java.time.ZonedDateTime
+import java.time.format.DateTimeFormatter
+import kotlin.math.min
+import java.time.Instant
+
+/**
+ * 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.header("Retry-After")?.let { value ->
+            val newInterval = value.toLongOrNull()
+                ?.takeIf { it > 0 }
+                ?.let { ofSeconds(it) }
+                ?: runCatching {
+                    val date = DateTimeFormatter.RFC_1123_DATE_TIME.parse(value, ZonedDateTime::from)
+                    Duration.between(Instant.now(), date.toInstant())
+                }.getOrNull()
+
+            if (newInterval != null && !newInterval.isNegative)
+                pollingInterval = newInterval
+        }
+
+        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
+
+                // TODO: Make non-blocking
+                val waitMillis = pollingInterval.toMillis()
+                var slept = 0L
+                while (slept < waitMillis && !emitter.isDisposed) {
+                    val step = min(500L, waitMillis - slept)
+                    Thread.sleep(step)
+                    slept += step
+                }
+            }
+
+            emitter.onComplete()
+        }
+            .distinctUntilChanged()
+            .subscribeOn(Schedulers.io())
+}

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

@@ -0,0 +1,18 @@
+package net.typedrest.endpoints.reactive
+
+import io.reactivex.rxjava3.core.Observable
+import net.typedrest.endpoints.Endpoint
+
+/**
+ * Endpoint for a stream of [TEntity]s.
+ *
+ * @param TEntity The type of individual elements in the stream.
+ */
+interface StreamingEndpoint<TEntity : Any> : Endpoint {
+    /**
+     * Provides an [Observable] stream of entities.
+     *
+     * @return A cold observable. HTTP communication only starts once [Observable.subscribe] is invoked.
+     */
+    fun getObservable(): Observable<TEntity>
+}

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

@@ -0,0 +1,62 @@
+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.AbstractEndpoint
+import net.typedrest.http.entitySequence
+import net.typedrest.http.uri
+import okhttp3.Request
+import okio.ByteString.Companion.encodeUtf8
+import java.net.URI
+
+/**
+ * Endpoint for a stream of [TEntity]s using a persistent HTTP connection.
+ *
+ * @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 stream.
+ * @param separator The character sequence used to detect that a new element starts in an HTTP stream.
+ * @param TEntity The type of individual elements in the stream.
+ */
+class StreamingEndpointImpl<TEntity : Any>(
+    referrer: Endpoint,
+    relativeUri: URI,
+    private val entityType: Class<TEntity>,
+    private val separator: String = "\n"
+) : AbstractEndpoint(referrer, relativeUri), StreamingEndpoint<TEntity> {
+    /**
+     * Creates a new streaming 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 stream.
+     * @param separator The character sequence used to detect that a new element starts in an HTTP stream.
+     */
+    constructor(
+        referrer: Endpoint,
+        relativeUri: String,
+        entityType: Class<TEntity>,
+        separator: String = "\n"
+    ) : this(referrer, URI(relativeUri), entityType, separator)
+
+    /**
+     * The size of the buffer used to collect data for deserialization in bytes.
+     */
+    var bufferSize: Int = DEFAULT_BUFFER_SIZE
+
+    override fun getObservable(): Observable<TEntity> = Observable.create<TEntity> { emitter ->
+        val call = httpClient.newCall(Request.Builder().get().uri(uri).build())
+        emitter.setCancellable(call::cancel)
+
+        call.execute()
+            .use { response ->
+                response
+                    .entitySequence(serializers.first(), separator.encodeUtf8(), bufferSize, entityType)
+                    .takeWhile { !emitter.isDisposed }
+                    .forEach(emitter::onNext)
+            }
+
+        emitter.onComplete()
+    }.subscribeOn(Schedulers.io())
+}

typedrest-reactive/src/main/kotlin/net/typedrest/http/HttpEntitySequence.kt

@@ -0,0 +1,52 @@
+package net.typedrest.http
+
+import net.typedrest.serializers.Serializer
+import okhttp3.MediaType
+import okhttp3.Response
+import okhttp3.ResponseBody.Companion.toResponseBody
+import okio.Buffer
+import okio.ByteString
+
+/**
+ * Streams deserialized entities from an HTTP response body.
+ *
+ * @param serializer Used to deserialize entities in the body.
+ * @param separator The character sequence used to detect that a new element starts in an HTTP stream.
+ * @param bufferSize The size of the buffer used to collect data for deserialization in bytes.
+ * @param entityType The entity type this stream provides.
+ * @param TEntity The entity type this stream provides.
+ */
+fun <TEntity : Any> Response.entitySequence(
+    serializer: Serializer,
+    separator: ByteString,
+    bufferSize: Int,
+    entityType: Class<TEntity>
+): Sequence<TEntity> = sequence {
+    val src = body!!.source()
+    val buf = Buffer()
+
+    while (src.read(buf, bufferSize.toLong()) != -1L || !buf.exhausted()) {
+        var separatorIndex = buf.indexOf(separator)
+        while (separatorIndex != -1L) {
+            val bytes = buf.readByteArray(separatorIndex)
+            buf.skip(separator.size.toLong())
+
+            deserialize(serializer, bytes, entityType)?.let { yield(it) }
+            separatorIndex = buf.indexOf(separator)
+        }
+    }
+
+    if (!buf.exhausted()) {
+        deserialize(serializer, buf.readByteArray(), entityType)?.let { yield(it) }
+    }
+}
+
+private fun <T : Any> deserialize(
+    serializer: Serializer,
+    bytes: ByteArray,
+    type: Class<T>
+): T? {
+    if (bytes.isEmpty()) return null
+    val mediaType: MediaType? = serializer.supportedMediaTypes.firstOrNull()
+    return serializer.deserialize(bytes.toResponseBody(mediaType), type)
+}

typedrest-reactive/src/test/kotlin/net/typedrest/MockEntity.kt

@@ -0,0 +1,3 @@
+package net.typedrest
+
+data class MockEntity(val id: Long, val name: String)