diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index b00b67f..a361a49 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -68,11 +68,12 @@ jobs:
if: always()
with:
report_paths: |
- reactive/api/build/test-results/test/TEST-*.xml
- reactive/core/build/test-results/test/TEST-*.xml
- reactive/java-flow/build/test-results/test/TEST-*.xml
- reactive/kotlin/build/test-results/test/TEST-*.xml
- reactive/reactivestreams/build/test-results/test/TEST-*.xml
+ reactive/datasource/api/build/test-results/test/TEST-*.xml
+ reactive/datasource/core/build/test-results/test/TEST-*.xml
+ reactive/datasource/java-flow/build/test-results/test/TEST-*.xml
+ reactive/datasource/kotlin/build/test-results/test/TEST-*.xml
+ reactive/datasource/reactivestreams/build/test-results/test/TEST-*.xml
+ reactive/streamlink/kotlin/build/test-results/test/TEST-*.xml
spring/build/test-results/test/TEST-*.xml
util/build/test-results/test/TEST-*.xml
diff --git a/.idea/compiler.xml b/.idea/compiler.xml
index 9cce895..47f80bf 100644
--- a/.idea/compiler.xml
+++ b/.idea/compiler.xml
@@ -16,6 +16,7 @@
+
diff --git a/.idea/gradle.xml b/.idea/gradle.xml
index c630845..207c5ef 100644
--- a/.idea/gradle.xml
+++ b/.idea/gradle.xml
@@ -25,12 +25,14 @@
+
-
-
-
-
-
+
+
+
+
+
+
diff --git a/AGENTS.md b/AGENTS.md
index 04ff343..42eb718 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -35,36 +35,44 @@ The repo is typically checked out on Windows. PowerShell uses `.\gradlew.bat`; B
### Module layout and dependency direction
```
-reactive/api ← config DSL types (ActiveConfig, ChannelConfig, ContainerEvent, …)
-reactive/core ← Binder + IFlowAdapter — does the real DataSource SDK plumbing
- Everything funnels through here as kotlinx.coroutines Flow
-reactive/kotlin ← Bind DSL for Flow } each contributes a Bind facade
-reactive/java-flow ← Bind DSL for java.util.concurrent.Flow.Publisher
-reactive/reactivestreams ← Bind DSL for org.reactivestreams.Publisher
-
-spring ← spring-boot-starter-datasource (depends on reactive/kotlin)
+reactive/datasource/api ← config DSL types (ActiveConfig, ChannelConfig, ContainerEvent, …)
+reactive/datasource/core ← Binder + IFlowAdapter — does the real DataSource SDK plumbing
+ Everything funnels through here as kotlinx.coroutines Flow
+reactive/datasource/kotlin ← Bind DSL for Flow } each contributes a Bind facade
+reactive/datasource/java-flow ← Bind DSL for java.util.concurrent.Flow.Publisher
+reactive/datasource/reactivestreams ← Bind DSL for org.reactivestreams.Publisher
+
+reactive/streamlink/kotlin ← sl4jx-kotlin — Flow-based StreamLink client wrapper
+ (StreamLink + KeyMaster). Depends on datasourcex-util for Subject.
+
+kotest ← datasourcex-kotest — Liberator-in-a-container Kotest harness
+ (LiberatorContainerExtension) for StreamLink client tests. Client-side
+ only: api-depends on sl4jx-kotlin + testcontainers, no datasourcex
+ runtime dependency. Its docker-backed integrationTest is not in `check`.
+spring ← spring-boot-starter-datasource (depends on reactive/datasource/kotlin)
util ← datasourcex-util — FlowMap, custom Flow operators, AntPatternNamespace,
- Fory serialization helpers. No DataSource SDK dependency.
+ Subject, Fory serialization helpers. `api`-depends on the DataSource
+ server SDK (`api(libs.datasource)`).
examples/ ← spring-java, spring-kotlin, spring-kotlin-chat — manual smoke tests
api-docs/ ← aggregated Dokka site (published to GitHub Pages from main)
```
-The published Maven coordinates use the module's renamed name (`datasourcex-kotlin`, `spring-boot-starter-datasource`, etc.), not the Gradle path — set up in `settings.gradle.kts`.
+The five datasource reactive modules sit under `reactive/datasource/` on disk so the StreamLink SDK family can sit beside them under `reactive/streamlink/`. A `projectDir` override in `settings.gradle.kts` relocates only the directory; the `include("reactive:kotlin")` coordinate, the resulting project path (`:reactive:datasourcex-kotlin`, `:reactive:datasourcex-reactive-core`, `:reactive:sl4jx-kotlin`, etc. — set by the `.name` overrides), and the published Maven coordinates are all unchanged. Published coordinates use the renamed name (`datasourcex-kotlin`, `sl4jx-kotlin`, `spring-boot-starter-datasource`, etc.), not the include path.
### Code generation (read this before editing reactive modules)
-The three `reactive/{kotlin,java-flow,reactivestreams}` modules are NOT three parallel hand-written implementations. Each runs a `generateApi` Gradle task (defined in `buildSrc/src/main/kotlin/GenerateApi.kt`, applied via `common-reactive-library`) that emits Kotlin source for the `Bind`, `BindActive`, `BindActiveContainer`, `BindChannel`, `BindBroadcast` DSL classes plus their `Mapping`/`Json`/`Record` flavours.
+The three `reactive/datasource/{kotlin,java-flow,reactivestreams}` modules are NOT three parallel hand-written implementations. Each runs a `generateApi` Gradle task (defined in `buildSrc/src/main/kotlin/GenerateApi.kt`, applied via `common-reactive-library`) that emits Kotlin source for the `Bind`, `BindActive`, `BindActiveContainer`, `BindChannel`, `BindBroadcast` DSL classes plus their `Mapping`/`Json`/`Record` flavours.
Each module's `build.gradle.kts` sets the variant: `tasks.generateApi { publisherType = "kotlin" | "java" | "reactivestreams" }`.
Implications:
-- If you grep for `BindActiveJson` and find nothing, that's because it's generated. Run `./gradlew :reactive:datasourcex-kotlin:generateApi` and look in `reactive/kotlin/build/generated/sources/generateApi/main/kotlin/`.
+- If you grep for `BindActiveJson` and find nothing, that's because it's generated. Run `./gradlew :reactive:datasourcex-kotlin:generateApi` and look in `reactive/datasource/kotlin/build/generated/sources/generateApi/main/kotlin/`.
- The shape of the generated DSL lives in `buildSrc/src/main/kotlin/{Active,ActiveContainer,Channel,Broadcast,Functions}.kt`. Edit those to change the generated API, not the build output.
-- All three published variants share the runtime path through `reactive/core/Binder`. `IFlowAdapter` is the small bridge that converts each library's publisher type to/from `Flow`.
+- All three published variants share the runtime path through `reactive/datasource/core/Binder`. `IFlowAdapter` is the small bridge that converts each library's publisher type to/from `Flow`.
### Spring starter
-`spring` depends on `reactive/kotlin` and exposes endpoints via custom annotations on top of Spring Messaging:
+`spring` depends on `reactive/datasource/kotlin` and exposes endpoints via custom annotations on top of Spring Messaging:
- `@DataService` marks a controller. `@DataMessageMapping("/subject/{var}")` maps a subject pattern to a method (analogous to `@MessageMapping`).
- `@IngressDestinationVariable` extracts path variables.
- Methods return `Flow` / `Flux` / `Publisher` for active subjects, or accept/return `Flow` pairs for channels.
diff --git a/README.md b/README.md
index c04dacb..e4be4bc 100644
--- a/README.md
+++ b/README.md
@@ -42,6 +42,24 @@ Then refer to the documentation:
* [datasourcex-kotlin](https://caplin.github.io/DataSource-Extensions/reactive/datasourcex-kotlin)
* [datasourcex-java-flow](https://caplin.github.io/DataSource-Extensions/reactive/datasourcex-java-flow)
+## StreamLink
+
+`sl4jx-kotlin` is a [Kotlin Flow](https://kotlinlang.org/docs/flow.html)-based client wrapper over
+Caplin's StreamLink and KeyMaster SDKs, for consuming data from a Liberator.
+
+### Usage
+
+Add the following dependency to your project:
+
+```kotlin
+dependencies {
+ implementation("com.caplin.integration.datasourcex:sl4jx-kotlin:")
+}
+```
+
+Then refer to
+the [documentation](https://caplin.github.io/DataSource-Extensions/reactive/sl4jx-kotlin).
+
## Spring
This module provides a starter for integrating Caplin DataSource with your
@@ -61,4 +79,22 @@ dependencies {
Then refer to
the [documentation](https://caplin.github.io/DataSource-Extensions/spring-boot-starter-datasource),
-[hands-on tutorial](./spring/docs/GUIDE.md) and [examples](./examples).
\ No newline at end of file
+[hands-on tutorial](./spring/docs/GUIDE.md) and [examples](./examples).
+
+## Testing
+
+`datasourcex-kotest` is a [Kotest](https://kotest.io/) test-support harness that runs a Liberator in
+a [Testcontainers](https://testcontainers.com/) container for StreamLink client tests.
+
+### Usage
+
+Add the following dependency to your project:
+
+```kotlin
+dependencies {
+ testImplementation("com.caplin.integration.datasourcex:datasourcex-kotest:")
+}
+```
+
+Then refer to
+the [documentation](https://caplin.github.io/DataSource-Extensions/datasourcex-kotest).
\ No newline at end of file
diff --git a/api-docs/build.gradle.kts b/api-docs/build.gradle.kts
index f2e4bc3..fcc5c8e 100644
--- a/api-docs/build.gradle.kts
+++ b/api-docs/build.gradle.kts
@@ -7,6 +7,8 @@ dependencies {
dokka(project(":reactive:datasourcex-java-flow"))
dokka(project(":reactive:datasourcex-kotlin"))
dokka(project(":reactive:datasourcex-reactivestreams"))
+ dokka(project(":reactive:sl4jx-kotlin"))
+ dokka(project(":datasourcex-kotest"))
dokka(project(":spring-boot-starter-datasource"))
dokka(project(":datasourcex-util"))
}
diff --git a/build.gradle.kts b/build.gradle.kts
index e2b932f..d89d29a 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -17,6 +17,7 @@ dependencies {
kover(project(":reactive:datasourcex-java-flow"))
kover(project(":reactive:datasourcex-kotlin"))
kover(project(":reactive:datasourcex-reactivestreams"))
+ kover(project(":reactive:sl4jx-kotlin"))
kover(project(":spring-boot-starter-datasource"))
kover(project(":datasourcex-util"))
}
diff --git a/docs/adr/0002-subject-path-naming-convention.md b/docs/adr/0002-subject-path-naming-convention.md
new file mode 100644
index 0000000..a0500d0
--- /dev/null
+++ b/docs/adr/0002-subject-path-naming-convention.md
@@ -0,0 +1,16 @@
+# A subject's string form is named `path`; a `Subject` object is named `subject`
+
+Across both product lines (`datasourcex` and `sl4jx`), a variable, parameter, or property is named
+by the type it holds: `path` when it is the `String` rendering of a subject, `subject` when it is a
+`Subject`. The convention applies to public and internal code alike and to the `buildSrc` code
+generation, which emits `path: String` DSL parameters (`pathParameter`). Method names describe an
+operation rather than a variable — `getSubject(subject: Subject)`, `extractPathParameters(path: String)` —
+and are governed by what they do, not by this rule.
+
+## Consequences
+
+New and renamed members follow the convention directly. A rename that would break binary
+compatibility — a public `subject: String` property, whose getter would change from `getSubject()` to
+`getPath()` — is introduced by **deprecate-and-add**: the `path` member is added and the old name is
+kept as a `@Deprecated` alias delegating to it. This keeps the convention shippable within a minor
+(3.x) release. The deprecated aliases are removed in a later major release.
diff --git a/docs/adr/0003-no-data-classes-in-public-api.md b/docs/adr/0003-no-data-classes-in-public-api.md
new file mode 100644
index 0000000..035ecb4
--- /dev/null
+++ b/docs/adr/0003-no-data-classes-in-public-api.md
@@ -0,0 +1,24 @@
+# Public value types are plain classes, not `data class`es
+
+Public API value types across both product lines (`datasourcex`, `sl4jx`) are declared as ordinary
+classes with hand-written `equals`/`hashCode`/`toString` where value semantics are wanted, never
+`data class`. A `data class` bakes `copy(...)` and `componentN()` into the binary-compatible surface,
+so adding or reordering a constructor property later is a breaking change — the `copy`/`componentN`
+signatures shift, and `@JvmOverloads` does not extend to the generated `copy`. Hand-writing the value
+methods keeps the constructor free to evolve within a minor release. See
+[Public API challenges in Kotlin](https://jakewharton.com/public-api-challenges-in-kotlin/).
+
+## Consequences
+
+- New public value types follow this directly — e.g. the sl4jx `UpdateEvent`/`StatusEvent`/
+ `ErrorEvent` and `ContainerChange.*`, matching the existing `reactive.api` events (`ContainerEvent`,
+ `BroadcastEvent`), which expose only the `componentN` they choose to hand-write.
+- Losing `data` does not mean losing `copy`: hand-write it when it is wanted. Because you own the
+ signature, a later parameter is added as an overload rather than silently shifting the generated
+ one. `AntPatternNamespace` is the in-repo example — a plain class with a hand-written `copy`.
+- A type may support destructuring by hand-writing `componentN`; it need not expose `copy`.
+- A type that holds non-value fields (streams, containers, connections) is a plain class with no
+ `equals`/`hashCode` at all, rather than a `data class` whose generated equality would be meaningless.
+- Removing a shipped `data class`'s generated `copy`/`componentN` is itself binary-breaking, so a
+ released type is converted by hand-writing whichever of those members callers already use; only a
+ genuinely unused member is dropped outright.
diff --git a/examples/spring-kotlin/build.gradle.kts b/examples/spring-kotlin/build.gradle.kts
index 557613a..580a64f 100644
--- a/examples/spring-kotlin/build.gradle.kts
+++ b/examples/spring-kotlin/build.gradle.kts
@@ -19,3 +19,26 @@ dependencies {
implementation(project(":spring-boot-starter-datasource"))
implementation("org.springframework.boot:spring-boot-starter")
}
+
+// Docker-backed end-to-end test: starts a Liberator Testcontainer, runs this adapter against it and
+// subscribes as a StreamLink client. Not wired into `check`; run on demand with
+// `./gradlew :examples:spring-kotlin:integrationTest`.
+testing {
+ suites {
+ register("integrationTest") {
+ dependencies {
+ implementation(project())
+ implementation(project(":datasourcex-kotest"))
+ implementation(libs.kotest.runner)
+ implementation(libs.kotest.assertions)
+ implementation(libs.turbine)
+ }
+ targets.configureEach { testTask.configure { useJUnitPlatform() } }
+ }
+ }
+}
+
+configurations {
+ named("integrationTestImplementation") { extendsFrom(configurations.implementation.get()) }
+ named("integrationTestRuntimeOnly") { extendsFrom(configurations.runtimeOnly.get()) }
+}
diff --git a/examples/spring-kotlin/src/integrationTest/kotlin/example/SpringKotlinIntegrationTest.kt b/examples/spring-kotlin/src/integrationTest/kotlin/example/SpringKotlinIntegrationTest.kt
new file mode 100644
index 0000000..f96d9a3
--- /dev/null
+++ b/examples/spring-kotlin/src/integrationTest/kotlin/example/SpringKotlinIntegrationTest.kt
@@ -0,0 +1,119 @@
+package example
+
+import app.cash.turbine.test
+import com.caplin.integration.datasourcex.util.LifecycleDataSource
+import com.caplin.integration.datasourcex.util.Subject
+import com.caplin.integration.kotest.LiberatorContainerExtension
+import com.caplin.integration.streamlinkx.StreamLinkConnection.Companion.getSubject
+import com.caplin.integration.streamlinkx.filterIsUpdate
+import com.caplin.integration.streamlinkx.runningFold
+import io.kotest.core.extensions.install
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.collections.shouldHaveSize
+import io.kotest.matchers.nulls.shouldNotBeNull
+import io.kotest.matchers.shouldBe
+import io.kotest.matchers.string.shouldNotBeBlank
+import kotlin.time.Duration.Companion.seconds
+import kotlinx.coroutines.withTimeout
+import org.springframework.beans.factory.getBean
+import org.springframework.boot.builder.SpringApplicationBuilder
+
+/**
+ * End-to-end test: starts a Liberator Testcontainer, runs the `spring-kotlin` example adapter
+ * against it (pointed at the container's mapped DataSource port purely by a startup property) and
+ * subscribes as a StreamLink client, proving the Spring starter serves data through a live
+ * Liberator. Liberator uses dynamic peers/services, so it auto-discovers the adapter with no static
+ * routing config.
+ */
+class SpringKotlinIntegrationTest :
+ FunSpec(
+ {
+ val liberator = install(LiberatorContainerExtension())
+
+ // Run the example unchanged, overriding only the outgoing peer to reach the container.
+ // Passed as a command-line arg so it outranks the example's application.properties.
+ val application =
+ SpringApplicationBuilder(SpringKotlinApplication::class.java)
+ .run(
+ "--caplin.datasource.managed.peer.outgoing=" +
+ "ws://${liberator.containerState.host}:${liberator.dataSourcePort}",
+ )
+ autoClose(AutoCloseable { application.close() })
+
+ val dataSource = application.getBean()
+
+ val streamLink = liberator.connect("alice")
+ autoClose(streamLink)
+
+ beforeSpec {
+ withTimeout(60.seconds) {
+ dataSource.awaitConnected()
+ streamLink.awaitConnected()
+ }
+ }
+
+ context("a client subscribed to the spring-kotlin adapter") {
+ test("serves the public payload stream end-to-end") {
+ streamLink
+ .getSubject(Subject("public", "stream", "foo", "42"))
+ .filterIsUpdate()
+ .test(timeout = 30.seconds) {
+ awaitItem().run {
+ version shouldBe 0
+ parameter1 shouldBe "foo"
+ parameter2 shouldBe 42
+ userId shouldBe null
+ sessionId shouldBe null
+ uuid.shouldNotBeNull()
+ }
+ awaitItem().version shouldBe 1
+ cancelAndIgnoreRemainingEvents()
+ }
+ }
+
+ test("serves the public container's initial bulk of 10 rows") {
+ streamLink
+ .getContainer(Subject("public", "container", "foo", "42"))
+ .filterIsUpdate()
+ .runningFold()
+ .test(timeout = 30.seconds) {
+ var rows = awaitItem()
+ while (rows.size < 10) rows = awaitItem()
+ rows shouldHaveSize 10
+ cancelAndIgnoreRemainingEvents()
+ }
+ }
+
+ test("threads the injected USER_ID into the payload") {
+ // The client omits the username segment; Liberator injects %u (the authenticated
+ // user) via the object map the starter registered from @IngressDestinationVariable.
+ streamLink
+ .getSubject(Subject("user", "stream", "foo", "42"))
+ .filterIsUpdate()
+ .test(timeout = 30.seconds) {
+ awaitItem().run {
+ userId shouldBe "alice"
+ parameter1 shouldBe "foo"
+ parameter2 shouldBe 42
+ }
+ cancelAndIgnoreRemainingEvents()
+ }
+ }
+
+ test("threads the injected USER_ID and persistent-session id into the payload") {
+ // Both the username (%u) and the persistent-session id (%g) are injected by
+ // Liberator.
+ streamLink
+ .getSubject(Subject("session", "stream", "foo", "42"))
+ .filterIsUpdate()
+ .test(timeout = 30.seconds) {
+ awaitItem().run {
+ userId shouldBe "alice"
+ sessionId.shouldNotBeNull().shouldNotBeBlank()
+ }
+ cancelAndIgnoreRemainingEvents()
+ }
+ }
+ }
+ },
+ )
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 65ee58a..2b87acb 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -11,12 +11,20 @@ fory = "1.1.0"
# Caplin Dependencies
datasource = "8.0.12-1871-367ddb14"
+streamlink = "8.0.7-1825-026106f"
+keymaster = "8.0.3-559-041a628"
+
+# Logging (kotlin-logging is not managed by the Spring Boot BOM; slf4j is)
+kotlin-logging = "7.0.13"
# Test Dependencies
kotest = "6.2.2"
turbine = "1.2.1"
mockk = "1.14.11"
+# testcontainers is managed by the Spring Boot BOM; kotest-extensions-testcontainers tracks the
+# kotest core version line (io.kotest:… at 6.2.x, depending on testcontainers 2.x).
+
# Spring BOM should manage other dependencies versions.
# Plugins
@@ -39,11 +47,19 @@ jmh-generator = { module = "org.openjdk.jmh:jmh-generator-annprocess", version.r
# Caplin Dependencies
datasource = { module = "com.caplin.platform.integration.java:datasource", version.ref = "datasource" }
+streamlink = { module = "com.caplin.streamlink:StreamLinkJava", version.ref = "streamlink" }
+keymaster = { module = "com.caplin.platform.KeyMaster:KeyMasterMain", version.ref = "keymaster" }
+
+# Logging
+kotlin-logging = { module = "io.github.oshai:kotlin-logging-jvm", version.ref = "kotlin-logging" }
# Test Dependencies
kotest-runner = { module = "io.kotest:kotest-runner-junit5", version.ref = "kotest" }
kotest-assertions = { module = "io.kotest:kotest-assertions-core", version.ref = "kotest" }
kotest-extensions-spring = { module = "io.kotest:kotest-extensions-spring", version.ref = "kotest" }
+kotest-framework-engine = { module = "io.kotest:kotest-framework-engine", version.ref = "kotest" }
+kotest-extensions-testcontainers = { module = "io.kotest:kotest-extensions-testcontainers", version.ref = "kotest" }
+testcontainers = { module = "org.testcontainers:testcontainers" }
turbine = { module = "app.cash.turbine:turbine", version.ref = "turbine" }
mockk = { module = "io.mockk:mockk", version.ref = "mockk" }
diff --git a/kotest/api/datasourcex-kotest.api b/kotest/api/datasourcex-kotest.api
new file mode 100644
index 0000000..1f201a2
--- /dev/null
+++ b/kotest/api/datasourcex-kotest.api
@@ -0,0 +1,72 @@
+public final class com/caplin/integration/kotest/ExtensionsKt {
+ public static final fun getAwaitConnected (Lcom/caplin/integration/streamlinkx/StreamLinkConnection;)Lkotlin/jvm/functions/Function2;
+}
+
+public final class com/caplin/integration/kotest/LiberatorContainerExtension : io/kotest/core/extensions/MountableExtension, io/kotest/core/listeners/AfterProjectListener, io/kotest/core/listeners/TestListener {
+ public static final field DEFAULT_BASE_IMAGE Ljava/lang/String;
+ public static final field LIBERATOR_DATA_SOURCE_PORT I
+ public static final field LIBERATOR_HTTP_PORT I
+ public fun ()V
+ public fun (Lcom/caplin/integration/kotest/LiberatorContainerExtension$ContainerConfig;Ltools/jackson/databind/ObjectMapper;)V
+ public synthetic fun (Lcom/caplin/integration/kotest/LiberatorContainerExtension$ContainerConfig;Ltools/jackson/databind/ObjectMapper;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+ public fun afterAny (Lio/kotest/core/test/TestCase;Lio/kotest/engine/test/TestResult;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun afterContainer (Lio/kotest/core/test/TestCase;Lio/kotest/engine/test/TestResult;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun afterEach (Lio/kotest/core/test/TestCase;Lio/kotest/engine/test/TestResult;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun afterInvocation (Lio/kotest/core/test/TestCase;ILkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun afterProject (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun afterSpec (Lio/kotest/core/spec/Spec;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun afterTest (Lio/kotest/core/test/TestCase;Lio/kotest/engine/test/TestResult;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun beforeAny (Lio/kotest/core/test/TestCase;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun beforeContainer (Lio/kotest/core/test/TestCase;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun beforeEach (Lio/kotest/core/test/TestCase;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun beforeInvocation (Lio/kotest/core/test/TestCase;ILkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun beforeSpec (Lio/kotest/core/spec/Spec;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun beforeTest (Lio/kotest/core/test/TestCase;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun finalizeSpec (Lkotlin/reflect/KClass;Ljava/util/Map;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun mount (Lkotlin/jvm/functions/Function1;)Lcom/caplin/integration/kotest/LiberatorContainerExtension$Liberator;
+ public synthetic fun mount (Lkotlin/jvm/functions/Function1;)Ljava/lang/Object;
+ public fun prepareSpec (Lkotlin/reflect/KClass;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
+public final class com/caplin/integration/kotest/LiberatorContainerExtension$Config {
+ public fun ()V
+}
+
+public abstract interface class com/caplin/integration/kotest/LiberatorContainerExtension$ContainerConfig {
+ public abstract fun getContainer ()Lorg/testcontainers/containers/GenericContainer;
+ public abstract fun getDataSourcePort ()I
+ public abstract fun getHttpPort ()I
+ public abstract fun getKeymasterConfiguration ()Lcom/caplin/keymaster/IKeyMasterConfiguration;
+}
+
+public final class com/caplin/integration/kotest/LiberatorContainerExtension$CustomContainerConfig : com/caplin/integration/kotest/LiberatorContainerExtension$ContainerConfig {
+ public fun (Lcom/caplin/keymaster/IKeyMasterConfiguration;IILorg/testcontainers/containers/GenericContainer;)V
+ public fun getContainer ()Lorg/testcontainers/containers/GenericContainer;
+ public fun getDataSourcePort ()I
+ public fun getHttpPort ()I
+ public fun getKeymasterConfiguration ()Lcom/caplin/keymaster/IKeyMasterConfiguration;
+}
+
+public final class com/caplin/integration/kotest/LiberatorContainerExtension$DefaultContainerConfig : com/caplin/integration/kotest/LiberatorContainerExtension$ContainerConfig {
+ public fun ()V
+ public fun (Ljava/io/InputStream;Ljava/io/InputStream;Ljava/util/Map;Ljava/lang/String;)V
+ public synthetic fun (Ljava/io/InputStream;Ljava/io/InputStream;Ljava/util/Map;Ljava/lang/String;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+ public final fun getAdapterConfig ()Ljava/io/InputStream;
+ public final fun getBaseImage ()Ljava/lang/String;
+ public final fun getBeforeScript ()Ljava/io/InputStream;
+ public fun getContainer ()Lorg/testcontainers/containers/GenericContainer;
+ public fun getDataSourcePort ()I
+ public final fun getExtraFiles ()Ljava/util/Map;
+ public fun getHttpPort ()I
+ public fun getKeymasterConfiguration ()Lcom/caplin/keymaster/IKeyMasterConfiguration;
+}
+
+public final class com/caplin/integration/kotest/LiberatorContainerExtension$Liberator {
+ public fun (Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;IILorg/testcontainers/containers/ContainerState;)V
+ public final fun connect (Ljava/lang/String;)Lcom/caplin/integration/streamlinkx/StreamLinkConnection;
+ public final fun getContainerState ()Lorg/testcontainers/containers/ContainerState;
+ public final fun getDataSourcePort ()I
+ public final fun getHttpPort ()I
+ public final fun getStreamLinkConnectionFactory ()Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;
+}
+
diff --git a/kotest/build.gradle.kts b/kotest/build.gradle.kts
new file mode 100644
index 0000000..27604b0
--- /dev/null
+++ b/kotest/build.gradle.kts
@@ -0,0 +1,34 @@
+plugins { `common-library` }
+
+description = "Kotest test-support: a Liberator-in-a-container harness for StreamLink client tests"
+
+dependencies {
+ api(platform(libs.spring.boot.dependencies))
+ api(project(":reactive:sl4jx-kotlin"))
+ api(libs.testcontainers)
+ api(libs.kotest.framework.engine)
+ api(libs.kotest.extensions.testcontainers)
+ // jacksonObjectMapper() default; sl4jx-kotlin scopes jackson-module-kotlin as implementation, so
+ // it is not transitive here (jackson-databind is, via sl4jx-kotlin's api).
+ implementation("tools.jackson.module:jackson-module-kotlin")
+}
+
+// Docker-backed integration test against a live Liberator; intentionally not wired into `check`.
+testing {
+ suites {
+ register("integrationTest") {
+ useJUnitJupiter()
+ dependencies {
+ implementation(project())
+ implementation(libs.kotest.runner)
+ implementation(libs.kotest.assertions)
+ implementation(libs.turbine)
+ runtimeOnly("org.slf4j:slf4j-simple")
+ }
+ }
+ }
+}
+
+// Keep the docker-backed integration test off `check`: Kover otherwise triggers every Test task
+// when generating coverage, which `check` depends on.
+kover { currentProject { instrumentation { disabledForTestTasks.add("integrationTest") } } }
diff --git a/kotest/src/integrationTest/kotlin/com/caplin/integration/streamlinkx/StreamLinkTest.kt b/kotest/src/integrationTest/kotlin/com/caplin/integration/streamlinkx/StreamLinkTest.kt
new file mode 100644
index 0000000..545aac7
--- /dev/null
+++ b/kotest/src/integrationTest/kotlin/com/caplin/integration/streamlinkx/StreamLinkTest.kt
@@ -0,0 +1,29 @@
+package com.caplin.integration.streamlinkx
+
+import app.cash.turbine.test
+import com.caplin.integration.datasourcex.util.Subject
+import com.caplin.integration.kotest.LiberatorContainerExtension
+import io.kotest.core.extensions.install
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.maps.shouldContainKey
+
+class StreamLinkTest :
+ FunSpec(
+ {
+ val liberator = install(LiberatorContainerExtension())
+
+ context("As admin user") {
+ val streamLink = liberator.connect("admin")
+ streamLink.connect()
+ streamLink.awaitConnected()
+
+ test("Can fetch /SYSTEM/INFO") {
+ streamLink.getSubject(Subject("SYSTEM", "INFO")).filterIsUpdate().test {
+ awaitItem() shouldContainKey "NodeID"
+ }
+ }
+
+ streamLink.disconnect()
+ }
+ },
+ )
diff --git a/kotest/src/main/kotlin/com/caplin/integration/kotest/Extensions.kt b/kotest/src/main/kotlin/com/caplin/integration/kotest/Extensions.kt
new file mode 100644
index 0000000..5b11c9c
--- /dev/null
+++ b/kotest/src/main/kotlin/com/caplin/integration/kotest/Extensions.kt
@@ -0,0 +1,15 @@
+package com.caplin.integration.kotest
+
+import com.caplin.integration.streamlinkx.StreamLinkConnection
+import io.kotest.core.spec.Spec
+import kotlinx.coroutines.Dispatchers.Default
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeoutOrNull
+
+val StreamLinkConnection.awaitConnected: suspend (Spec) -> Unit
+ get() = {
+ withContext(Default) {
+ withTimeoutOrNull(1000) { awaitConnected() }
+ ?: error("Timed out waiting to connect for $username")
+ }
+ }
diff --git a/kotest/src/main/kotlin/com/caplin/integration/kotest/LiberatorContainerExtension.kt b/kotest/src/main/kotlin/com/caplin/integration/kotest/LiberatorContainerExtension.kt
new file mode 100644
index 0000000..460b847
--- /dev/null
+++ b/kotest/src/main/kotlin/com/caplin/integration/kotest/LiberatorContainerExtension.kt
@@ -0,0 +1,215 @@
+package com.caplin.integration.kotest
+
+import com.caplin.integration.kotest.LiberatorContainerExtension.Config
+import com.caplin.integration.kotest.LiberatorContainerExtension.Liberator
+import com.caplin.integration.streamlinkx.StreamLinkConnection
+import com.caplin.integration.streamlinkx.StreamLinkConnectionFactory
+import com.caplin.integration.streamlinkx.StreamLinkConnectionFactory.Companion.convertDerToPem
+import com.caplin.keymaster.IKeyMasterConfiguration
+import com.caplin.keymaster.KeyMasterHashingAlgorithm.SHA256
+import com.caplin.keymaster.PEMPKCS8KeyMasterConfiguration
+import io.kotest.core.extensions.MountableExtension
+import io.kotest.core.listeners.AfterProjectListener
+import io.kotest.core.listeners.TestListener
+import java.io.InputStream
+import java.security.KeyPairGenerator
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+import org.testcontainers.containers.ContainerState
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.images.builder.ImageFromDockerfile
+import org.testcontainers.images.builder.Transferable
+import tools.jackson.databind.ObjectMapper
+import tools.jackson.module.kotlin.jacksonObjectMapper
+
+/**
+ * A Kotest [MountableExtension] that runs a Caplin Liberator inside a Testcontainers container for
+ * the duration of a test project.
+ *
+ * Mount it on a spec to obtain a [Liberator] handle, which exposes the mapped ports and a factory
+ * for opening authenticated StreamLink connections. The container is started lazily on first
+ * [mount] and stopped once via [afterProject] when the project completes.
+ *
+ * ```
+ * class MySpec : FunSpec({
+ * val liberator = install(LiberatorContainerExtension())
+ * test("...") {
+ * val connection = liberator.connect("alice")
+ * autoClose(connection)
+ * // ...
+ * }
+ * })
+ * ```
+ *
+ * @param containerConfig how the container is built and how clients authenticate against it.
+ * Defaults to a [DefaultContainerConfig] that builds a Liberator image with a generated KeyMaster
+ * key pair. Supply a [CustomContainerConfig] to run against a pre-built container.
+ * @param objectMapper Jackson mapper used when (de)serialising messages over the connection.
+ */
+class LiberatorContainerExtension(
+ private val containerConfig: ContainerConfig = DefaultContainerConfig(),
+ private val objectMapper: ObjectMapper = jacksonObjectMapper(),
+) : MountableExtension, AfterProjectListener, TestListener {
+
+ private companion object {
+ const val LIBERATOR_HTTP_PORT = 18080
+ const val LIBERATOR_DATA_SOURCE_PORT = 19000
+ const val DEFAULT_BASE_IMAGE = "docker-release.caplin.com/platform/core:8.0.19"
+ }
+
+ /**
+ * Describes the container to run and how StreamLink clients authenticate against it.
+ *
+ * Implemented by [DefaultContainerConfig] (builds a Liberator image on the fly) and
+ * [CustomContainerConfig] (wraps a container and configuration you supply).
+ */
+ sealed interface ContainerConfig {
+ /** KeyMaster configuration used to sign the credentials of connecting clients. */
+ val keymasterConfiguration: IKeyMasterConfiguration
+ /** The container to start; its exposed ports are mapped to random host ports. */
+ val container: GenericContainer<*>
+ /** The container port serving the Liberator HTTP/RTTP endpoint. */
+ val httpPort: Int
+ /** The container port serving the DataSource endpoint. */
+ val dataSourcePort: Int
+ }
+
+ /**
+ * A [ContainerConfig] that builds a Liberator image from the bundled Dockerfile and generates a
+ * fresh RSA KeyMaster key pair, deploying the public key into the container.
+ *
+ * @param adapterConfig optional adapter configuration copied to `/app/adapter.conf`.
+ * @param beforeScript optional script copied to `/app/beforeScript.sh` and run before startup.
+ * @param extraFiles additional files to copy into the container, keyed by destination path.
+ * @param baseImage the base Docker image the Liberator image is built from.
+ */
+ class DefaultContainerConfig(
+ val adapterConfig: InputStream? = null,
+ val beforeScript: InputStream? = null,
+ val extraFiles: Map = mutableMapOf(),
+ val baseImage: String = DEFAULT_BASE_IMAGE,
+ ) : ContainerConfig {
+ private val keys = KeyPairGenerator.getInstance("RSA").apply { initialize(1024) }.genKeyPair()
+
+ override val keymasterConfiguration: IKeyMasterConfiguration =
+ PEMPKCS8KeyMasterConfiguration(
+ keys.private.encoded.convertDerToPem().byteInputStream(),
+ SHA256,
+ null,
+ )
+
+ override val container: GenericContainer<*> =
+ GenericContainer(
+ ImageFromDockerfile()
+ .withFileFromClasspath("liberator.sh", "default-liberator/liberator.sh")
+ .withFileFromClasspath("Dockerfile", "default-liberator/Dockerfile")
+ .withBuildArg("BASE_IMAGE", baseImage),
+ )
+ .withExposedPorts(
+ LIBERATOR_HTTP_PORT,
+ LIBERATOR_DATA_SOURCE_PORT,
+ )
+ .withCopyToContainer(
+ Transferable.of(keys.public.encoded),
+ "/app/DeploymentFramework/global_config/ssl/keymaster_public.der",
+ )
+ override val httpPort = LIBERATOR_HTTP_PORT
+ override val dataSourcePort = LIBERATOR_DATA_SOURCE_PORT
+ }
+
+ /**
+ * A [ContainerConfig] for running against a container and KeyMaster configuration you supply
+ * yourself, rather than the one built by [DefaultContainerConfig].
+ */
+ class CustomContainerConfig(
+ override val keymasterConfiguration: IKeyMasterConfiguration,
+ override val httpPort: Int,
+ override val dataSourcePort: Int,
+ override val container: GenericContainer<*>,
+ ) : ContainerConfig
+
+ /** Configuration receiver passed to [mount]; currently exposes no options. */
+ class Config
+
+ /**
+ * A handle to a running Liberator container, returned when the extension is mounted.
+ *
+ * @property streamLinkConnectionFactory factory for opening StreamLink connections to the
+ * container, pre-configured with the host, port and KeyMaster credentials.
+ * @property dataSourcePort the host port mapped to the container's DataSource port.
+ * @property httpPort the host port mapped to the container's HTTP/RTTP port.
+ * @property containerState the underlying Testcontainers container state.
+ */
+ class Liberator(
+ val streamLinkConnectionFactory: StreamLinkConnectionFactory,
+ val dataSourcePort: Int,
+ val httpPort: Int,
+ val containerState: ContainerState,
+ ) {
+ /** Opens a StreamLink connection authenticated as [username]. */
+ fun connect(username: String): StreamLinkConnection =
+ streamLinkConnectionFactory.connect(username)
+ }
+
+ private val container: GenericContainer<*> = containerConfig.container
+
+ private val liberator by lazy {
+ Liberator(
+ StreamLinkConnectionFactory(
+ "rttp://${container.host}:${container.getMappedPort(containerConfig.httpPort)}",
+ containerConfig.keymasterConfiguration,
+ objectMapper,
+ ),
+ container.getMappedPort(containerConfig.dataSourcePort),
+ container.getMappedPort(containerConfig.httpPort),
+ container,
+ )
+ }
+
+ /**
+ * Starts the container (if not already running) and returns a [Liberator] handle.
+ *
+ * For a [DefaultContainerConfig], any configured extra files, before-script and adapter config
+ * are copied into the container before it is started. Subsequent calls return the same handle
+ * without restarting the container.
+ */
+ override fun mount(configure: Config.() -> Unit): Liberator {
+
+ if (container.containerId != null) return liberator
+
+ val config = Config()
+ config.configure()
+
+ when (val containerConfig = containerConfig) {
+ is CustomContainerConfig -> {}
+
+ is DefaultContainerConfig -> {
+ containerConfig.extraFiles.forEach { (path, stream) ->
+ container.withCopyToContainer(Transferable.of(stream.readBytes()), path)
+ }
+
+ containerConfig.beforeScript?.let {
+ container.withCopyToContainer(
+ Transferable.of(it.readBytes()),
+ "/app/beforeScript.sh",
+ )
+ }
+
+ containerConfig.adapterConfig?.let {
+ container.withCopyToContainer(
+ Transferable.of(it.readBytes()),
+ "/app/adapter.conf",
+ )
+ }
+ }
+ }
+
+ container.start()
+ return liberator
+ }
+
+ /** Stops the container once the test project has finished. */
+ override suspend fun afterProject() {
+ withContext(Dispatchers.IO) { container.stop() }
+ }
+}
diff --git a/kotest/src/main/resources/default-liberator/Dockerfile b/kotest/src/main/resources/default-liberator/Dockerfile
new file mode 100644
index 0000000..221c372
--- /dev/null
+++ b/kotest/src/main/resources/default-liberator/Dockerfile
@@ -0,0 +1,37 @@
+ARG BASE_IMAGE
+FROM ${BASE_IMAGE} AS platform
+
+USER root
+
+RUN microdnf update -y && \
+ microdnf install -y unzip wget ca-certificates && \
+ microdnf upgrade -y && \
+ DeploymentFramework/dfw deactivate PermissioningService && \
+ DeploymentFramework/dfw deactivate ChartingService && \
+ DeploymentFramework/dfw deactivate AlertsService && \
+ DeploymentFramework/dfw deactivate WatchlistService && \
+ DeploymentFramework/dfw deactivate TransformerJMX && \
+ DeploymentFramework/dfw deactivate BlotterExport && \
+ DeploymentFramework/dfw deactivate PersistenceService && \
+ DeploymentFramework/dfw deactivate DecisionSupportService && \
+ DeploymentFramework/dfw deactivate RefinerService && \
+ DeploymentFramework/dfw deactivate PersistenceServiceClient && \
+ DeploymentFramework/dfw deactivate TransformerToLiberatorConnection && \
+ DeploymentFramework/dfw activate OpenPermissioning && \
+ rm -rf /tmp/yum*
+
+COPY liberator.sh liberator.sh
+
+RUN chmod 755 liberator.sh
+
+USER platform
+
+ENV SSL_ENABLED=false
+ENV LOCAL_LABEL=liberator
+ENV DISCOVERY_HOST=null
+VOLUME /app/DeploymentFramework/global_config/ssl
+VOLUME /app/DeploymentFramework/global_config/licenses
+EXPOSE 19000
+EXPOSE 18080
+
+CMD ["/app/liberator.sh"]
diff --git a/kotest/src/main/resources/default-liberator/liberator.sh b/kotest/src/main/resources/default-liberator/liberator.sh
new file mode 100644
index 0000000..6cbb91c
--- /dev/null
+++ b/kotest/src/main/resources/default-liberator/liberator.sh
@@ -0,0 +1,49 @@
+#!/bin/bash
+
+cd $(dirname "$0") || exit
+
+FILE=DeploymentFramework/global_config/overrides/servers/Liberator/etc/rttpd.conf
+ORIGINAL="$FILE.orig"
+if [ ! -f "$ORIGINAL" ]; then
+ mv $FILE $ORIGINAL
+fi
+cp $ORIGINAL $FILE
+
+FILE=DeploymentFramework/global_config/environment.conf
+ORIGINAL="$FILE.orig"
+if [ ! -f "$ORIGINAL" ]; then
+ mv $FILE $ORIGINAL
+fi
+cp $ORIGINAL $FILE
+
+echo $'
+ datasrc-pkt-log /dev/null
+ datasrc-ws-port 19000
+ disable-default-data-service TRUE
+ ' >> DeploymentFramework/global_config/overrides/servers/Liberator/etc/rttpd.conf
+
+echo $'
+ datasrc-local-label ${ENV:LOCAL_LABEL}
+ ' >> DeploymentFramework/global_config/overrides/servers/Liberator/etc/rttpd.conf
+
+echo $'
+ http-port 18080
+ ' >> DeploymentFramework/global_config/overrides/servers/Liberator/etc/rttpd.conf
+
+if [ -f /app/adapter.conf ]; then
+ cat /app/adapter.conf >> DeploymentFramework/global_config/overrides/servers/Liberator/etc/rttpd.conf
+fi
+
+if [ -f /app/beforeScript.sh ]; then
+ source /app/beforeScript.sh
+fi
+
+DeploymentFramework/dfw dump > /dev/null
+
+echo '
+*****Active config*****'
+grep -v '^#' /app/DeploymentFramework/global_config/dump/Liberator/rttpd.conf
+echo '*****End active config*****
+'
+
+DeploymentFramework/dfw start-fg Liberator
diff --git a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt b/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt
deleted file mode 100644
index 0a17c80..0000000
--- a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt
+++ /dev/null
@@ -1,9 +0,0 @@
-package com.caplin.integration.datasourcex.reactive.core
-
-internal sealed interface InternalContainerEvent {
- val subject: String
-
- data class Inserted(override val subject: String) : InternalContainerEvent
-
- data class Removed(override val subject: String) : InternalContainerEvent
-}
diff --git a/reactive/api/README.md b/reactive/datasource/api/README.md
similarity index 100%
rename from reactive/api/README.md
rename to reactive/datasource/api/README.md
diff --git a/reactive/api/api/datasourcex-reactive-api.api b/reactive/datasource/api/api/datasourcex-reactive-api.api
similarity index 99%
rename from reactive/api/api/datasourcex-reactive-api.api
rename to reactive/datasource/api/api/datasourcex-reactive-api.api
index 6d5a522..532b9a5 100644
--- a/reactive/api/api/datasourcex-reactive-api.api
+++ b/reactive/datasource/api/api/datasourcex-reactive-api.api
@@ -78,6 +78,7 @@ public final class com/caplin/integration/datasourcex/reactive/api/BroadcastEven
public final fun component1 ()Ljava/lang/String;
public final fun component2 ()Ljava/lang/Object;
public fun equals (Ljava/lang/Object;)Z
+ public final fun getPath ()Ljava/lang/String;
public final fun getSubject ()Ljava/lang/String;
public final fun getValue ()Ljava/lang/Object;
public fun hashCode ()I
diff --git a/reactive/api/build.gradle.kts b/reactive/datasource/api/build.gradle.kts
similarity index 100%
rename from reactive/api/build.gradle.kts
rename to reactive/datasource/api/build.gradle.kts
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveConfig.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveConfig.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveConfig.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveConfig.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveContainerConfig.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveContainerConfig.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveContainerConfig.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveContainerConfig.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindContext.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindContext.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindContext.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindContext.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindMarker.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindMarker.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindMarker.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindMarker.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastConfig.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastConfig.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastConfig.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastConfig.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastEvent.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastEvent.kt
similarity index 58%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastEvent.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastEvent.kt
index a5e8853..2978ef3 100644
--- a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastEvent.kt
+++ b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastEvent.kt
@@ -1,7 +1,11 @@
package com.caplin.integration.datasourcex.reactive.api
-class BroadcastEvent(val subject: String, val value: T) {
- operator fun component1(): String = subject
+class BroadcastEvent(val path: String, val value: T) {
+ @Deprecated("Renamed to path", ReplaceWith("path"))
+ val subject: String
+ get() = path
+
+ operator fun component1(): String = path
operator fun component2(): T = value
@@ -11,19 +15,19 @@ class BroadcastEvent(val subject: String, val value: T) {
other as BroadcastEvent<*>
- if (subject != other.subject) return false
+ if (path != other.path) return false
if (value != other.value) return false
return true
}
override fun hashCode(): Int {
- var result = subject.hashCode()
+ var result = path.hashCode()
result = 31 * result + value.hashCode()
return result
}
override fun toString(): String {
- return "BroadcastEvent(subject='$subject', value=$value)"
+ return "BroadcastEvent(path='$path', value=$value)"
}
}
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelConfig.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelConfig.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelConfig.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelConfig.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequest.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequest.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequest.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequest.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequestSupplier.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequestSupplier.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequestSupplier.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequestSupplier.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelSupplier.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelSupplier.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelSupplier.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelSupplier.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelType.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelType.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelType.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelType.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ConfigBlock.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ConfigBlock.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ConfigBlock.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ConfigBlock.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ContainerEvent.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ContainerEvent.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ContainerEvent.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ContainerEvent.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/DataSourceSettings.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/DataSourceSettings.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/DataSourceSettings.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/DataSourceSettings.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/InsertAt.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/InsertAt.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/InsertAt.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/InsertAt.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathSupplier.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathSupplier.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathSupplier.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathSupplier.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesChannelSupplier.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesChannelSupplier.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesChannelSupplier.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesChannelSupplier.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesSupplier.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesSupplier.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesSupplier.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesSupplier.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RecordType.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RecordType.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RecordType.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RecordType.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/Request.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/Request.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/Request.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/Request.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RequestSupplier.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RequestSupplier.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RequestSupplier.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RequestSupplier.kt
diff --git a/reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ServiceConfig.kt b/reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ServiceConfig.kt
similarity index 100%
rename from reactive/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ServiceConfig.kt
rename to reactive/datasource/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ServiceConfig.kt
diff --git a/reactive/core/api/datasourcex-reactive-core.api b/reactive/datasource/core/api/datasourcex-reactive-core.api
similarity index 100%
rename from reactive/core/api/datasourcex-reactive-core.api
rename to reactive/datasource/core/api/datasourcex-reactive-core.api
diff --git a/reactive/core/build.gradle.kts b/reactive/datasource/core/build.gradle.kts
similarity index 100%
rename from reactive/core/build.gradle.kts
rename to reactive/datasource/core/build.gradle.kts
diff --git a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Binder.kt b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Binder.kt
similarity index 88%
rename from reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Binder.kt
rename to reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Binder.kt
index 2e825e2..49b5d2e 100644
--- a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Binder.kt
+++ b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Binder.kt
@@ -99,12 +99,12 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
objectMappings.filterValues { it in USERNAME_OBJECT_MAP_TOKENS }.keys
)
- /** Decomposes [subject] into a [Request] using this namespace's extraction rules. */
- private fun AntPatternNamespace.request(subject: String): Request =
+ /** Decomposes [path] into a [Request] using this namespace's extraction rules. */
+ private fun AntPatternNamespace.request(path: String): Request =
Request(
- extractPathParameters(subject),
- extractPathVariables(subject),
- extractQueryParameters(subject),
+ extractPathParameters(path),
+ extractPathVariables(path),
+ extractQueryParameters(path),
)
private data class ServiceInfo(
@@ -164,12 +164,12 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
}
subscriptions
- .computeIfAbsent(request.subject) { subject ->
+ .computeIfAbsent(request.subject) { path ->
val state =
- supplier(namespace.request(subject))
+ supplier(namespace.request(path))
.catch {
- subscriptions.remove(subject)
- publisher.publishNotFound(subject)
+ subscriptions.remove(path)
+ publisher.publishNotFound(path)
}
.stateIn(
dataSource,
@@ -203,8 +203,8 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
serviceInfo?.registerNamespace(namespace, config.objectMappings)
with(JsonContext()) {
- bindActiveSubjects(namespace, { supplier(namespace.request(it)) }) { subject, value ->
- createMessage(subject, value)
+ bindActiveSubjects(namespace, { supplier(namespace.request(it)) }) { path, value ->
+ createMessage(path, value)
}
}
}
@@ -220,8 +220,8 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
serviceInfo?.registerNamespace(namespace, config.objectMappings)
with(RecordContext(config.images, config.recordType)) {
- bindActiveSubjects(namespace, { supplier(namespace.request(it)) }) { subject, value ->
- createMessage(subject, value)
+ bindActiveSubjects(namespace, { supplier(namespace.request(it)) }) { path, value ->
+ createMessage(path, value)
}
}
}
@@ -238,8 +238,8 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
namespace,
config,
{ supplier(namespace.request(it)) },
- ) { subject, value ->
- createMessage(subject, value)
+ ) { path, value ->
+ createMessage(path, value)
}
}
}
@@ -256,8 +256,8 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
namespace,
config,
{ supplier(namespace.request(it)) },
- ) { subject, value ->
- createMessage(subject, value)
+ ) { path, value ->
+ createMessage(path, value)
}
}
}
@@ -482,10 +482,10 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
.onSuccess {
val message =
publisher.messageFactory.createMappingMessage(
- it.subject,
+ it.path,
it.value.path,
)
- cachedEvents?.put(it.subject, message)
+ cachedEvents?.put(it.path, message)
publisher.publishMappingMessage(message)
}
@@ -537,8 +537,8 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
upstreamChannel.onReceiveCatching { result ->
result
.onSuccess {
- val message = publisher.messageFactory.createMessage(it.subject, it.value)
- cachedEvents?.put(it.subject, message)
+ val message = publisher.messageFactory.createMessage(it.path, it.value)
+ cachedEvents?.put(it.path, message)
// Implementation of this appears to set the image flag and publish to all peers,
// so we don't
@@ -563,8 +563,8 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
private fun bindContainers(
containerSubjectPattern: AntPatternNamespace,
config: ActiveContainerConfig,
- createFlow: (containerSubject: String) -> Flow>,
- createMessage: CachedMessageFactory.(subject: String, value: T) -> Message,
+ createFlow: (containerPath: String) -> Flow>,
+ createMessage: CachedMessageFactory.(path: String, value: T) -> Message,
) {
val rowPattern =
AntPatternNamespace("${containerSubjectPattern.pattern}${config.rowPathSuffix}/{itemId}")
@@ -587,20 +587,16 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
publisher = cachingPublisher
}
- override fun onRequest(subject: String): Unit =
+ override fun onRequest(path: String): Unit =
with(publisher) {
- val containerSubject =
- subject
- .split("/")
- .dropLast(1)
- .joinToString("/")
- .removeSuffix(config.rowPathSuffix)
-
- val container = containers[containerSubject]
+ val containerPath =
+ path.split("/").dropLast(1).joinToString("/").removeSuffix(config.rowPathSuffix)
+
+ val container = containers[containerPath]
val job =
if (container == null) {
- logger.warn { "Container $containerSubject does not exist" }
- publishNotFound(subject)
+ logger.warn { "Container $containerPath does not exist" }
+ publishNotFound(path)
null
} else {
@@ -611,24 +607,24 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
// the record, as we assume
// that the removeElement is enough.
container
- .getRowFlow(subject)
- .map { value -> cachedMessageFactory.createMessage(subject, value) }
+ .getRowFlow(path)
+ .map { value -> cachedMessageFactory.createMessage(path, value) }
.onCompletion { throwable ->
if (throwable == null) {
- logger.warn { "Row $subject does not exist" }
- publishNotFound(subject)
+ logger.warn { "Row $path does not exist" }
+ publishNotFound(path)
}
}
.onEach { message -> publish(message) }
.launchIn(container.scope)
}
- job?.apply { invokeOnCompletion { subscriptions.remove(subject) } }
- ?.also { subscriptions[subject] = it }
+ job?.apply { invokeOnCompletion { subscriptions.remove(path) } }
+ ?.also { subscriptions[path] = it }
}
- override fun onDiscard(subject: String) {
- subscriptions.remove(subject)?.cancel()
+ override fun onDiscard(path: String) {
+ subscriptions.remove(path)?.cancel()
}
},
)
@@ -645,50 +641,49 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
publisher = cachingPublisher
}
- override fun onRequest(subject: String): Unit =
+ override fun onRequest(path: String): Unit =
with(publisher) {
dataSource
.launch {
val container =
Container(
- containerSubject = subject,
+ containerPath = path,
config = config,
supplier = createFlow,
scope = this,
)
- containers[subject] = container
+ containers[path] = container
try {
container.containerEventsFlow
.onCompletion { throwable ->
- if (throwable == null) publishNotFound(subject)
+ if (throwable == null) publishNotFound(path)
}
.catch { throwable ->
- publishNotFound(subject)
- logger.warn(throwable) { "Unhandled exception in $subject" }
+ publishNotFound(path)
+ logger.warn(throwable) { "Unhandled exception in $path" }
}
.collect { containerEvents ->
val containerMessage =
- cachedMessageFactory.createContainerMessage(subject).apply {
+ cachedMessageFactory.createContainerMessage(path).apply {
isDoNotAuthenticate = true
containerEvents.forEach { containerEvent ->
when (containerEvent) {
is InternalContainerEvent.Inserted -> {
logger.debug {
- "Inserted row ${containerEvent.subject} in $subject"
+ "Inserted row ${containerEvent.path} in $path"
}
when (config.insertAt) {
- InsertAt.HEAD ->
- insertElement(containerEvent.subject, 0)
+ InsertAt.HEAD -> insertElement(containerEvent.path, 0)
- InsertAt.TAIL -> addElement(containerEvent.subject)
+ InsertAt.TAIL -> addElement(containerEvent.path)
}
}
is InternalContainerEvent.Removed -> {
logger.debug {
- "Removed row ${containerEvent.subject} in $subject"
+ "Removed row ${containerEvent.path} in $path"
}
- removeElement(containerEvent.subject)
+ removeElement(containerEvent.path)
}
}
}
@@ -696,16 +691,16 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
publish(containerMessage)
}
} finally {
- containers.remove(subject)
+ containers.remove(path)
}
cancel()
}
- .apply { invokeOnCompletion { subscriptions.remove(subject) } }
- .also { job -> subscriptions[subject] = job }
+ .apply { invokeOnCompletion { subscriptions.remove(path) } }
+ .also { job -> subscriptions[path] = job }
}
- override fun onDiscard(subject: String) {
- subscriptions.remove(subject)?.cancel()
+ override fun onDiscard(path: String) {
+ subscriptions.remove(path)?.cancel()
}
},
)
@@ -713,8 +708,8 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
private fun bindActiveSubjects(
namespace: AntPatternNamespace,
- createFlow: (subject: String) -> Flow,
- createMessage: CachedMessageFactory.(subject: String, value: T) -> Message,
+ createFlow: (path: String) -> Flow,
+ createMessage: CachedMessageFactory.(path: String, value: T) -> Message,
) {
dataSource.createCachingPublisher(
namespace,
@@ -728,40 +723,40 @@ private constructor(val dataSource: ScopedDataSource, private val serviceInfo: S
publisher = cachingPublisher
}
- override fun onRequest(subject: String) {
- check(!subscriptions.containsKey(subject)) {
- "Multiple subscriptions to the same subject ($subject)"
+ override fun onRequest(path: String) {
+ check(!subscriptions.containsKey(path)) {
+ "Multiple subscriptions to the same subject ($path)"
}
- subscriptions[subject] =
- flow { emitAll(createFlow(subject)) }
- .map { t -> publisher.cachedMessageFactory.createMessage(subject, t) }
+ subscriptions[path] =
+ flow { emitAll(createFlow(path)) }
+ .map { t -> publisher.cachedMessageFactory.createMessage(path, t) }
.onEach { message -> publisher.publish(message) }
.onCompletion { throwable ->
- if (throwable == null) publisher.publishNotFound(subject)
+ if (throwable == null) publisher.publishNotFound(path)
}
.catch { throwable ->
- logger.warn(throwable) { "Unhandled exception in $subject" }
- publisher.publishNotFound(subject)
+ logger.warn(throwable) { "Unhandled exception in $path" }
+ publisher.publishNotFound(path)
}
.launchIn(dataSource)
- .apply { invokeOnCompletion { subscriptions.remove(subject) } }
+ .apply { invokeOnCompletion { subscriptions.remove(path) } }
}
- override fun onDiscard(subject: String) {
- subscriptions.remove(subject)?.cancel()
+ override fun onDiscard(path: String) {
+ subscriptions.remove(path)?.cancel()
}
},
)
}
- private fun CachingPublisher.publishNotFound(subject: String) {
+ private fun CachingPublisher.publishNotFound(path: String) {
publishSubjectErrorEvent(
- cachedMessageFactory.createSubjectErrorEvent(subject, SubjectError.NotFound),
+ cachedMessageFactory.createSubjectErrorEvent(path, SubjectError.NotFound),
)
}
- private fun Publisher.publishNotFound(subject: String) {
- publishSubjectErrorEvent(messageFactory.createSubjectErrorEvent(subject, SubjectError.NotFound))
+ private fun Publisher.publishNotFound(path: String) {
+ publishSubjectErrorEvent(messageFactory.createSubjectErrorEvent(path, SubjectError.NotFound))
}
private fun logBind(
diff --git a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Container.kt b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Container.kt
similarity index 83%
rename from reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Container.kt
rename to reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Container.kt
index 0e91443..c054deb 100644
--- a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Container.kt
+++ b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Container.kt
@@ -27,7 +27,7 @@ import kotlinx.coroutines.flow.takeWhile
import kotlinx.coroutines.time.withTimeoutOrNull
internal class Container(
- private val containerSubject: String,
+ private val containerPath: String,
private val config: ActiveContainerConfig,
private val supplier: (String) -> Flow>,
internal val scope: CoroutineScope,
@@ -47,26 +47,24 @@ internal class Container(
private val containerState: Flow> =
flow {
- logger.info { "Creating container $containerSubject" }
+ logger.info { "Creating container $containerPath" }
emit(ContainerState.Initialising)
var rows = persistentMapOf>()
- supplier(containerSubject)
+ supplier(containerPath)
.onStart { emit(ContainerEvent.Bulk(emptyList())) }
.bufferingDebounce(config.structureDebounce)
.collect { containerEvents ->
rows =
rows.mutate { mutableRows ->
fun handleEvent(event: ContainerEvent.RowEvent) {
- val rowSubject =
- "$containerSubject${config.rowPathSuffix}/${URLEncoder.encode(event.key, Charsets.UTF_8)}"
+ val rowPath =
+ "$containerPath${config.rowPathSuffix}/${URLEncoder.encode(event.key, Charsets.UTF_8)}"
when (event) {
- is ContainerEvent.RowEvent.Remove -> mutableRows.remove(rowSubject)
+ is ContainerEvent.RowEvent.Remove -> mutableRows.remove(rowPath)
is ContainerEvent.RowEvent.Upsert ->
- mutableRows[rowSubject]?.let { it.value = event.value }
- ?: run {
- mutableRows[rowSubject] = MutableStateFlow(event.value)
- }
+ mutableRows[rowPath]?.let { it.value = event.value }
+ ?: run { mutableRows[rowPath] = MutableStateFlow(event.value) }
}
}
@@ -85,14 +83,14 @@ internal class Container(
.distinctUntilChanged()
.onCompletion {
if (it == null) {
- logger.info { "Completing container $containerSubject" }
+ logger.info { "Completing container $containerPath" }
emit(ContainerState.Completed())
} else if (it is CancellationException) {
- logger.info { "Cancelling container $containerSubject" }
+ logger.info { "Cancelling container $containerPath" }
}
}
.catch { e ->
- logger.warn(e) { "Unhandled exception in container $containerSubject" }
+ logger.warn(e) { "Unhandled exception in container $containerPath" }
emit(ContainerState.Completed(e))
}
.shareIn(
@@ -122,12 +120,12 @@ internal class Container(
}
}
- fun getRowFlow(rowSubject: String): Flow = flow {
+ fun getRowFlow(rowPath: String): Flow = flow {
val rowFlow =
withTimeoutOrNull(config.rowRequestTimeout) {
containerState
.filterIsInstance>()
- .map { content -> content.records[rowSubject] }
+ .map { content -> content.records[rowPath] }
.filterNotNull()
.first()
}
diff --git a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/IFlowAdapter.kt b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/IFlowAdapter.kt
similarity index 100%
rename from reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/IFlowAdapter.kt
rename to reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/IFlowAdapter.kt
diff --git a/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt
new file mode 100644
index 0000000..4493f7c
--- /dev/null
+++ b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt
@@ -0,0 +1,9 @@
+package com.caplin.integration.datasourcex.reactive.core
+
+internal sealed interface InternalContainerEvent {
+ val path: String
+
+ data class Inserted(override val path: String) : InternalContainerEvent
+
+ data class Removed(override val path: String) : InternalContainerEvent
+}
diff --git a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/JsonContext.kt b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/JsonContext.kt
similarity index 62%
rename from reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/JsonContext.kt
rename to reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/JsonContext.kt
index c044370..f7256ce 100644
--- a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/JsonContext.kt
+++ b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/JsonContext.kt
@@ -5,6 +5,6 @@ import com.caplin.datasource.messaging.json.JsonMessage
internal class JsonContext {
- fun CachedMessageFactory.createMessage(subject: String, value: Any): JsonMessage =
- createJsonMessage(subject, value)
+ fun CachedMessageFactory.createMessage(path: String, value: Any): JsonMessage =
+ createJsonMessage(path, value)
}
diff --git a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/RecordContext.kt b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/RecordContext.kt
similarity index 70%
rename from reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/RecordContext.kt
rename to reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/RecordContext.kt
index 887fbbb..693e4e2 100644
--- a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/RecordContext.kt
+++ b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/RecordContext.kt
@@ -18,19 +18,19 @@ internal class RecordContext(
isImage = images
}
- internal fun CachedMessageFactory.createMessage(subject: String, fields: Map) =
+ internal fun CachedMessageFactory.createMessage(path: String, fields: Map) =
when (recordType) {
- RecordType.GENERIC -> createGenericMessage(subject)
- RecordType.TYPE1 -> createRecordType1Message(subject)
+ RecordType.GENERIC -> createGenericMessage(path)
+ RecordType.TYPE1 -> createRecordType1Message(path)
}.apply {
fields.forEach { (key, value) -> setField(key.toString(), value.toString()) }
isImage = images
}
- internal fun MessageFactory.createMessage(subject: String, fields: Map) =
+ internal fun MessageFactory.createMessage(path: String, fields: Map) =
when (recordType) {
- RecordType.GENERIC -> createGenericMessage(subject)
- RecordType.TYPE1 -> createRecordType1Message(subject)
+ RecordType.GENERIC -> createGenericMessage(path)
+ RecordType.TYPE1 -> createRecordType1Message(path)
}.apply {
fields.forEach { (key, value) -> setField(key.toString(), value.toString()) }
isImage = images
diff --git a/reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/ScopedDataSource.kt b/reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/ScopedDataSource.kt
similarity index 100%
rename from reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/ScopedDataSource.kt
rename to reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/ScopedDataSource.kt
diff --git a/reactive/core/src/test/kotlin/com/caplin/integration/datasourcex/reactive/core/BinderTest.kt b/reactive/datasource/core/src/test/kotlin/com/caplin/integration/datasourcex/reactive/core/BinderTest.kt
similarity index 100%
rename from reactive/core/src/test/kotlin/com/caplin/integration/datasourcex/reactive/core/BinderTest.kt
rename to reactive/datasource/core/src/test/kotlin/com/caplin/integration/datasourcex/reactive/core/BinderTest.kt
diff --git a/reactive/java-flow/README.md b/reactive/datasource/java-flow/README.md
similarity index 100%
rename from reactive/java-flow/README.md
rename to reactive/datasource/java-flow/README.md
diff --git a/reactive/java-flow/api/datasourcex-java-flow.api b/reactive/datasource/java-flow/api/datasourcex-java-flow.api
similarity index 100%
rename from reactive/java-flow/api/datasourcex-java-flow.api
rename to reactive/datasource/java-flow/api/datasourcex-java-flow.api
diff --git a/reactive/java-flow/build.gradle.kts b/reactive/datasource/java-flow/build.gradle.kts
similarity index 100%
rename from reactive/java-flow/build.gradle.kts
rename to reactive/datasource/java-flow/build.gradle.kts
diff --git a/reactive/java-flow/src/main/kotlin/com/caplin/integration/datasourcex/reactive/java/Bind.kt b/reactive/datasource/java-flow/src/main/kotlin/com/caplin/integration/datasourcex/reactive/java/Bind.kt
similarity index 100%
rename from reactive/java-flow/src/main/kotlin/com/caplin/integration/datasourcex/reactive/java/Bind.kt
rename to reactive/datasource/java-flow/src/main/kotlin/com/caplin/integration/datasourcex/reactive/java/Bind.kt
diff --git a/reactive/java-flow/src/samples/kotlin/samples/JsonObject.kt b/reactive/datasource/java-flow/src/samples/kotlin/samples/JsonObject.kt
similarity index 100%
rename from reactive/java-flow/src/samples/kotlin/samples/JsonObject.kt
rename to reactive/datasource/java-flow/src/samples/kotlin/samples/JsonObject.kt
diff --git a/reactive/java-flow/src/samples/kotlin/samples/Samples.kt b/reactive/datasource/java-flow/src/samples/kotlin/samples/Samples.kt
similarity index 100%
rename from reactive/java-flow/src/samples/kotlin/samples/Samples.kt
rename to reactive/datasource/java-flow/src/samples/kotlin/samples/Samples.kt
diff --git a/reactive/kotlin/README.md b/reactive/datasource/kotlin/README.md
similarity index 100%
rename from reactive/kotlin/README.md
rename to reactive/datasource/kotlin/README.md
diff --git a/reactive/kotlin/api/datasourcex-kotlin.api b/reactive/datasource/kotlin/api/datasourcex-kotlin.api
similarity index 100%
rename from reactive/kotlin/api/datasourcex-kotlin.api
rename to reactive/datasource/kotlin/api/datasourcex-kotlin.api
diff --git a/reactive/kotlin/build.gradle.kts b/reactive/datasource/kotlin/build.gradle.kts
similarity index 100%
rename from reactive/kotlin/build.gradle.kts
rename to reactive/datasource/kotlin/build.gradle.kts
diff --git a/reactive/kotlin/src/main/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/Bind.kt b/reactive/datasource/kotlin/src/main/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/Bind.kt
similarity index 100%
rename from reactive/kotlin/src/main/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/Bind.kt
rename to reactive/datasource/kotlin/src/main/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/Bind.kt
diff --git a/reactive/kotlin/src/samples/kotlin/samples/JsonObject.kt b/reactive/datasource/kotlin/src/samples/kotlin/samples/JsonObject.kt
similarity index 100%
rename from reactive/kotlin/src/samples/kotlin/samples/JsonObject.kt
rename to reactive/datasource/kotlin/src/samples/kotlin/samples/JsonObject.kt
diff --git a/reactive/kotlin/src/samples/kotlin/samples/Samples.kt b/reactive/datasource/kotlin/src/samples/kotlin/samples/Samples.kt
similarity index 100%
rename from reactive/kotlin/src/samples/kotlin/samples/Samples.kt
rename to reactive/datasource/kotlin/src/samples/kotlin/samples/Samples.kt
diff --git a/reactive/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/BindTest.kt b/reactive/datasource/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/BindTest.kt
similarity index 100%
rename from reactive/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/BindTest.kt
rename to reactive/datasource/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/BindTest.kt
diff --git a/reactive/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/LoadingList.kt b/reactive/datasource/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/LoadingList.kt
similarity index 100%
rename from reactive/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/LoadingList.kt
rename to reactive/datasource/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/LoadingList.kt
diff --git a/reactive/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/ProjectConfig.kt b/reactive/datasource/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/ProjectConfig.kt
similarity index 100%
rename from reactive/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/ProjectConfig.kt
rename to reactive/datasource/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/ProjectConfig.kt
diff --git a/reactive/reactivestreams/README.md b/reactive/datasource/reactivestreams/README.md
similarity index 100%
rename from reactive/reactivestreams/README.md
rename to reactive/datasource/reactivestreams/README.md
diff --git a/reactive/reactivestreams/api/datasourcex-reactivestreams.api b/reactive/datasource/reactivestreams/api/datasourcex-reactivestreams.api
similarity index 100%
rename from reactive/reactivestreams/api/datasourcex-reactivestreams.api
rename to reactive/datasource/reactivestreams/api/datasourcex-reactivestreams.api
diff --git a/reactive/reactivestreams/build.gradle.kts b/reactive/datasource/reactivestreams/build.gradle.kts
similarity index 100%
rename from reactive/reactivestreams/build.gradle.kts
rename to reactive/datasource/reactivestreams/build.gradle.kts
diff --git a/reactive/reactivestreams/src/main/kotlin/com/caplin/integration/datasourcex/reactive/reactivestreams/Bind.kt b/reactive/datasource/reactivestreams/src/main/kotlin/com/caplin/integration/datasourcex/reactive/reactivestreams/Bind.kt
similarity index 100%
rename from reactive/reactivestreams/src/main/kotlin/com/caplin/integration/datasourcex/reactive/reactivestreams/Bind.kt
rename to reactive/datasource/reactivestreams/src/main/kotlin/com/caplin/integration/datasourcex/reactive/reactivestreams/Bind.kt
diff --git a/reactive/reactivestreams/src/samples/kotlin/samples/JsonObject.kt b/reactive/datasource/reactivestreams/src/samples/kotlin/samples/JsonObject.kt
similarity index 100%
rename from reactive/reactivestreams/src/samples/kotlin/samples/JsonObject.kt
rename to reactive/datasource/reactivestreams/src/samples/kotlin/samples/JsonObject.kt
diff --git a/reactive/reactivestreams/src/samples/kotlin/samples/Samples.kt b/reactive/datasource/reactivestreams/src/samples/kotlin/samples/Samples.kt
similarity index 100%
rename from reactive/reactivestreams/src/samples/kotlin/samples/Samples.kt
rename to reactive/datasource/reactivestreams/src/samples/kotlin/samples/Samples.kt
diff --git a/reactive/streamlink/kotlin/api/sl4jx-kotlin.api b/reactive/streamlink/kotlin/api/sl4jx-kotlin.api
new file mode 100644
index 0000000..8b22942
--- /dev/null
+++ b/reactive/streamlink/kotlin/api/sl4jx-kotlin.api
@@ -0,0 +1,107 @@
+public abstract interface class com/caplin/integration/streamlinkx/ContainerChange {
+}
+
+public final class com/caplin/integration/streamlinkx/ContainerChange$Clear : com/caplin/integration/streamlinkx/ContainerChange {
+ public static final field INSTANCE Lcom/caplin/integration/streamlinkx/ContainerChange$Clear;
+ public fun equals (Ljava/lang/Object;)Z
+ public fun hashCode ()I
+ public fun toString ()Ljava/lang/String;
+}
+
+public abstract interface class com/caplin/integration/streamlinkx/ContainerChange$RowChange : com/caplin/integration/streamlinkx/ContainerChange {
+ public abstract fun getIndex ()I
+ public abstract fun getPath ()Ljava/lang/String;
+}
+
+public final class com/caplin/integration/streamlinkx/ContainerChange$RowChange$Added : com/caplin/integration/streamlinkx/ContainerChange$RowChange {
+ public fun (ILjava/lang/String;)V
+ public fun equals (Ljava/lang/Object;)Z
+ public fun getIndex ()I
+ public fun getPath ()Ljava/lang/String;
+ public fun hashCode ()I
+ public fun toString ()Ljava/lang/String;
+}
+
+public final class com/caplin/integration/streamlinkx/ContainerChange$RowChange$Removed : com/caplin/integration/streamlinkx/ContainerChange$RowChange {
+ public fun (ILjava/lang/String;)V
+ public fun equals (Ljava/lang/Object;)Z
+ public fun getIndex ()I
+ public fun getPath ()Ljava/lang/String;
+ public fun hashCode ()I
+ public fun toString ()Ljava/lang/String;
+}
+
+public final class com/caplin/integration/streamlinkx/ErrorEvent : com/caplin/integration/streamlinkx/Event {
+ public fun (Lcom/caplin/streamlink/SubscriptionErrorType;Lcom/caplin/streamlink/ErrorReason;)V
+ public fun equals (Ljava/lang/Object;)Z
+ public final fun getReason ()Lcom/caplin/streamlink/ErrorReason;
+ public final fun getType ()Lcom/caplin/streamlink/SubscriptionErrorType;
+ public fun hashCode ()I
+ public fun toString ()Ljava/lang/String;
+}
+
+public abstract interface class com/caplin/integration/streamlinkx/Event {
+}
+
+public final class com/caplin/integration/streamlinkx/StatusEvent : com/caplin/integration/streamlinkx/Event {
+ public fun (Lcom/caplin/streamlink/SubscriptionStatusType;Ljava/lang/String;Ljava/util/Map;)V
+ public fun equals (Ljava/lang/Object;)Z
+ public final fun getFields ()Ljava/util/Map;
+ public final fun getMessage ()Ljava/lang/String;
+ public final fun getType ()Lcom/caplin/streamlink/SubscriptionStatusType;
+ public fun hashCode ()I
+ public fun toString ()Ljava/lang/String;
+}
+
+public abstract interface class com/caplin/integration/streamlinkx/StreamLinkConnection : com/caplin/streamlink/StreamLink, java/lang/AutoCloseable {
+ public static final field Companion Lcom/caplin/integration/streamlinkx/StreamLinkConnection$Companion;
+ public fun awaitConnected (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public fun awaitServiceUp (Ljava/lang/String;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public abstract fun getChannel (Lcom/caplin/integration/datasourcex/util/Subject;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+ public abstract fun getChannel (Lcom/caplin/integration/datasourcex/util/Subject;Lkotlinx/coroutines/flow/Flow;Lkotlin/reflect/KClass;)Lkotlinx/coroutines/flow/Flow;
+ public abstract fun getContainer (Lcom/caplin/integration/datasourcex/util/Subject;)Lkotlinx/coroutines/flow/Flow;
+ public abstract fun getState ()Lkotlinx/coroutines/flow/SharedFlow;
+ public abstract fun getSubject (Lcom/caplin/integration/datasourcex/util/Subject;)Lkotlinx/coroutines/flow/Flow;
+ public abstract fun getSubject (Lcom/caplin/integration/datasourcex/util/Subject;Lkotlin/reflect/KClass;)Lkotlinx/coroutines/flow/Flow;
+ public abstract fun getUsername ()Ljava/lang/String;
+}
+
+public final class com/caplin/integration/streamlinkx/StreamLinkConnection$Companion {
+}
+
+public final class com/caplin/integration/streamlinkx/StreamLinkConnection$DefaultImpls {
+ public static fun awaitConnected (Lcom/caplin/integration/streamlinkx/StreamLinkConnection;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public static fun awaitServiceUp (Lcom/caplin/integration/streamlinkx/StreamLinkConnection;Ljava/lang/String;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
+public final class com/caplin/integration/streamlinkx/StreamLinkConnectionFactory {
+ public static final field Companion Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory$Companion;
+ public synthetic fun (Ljava/lang/String;Lcom/caplin/keymaster/IKeyMasterConfiguration;Ltools/jackson/databind/ObjectMapper;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
+ public final fun connect (Ljava/lang/String;)Lcom/caplin/integration/streamlinkx/StreamLinkConnection;
+}
+
+public final class com/caplin/integration/streamlinkx/StreamLinkConnectionFactory$Companion {
+ public final fun convertDerToPem ([BLjava/lang/String;)Ljava/lang/String;
+ public static synthetic fun convertDerToPem$default (Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory$Companion;[BLjava/lang/String;ILjava/lang/Object;)Ljava/lang/String;
+ public final fun invoke (Ljava/lang/String;Lcom/caplin/keymaster/IKeyMasterConfiguration;Ltools/jackson/databind/ObjectMapper;)Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;
+ public final fun invoke (Ljava/lang/String;Ljava/io/InputStream;Lcom/caplin/keymaster/KeyMasterHashingAlgorithm;Ltools/jackson/databind/ObjectMapper;)Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;
+ public final fun invoke (Ljava/lang/String;Ljava/security/PrivateKey;Lcom/caplin/keymaster/KeyMasterHashingAlgorithm;Ltools/jackson/databind/ObjectMapper;)Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;
+ public static synthetic fun invoke$default (Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory$Companion;Ljava/lang/String;Lcom/caplin/keymaster/IKeyMasterConfiguration;Ltools/jackson/databind/ObjectMapper;ILjava/lang/Object;)Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;
+ public static synthetic fun invoke$default (Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory$Companion;Ljava/lang/String;Ljava/io/InputStream;Lcom/caplin/keymaster/KeyMasterHashingAlgorithm;Ltools/jackson/databind/ObjectMapper;ILjava/lang/Object;)Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;
+ public static synthetic fun invoke$default (Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory$Companion;Ljava/lang/String;Ljava/security/PrivateKey;Lcom/caplin/keymaster/KeyMasterHashingAlgorithm;Ltools/jackson/databind/ObjectMapper;ILjava/lang/Object;)Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;
+}
+
+public final class com/caplin/integration/streamlinkx/TypesKt {
+ public static final fun awaitStatusOk (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+ public static final fun filterIsUpdate (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+ public static final fun runningFold (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+}
+
+public final class com/caplin/integration/streamlinkx/UpdateEvent : com/caplin/integration/streamlinkx/Event {
+ public fun (Ljava/lang/Object;)V
+ public fun equals (Ljava/lang/Object;)Z
+ public final fun getPayload ()Ljava/lang/Object;
+ public fun hashCode ()I
+ public fun toString ()Ljava/lang/String;
+}
+
diff --git a/reactive/streamlink/kotlin/build.gradle.kts b/reactive/streamlink/kotlin/build.gradle.kts
new file mode 100644
index 0000000..43f329b
--- /dev/null
+++ b/reactive/streamlink/kotlin/build.gradle.kts
@@ -0,0 +1,57 @@
+plugins { `common-library` }
+
+description = "Kotlin based Reactive StreamLink extensions"
+
+dependencies {
+ api(platform(libs.spring.boot.dependencies))
+ // Unifies on the shared util.Subject; transitively brings the DataSource server SDK onto the
+ // client classpath — an accepted trade-off (see spec #62).
+ api(project(":datasourcex-util"))
+ api(libs.streamlink) { exclude(group = "junit") }
+ api(libs.keymaster)
+ api("org.jetbrains.kotlinx:kotlinx-coroutines-core")
+ // Jackson 3 versions come from the Spring Boot BOM.
+ api("tools.jackson.core:jackson-databind")
+ api(libs.kotlin.collections.immutable)
+ implementation("tools.jackson.module:jackson-module-kotlin")
+ implementation(libs.zjsonpatch)
+ implementation(libs.kotlin.logging)
+
+ testRuntimeOnly("org.slf4j:slf4j-simple")
+ testImplementation(libs.kotest.runner)
+ testImplementation(libs.kotest.assertions)
+ testImplementation(libs.mockk)
+ testImplementation(libs.turbine)
+}
+
+// TODO This can be removed once PSL-889 is resolved, and we take a new StreamLink.
+publishing {
+ publications {
+ withType().configureEach {
+ pom {
+ withXml {
+ val root = asNode()
+ val dependenciesNode =
+ root.children().find { it is groovy.util.Node && it.name() == "dependencies" }
+ as? groovy.util.Node ?: return@withXml
+
+ dependenciesNode
+ .children()
+ .filterIsInstance()
+ .filter { it.name() == "dependency" }
+ .filter {
+ val groupId = it.get("groupId")?.toString()
+ val artifactId = it.get("artifactId")?.toString()
+ groupId == "com.caplin.streamlink" && artifactId == "StreamLinkJava"
+ }
+ .forEach { depNode ->
+ val exclusions = depNode.appendNode("exclusions")
+ val exclusion = exclusions.appendNode("exclusion")
+ exclusion.appendNode("groupId", "junit")
+ exclusion.appendNode("artifactId", "junit")
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/DefaultSubscriptionListener.kt b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/DefaultSubscriptionListener.kt
new file mode 100644
index 0000000..7904dd6
--- /dev/null
+++ b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/DefaultSubscriptionListener.kt
@@ -0,0 +1,88 @@
+package com.caplin.integration.streamlinkx
+
+import com.caplin.streamlink.BinaryEvent
+import com.caplin.streamlink.ChatEvent
+import com.caplin.streamlink.ContainerEvent
+import com.caplin.streamlink.DirectoryEvent
+import com.caplin.streamlink.JsonEvent
+import com.caplin.streamlink.NewsEvent
+import com.caplin.streamlink.PageEvent
+import com.caplin.streamlink.PermissionEvent
+import com.caplin.streamlink.RecordType1Event
+import com.caplin.streamlink.RecordType2Event
+import com.caplin.streamlink.RecordType3Event
+import com.caplin.streamlink.StoryEvent
+import com.caplin.streamlink.Subscription
+import com.caplin.streamlink.SubscriptionErrorEvent
+import com.caplin.streamlink.SubscriptionListener
+import com.caplin.streamlink.SubscriptionStatusEvent
+import kotlinx.collections.immutable.toPersistentMap
+import kotlinx.coroutines.channels.SendChannel
+import kotlinx.coroutines.channels.trySendBlocking
+
+internal class DefaultSubscriptionListener(private val channel: SendChannel>) :
+ SubscriptionListener {
+
+ override fun onRecordUpdate(subscription: Subscription, event: RecordType1Event) {
+ //
+ }
+
+ override fun onRecordType2Update(subscription: Subscription, event: RecordType2Event) {
+ //
+ }
+
+ override fun onRecordType3Update(subscription: Subscription, event: RecordType3Event) {
+ //
+ }
+
+ override fun onPermissionUpdate(subscription: Subscription, event: PermissionEvent) {
+ //
+ }
+
+ override fun onNewsUpdate(subscription: Subscription, event: NewsEvent) {
+ //
+ }
+
+ override fun onStoryUpdate(subscription: Subscription, event: StoryEvent) {
+ //
+ }
+
+ override fun onPageUpdate(subscription: Subscription, event: PageEvent) {
+ //
+ }
+
+ override fun onChatUpdate(subscription: Subscription, event: ChatEvent) {
+ //
+ }
+
+ override fun onDirectoryUpdate(subscription: Subscription, event: DirectoryEvent) {
+ //
+ }
+
+ override fun onContainerUpdate(subscription: Subscription, event: ContainerEvent) {
+ //
+ }
+
+ override fun onJsonUpdate(subscription: Subscription, event: JsonEvent) {
+ //
+ }
+
+ override fun onBinaryUpdate(p0: Subscription?, p1: BinaryEvent?) {
+ //
+ }
+
+ override fun onSubscriptionError(subscription: Subscription, event: SubscriptionErrorEvent) {
+ channel.trySendBlocking(ErrorEvent(event.error, event.errorReason))
+ channel.close()
+ }
+
+ override fun onSubscriptionStatus(subscription: Subscription, event: SubscriptionStatusEvent) {
+ channel.trySendBlocking(
+ StatusEvent(
+ event.status,
+ event.statusMessage.orEmpty(),
+ event.fields.toPersistentMap(),
+ ),
+ )
+ }
+}
diff --git a/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnection.kt b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnection.kt
new file mode 100644
index 0000000..66ea98b
--- /dev/null
+++ b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnection.kt
@@ -0,0 +1,93 @@
+package com.caplin.integration.streamlinkx
+
+import com.caplin.integration.datasourcex.util.Subject
+import com.caplin.streamlink.ConnectionState
+import com.caplin.streamlink.ConnectionStatusEvent
+import com.caplin.streamlink.ServiceStatus
+import com.caplin.streamlink.StreamLink
+import kotlin.reflect.KClass
+import kotlin.time.Duration.Companion.milliseconds
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.first
+
+/**
+ * A coroutine- and [Flow]-friendly wrapper around a Caplin [StreamLink] connection.
+ *
+ * Extends the raw [StreamLink] API with suspending helpers for awaiting connection and service
+ * state, and with `getSubject`/`getChannel` overloads that expose subscriptions and channels as
+ * cold [Flow]s of [Event]s. Instances are created via [StreamLinkConnectionFactory.connect].
+ *
+ * Implements [AutoCloseable]; closing disconnects the underlying StreamLink.
+ */
+interface StreamLinkConnection : StreamLink, AutoCloseable {
+
+ companion object {
+ private val SERVICE_UP_POLL_INTERVAL = 100.milliseconds
+
+ /**
+ * Opens a bidirectional channel on [subject], sending items from [send] and receiving updates
+ * deserialised to [R]. Reified convenience for [getChannel] taking an explicit [KClass].
+ */
+ inline fun StreamLinkConnection.getChannel(
+ subject: Subject,
+ send: Flow,
+ ): Flow> = getChannel(subject, send, R::class)
+
+ /**
+ * Subscribes to [subject], deserialising each JSON update to [T]. Reified convenience for
+ * [getSubject] taking an explicit [KClass].
+ */
+ inline fun StreamLinkConnection.getSubject(subject: Subject): Flow> =
+ getSubject(subject, T::class)
+ }
+
+ /** The username this connection is authenticated as. */
+ val username: String
+
+ /** A hot [SharedFlow] of connection status changes, replaying the latest event to collectors. */
+ val state: SharedFlow
+
+ /** Suspends until the connection reaches the [ConnectionState.LOGGEDIN] state. */
+ suspend fun awaitConnected() {
+ state.first { it.connectionState == ConnectionState.LOGGEDIN }
+ }
+
+ /** Suspends until the service named [name] reports [ServiceStatus.OK], polling periodically. */
+ suspend fun awaitServiceUp(name: String) {
+ while (true) {
+ if (
+ connectionCurrentState.serviceStatusArray
+ .firstOrNull { it.serviceName == name }
+ ?.serviceStatus == ServiceStatus.OK
+ )
+ return
+ delay(SERVICE_UP_POLL_INTERVAL)
+ }
+ }
+
+ /**
+ * Subscribes to the container at [subject], emitting a [ContainerChangeEvent] for each row add,
+ * remove or clear as its membership changes.
+ */
+ fun getContainer(subject: Subject): Flow
+
+ /**
+ * Opens a record channel on [subject], sending each emitted field map from [send] to the server
+ * and emitting the server's record updates as [RecordEvent]s.
+ */
+ fun getChannel(subject: Subject, send: Flow