From 92b367ab3c07209c8768d6d0c7aeef806a9b74b8 Mon Sep 17 00:00:00 2001 From: Ross Anderson Date: Tue, 14 Jul 2026 15:14:23 +0100 Subject: [PATCH 01/10] Regroup datasourcex reactive modules under reactive/datasource/* MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the five reactive modules (core, api, kotlin, java-flow, reactivestreams) physically into reactive/datasource/ so the StreamLink SDK family can sit beside them under reactive/ later. A projectDir override in settings.gradle.kts relocates only the directory: the include coordinates, project paths, published Maven coordinates, and .api snapshots are all unchanged. Pure structural prefactor — no public API or behaviour change. Repoint the CI test-report globs and AGENTS.md module-layout prose at the new locations. Closes #66 --- .github/workflows/build.yml | 10 +++---- AGENTS.md | 26 +++++++++---------- reactive/{ => datasource}/api/README.md | 0 .../api/api/datasourcex-reactive-api.api | 0 .../{ => datasource}/api/build.gradle.kts | 0 .../datasourcex/reactive/api/ActiveConfig.kt | 0 .../reactive/api/ActiveContainerConfig.kt | 0 .../datasourcex/reactive/api/BindContext.kt | 0 .../datasourcex/reactive/api/BindMarker.kt | 0 .../reactive/api/BroadcastConfig.kt | 0 .../reactive/api/BroadcastEvent.kt | 0 .../datasourcex/reactive/api/ChannelConfig.kt | 0 .../reactive/api/ChannelRequest.kt | 0 .../reactive/api/ChannelRequestSupplier.kt | 0 .../reactive/api/ChannelSupplier.kt | 0 .../datasourcex/reactive/api/ChannelType.kt | 0 .../datasourcex/reactive/api/ConfigBlock.kt | 0 .../reactive/api/ContainerEvent.kt | 0 .../reactive/api/DataSourceSettings.kt | 0 .../datasourcex/reactive/api/InsertAt.kt | 0 .../datasourcex/reactive/api/PathSupplier.kt | 0 .../api/PathVariablesChannelSupplier.kt | 0 .../reactive/api/PathVariablesSupplier.kt | 0 .../datasourcex/reactive/api/RecordType.kt | 0 .../datasourcex/reactive/api/Request.kt | 0 .../reactive/api/RequestSupplier.kt | 0 .../datasourcex/reactive/api/ServiceConfig.kt | 0 .../core/api/datasourcex-reactive-core.api | 0 .../{ => datasource}/core/build.gradle.kts | 0 .../datasourcex/reactive/core/Binder.kt | 0 .../datasourcex/reactive/core/Container.kt | 0 .../datasourcex/reactive/core/IFlowAdapter.kt | 0 .../reactive/core/InternalContainerEvent.kt | 0 .../datasourcex/reactive/core/JsonContext.kt | 0 .../reactive/core/RecordContext.kt | 0 .../reactive/core/ScopedDataSource.kt | 0 .../datasourcex/reactive/core/BinderTest.kt | 0 reactive/{ => datasource}/java-flow/README.md | 0 .../java-flow/api/datasourcex-java-flow.api | 0 .../java-flow/build.gradle.kts | 0 .../datasourcex/reactive/java/Bind.kt | 0 .../src/samples/kotlin/samples/JsonObject.kt | 0 .../src/samples/kotlin/samples/Samples.kt | 0 reactive/{ => datasource}/kotlin/README.md | 0 .../kotlin/api/datasourcex-kotlin.api | 0 .../{ => datasource}/kotlin/build.gradle.kts | 0 .../datasourcex/reactive/kotlin/Bind.kt | 0 .../src/samples/kotlin/samples/JsonObject.kt | 0 .../src/samples/kotlin/samples/Samples.kt | 0 .../datasourcex/reactive/kotlin/BindTest.kt | 0 .../reactive/kotlin/LoadingList.kt | 0 .../reactive/kotlin/ProjectConfig.kt | 0 .../reactivestreams/README.md | 0 .../api/datasourcex-reactivestreams.api | 0 .../reactivestreams/build.gradle.kts | 0 .../reactive/reactivestreams/Bind.kt | 0 .../src/samples/kotlin/samples/JsonObject.kt | 0 .../src/samples/kotlin/samples/Samples.kt | 0 settings.gradle.kts | 10 +++++++ 59 files changed, 28 insertions(+), 18 deletions(-) rename reactive/{ => datasource}/api/README.md (100%) rename reactive/{ => datasource}/api/api/datasourcex-reactive-api.api (100%) rename reactive/{ => datasource}/api/build.gradle.kts (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveConfig.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ActiveContainerConfig.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindContext.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BindMarker.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastConfig.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/BroadcastEvent.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelConfig.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequest.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelRequestSupplier.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelSupplier.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ChannelType.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ConfigBlock.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ContainerEvent.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/DataSourceSettings.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/InsertAt.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathSupplier.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesChannelSupplier.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/PathVariablesSupplier.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RecordType.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/Request.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/RequestSupplier.kt (100%) rename reactive/{ => datasource}/api/src/main/kotlin/com/caplin/integration/datasourcex/reactive/api/ServiceConfig.kt (100%) rename reactive/{ => datasource}/core/api/datasourcex-reactive-core.api (100%) rename reactive/{ => datasource}/core/build.gradle.kts (100%) rename reactive/{ => datasource}/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Binder.kt (100%) rename reactive/{ => datasource}/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/Container.kt (100%) rename reactive/{ => datasource}/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/IFlowAdapter.kt (100%) rename reactive/{ => datasource}/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt (100%) rename reactive/{ => datasource}/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/JsonContext.kt (100%) rename reactive/{ => datasource}/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/RecordContext.kt (100%) rename reactive/{ => datasource}/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/ScopedDataSource.kt (100%) rename reactive/{ => datasource}/core/src/test/kotlin/com/caplin/integration/datasourcex/reactive/core/BinderTest.kt (100%) rename reactive/{ => datasource}/java-flow/README.md (100%) rename reactive/{ => datasource}/java-flow/api/datasourcex-java-flow.api (100%) rename reactive/{ => datasource}/java-flow/build.gradle.kts (100%) rename reactive/{ => datasource}/java-flow/src/main/kotlin/com/caplin/integration/datasourcex/reactive/java/Bind.kt (100%) rename reactive/{ => datasource}/java-flow/src/samples/kotlin/samples/JsonObject.kt (100%) rename reactive/{ => datasource}/java-flow/src/samples/kotlin/samples/Samples.kt (100%) rename reactive/{ => datasource}/kotlin/README.md (100%) rename reactive/{ => datasource}/kotlin/api/datasourcex-kotlin.api (100%) rename reactive/{ => datasource}/kotlin/build.gradle.kts (100%) rename reactive/{ => datasource}/kotlin/src/main/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/Bind.kt (100%) rename reactive/{ => datasource}/kotlin/src/samples/kotlin/samples/JsonObject.kt (100%) rename reactive/{ => datasource}/kotlin/src/samples/kotlin/samples/Samples.kt (100%) rename reactive/{ => datasource}/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/BindTest.kt (100%) rename reactive/{ => datasource}/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/LoadingList.kt (100%) rename reactive/{ => datasource}/kotlin/src/test/kotlin/com/caplin/integration/datasourcex/reactive/kotlin/ProjectConfig.kt (100%) rename reactive/{ => datasource}/reactivestreams/README.md (100%) rename reactive/{ => datasource}/reactivestreams/api/datasourcex-reactivestreams.api (100%) rename reactive/{ => datasource}/reactivestreams/build.gradle.kts (100%) rename reactive/{ => datasource}/reactivestreams/src/main/kotlin/com/caplin/integration/datasourcex/reactive/reactivestreams/Bind.kt (100%) rename reactive/{ => datasource}/reactivestreams/src/samples/kotlin/samples/JsonObject.kt (100%) rename reactive/{ => datasource}/reactivestreams/src/samples/kotlin/samples/Samples.kt (100%) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index b00b67f..a9c981d 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -68,11 +68,11 @@ 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 spring/build/test-results/test/TEST-*.xml util/build/test-results/test/TEST-*.xml diff --git a/AGENTS.md b/AGENTS.md index 04ff343..132fd5b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,36 +35,36 @@ 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 + +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. 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 reactive modules sit under `reactive/datasource/` on disk so the StreamLink SDK family can sit beside them under `reactive/` later. 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`, etc. — set by the `.name` overrides), and the published Maven coordinates are all unchanged. Published coordinates use the renamed name (`datasourcex-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/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 100% rename from reactive/api/api/datasourcex-reactive-api.api rename to reactive/datasource/api/api/datasourcex-reactive-api.api 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 100% 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 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 100% 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 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 100% 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 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/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 similarity index 100% rename from reactive/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt rename to reactive/datasource/core/src/main/kotlin/com/caplin/integration/datasourcex/reactive/core/InternalContainerEvent.kt 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 100% 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 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 100% 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 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/settings.gradle.kts b/settings.gradle.kts index 6d67afe..13a7eaf 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -31,22 +31,32 @@ project(":util").name = "datasourcex-util" include("reactive:core") +project(":reactive:core").projectDir = file("reactive/datasource/core") + project(":reactive:core").name = "datasourcex-reactive-core" include("reactive:api") +project(":reactive:api").projectDir = file("reactive/datasource/api") + project(":reactive:api").name = "datasourcex-reactive-api" include("reactive:java-flow") +project(":reactive:java-flow").projectDir = file("reactive/datasource/java-flow") + project(":reactive:java-flow").name = "datasourcex-java-flow" include("reactive:kotlin") +project(":reactive:kotlin").projectDir = file("reactive/datasource/kotlin") + project(":reactive:kotlin").name = "datasourcex-kotlin" include("reactive:reactivestreams") +project(":reactive:reactivestreams").projectDir = file("reactive/datasource/reactivestreams") + project(":reactive:reactivestreams").name = "datasourcex-reactivestreams" include("spring") From 97af6c4fc034bfa71c5a4aa3d6488cf44a3ddcbf Mon Sep 17 00:00:00 2001 From: Ross Anderson Date: Tue, 14 Jul 2026 15:39:43 +0100 Subject: [PATCH 02/10] Apply path/subject naming convention to datasourcex; add ADR 0002 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A variable/parameter/property holding a subject's String form is named `path`; one typed `Subject` stays `subject` (ADR 0002). Applied across the public and internal datasourcex surface: - AntPatternNamespace: match/extract* String params → path. - reactive.core (Binder, Container, InternalContainerEvent, JsonContext, RecordContext): declared String-subject params/vals/properties → path / containerPath / rowPath. Caplin SDK property accesses (RequestEvent/DiscardEvent/Channel/JsonChannel.subject) and SDK method names are left untouched. - spring.internal DataSourceServerBootstrap: createFlow String params → path. - BroadcastEvent: `path` is now the property; `subject` retained as a @Deprecated alias delegating to it, so the only public .api change is the additive getPath() (ships in the 3.x minor, no ABI removal). The buildSrc code generator already emits `path: String`, unchanged. check + apiCheck green. Note: the BroadcastEvent constructor parameter is also renamed to `path`. This is ABI-safe (parameter names are not part of the .api) but is a source break for Kotlin callers constructing it with the `subject = ` named argument. Part of #65. Closes #67 --- .../0002-subject-path-naming-convention.md | 16 ++ .../api/api/datasourcex-reactive-api.api | 1 + .../reactive/api/BroadcastEvent.kt | 14 +- .../datasourcex/reactive/core/Binder.kt | 151 +++++++++--------- .../datasourcex/reactive/core/Container.kt | 28 ++-- .../reactive/core/InternalContainerEvent.kt | 6 +- .../datasourcex/reactive/core/JsonContext.kt | 4 +- .../reactive/core/RecordContext.kt | 12 +- .../internal/DataSourceServerBootstrap.kt | 16 +- .../datasourcex/util/AntPatternNamespace.kt | 24 ++- 10 files changed, 142 insertions(+), 130 deletions(-) create mode 100644 docs/adr/0002-subject-path-naming-convention.md 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/reactive/datasource/api/api/datasourcex-reactive-api.api b/reactive/datasource/api/api/datasourcex-reactive-api.api index 6d5a522..532b9a5 100644 --- a/reactive/datasource/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/datasource/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 index a5e8853..2978ef3 100644 --- a/reactive/datasource/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/datasource/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 index 2e825e2..49b5d2e 100644 --- a/reactive/datasource/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/datasource/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 index 0e91443..c054deb 100644 --- a/reactive/datasource/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/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 index 0a17c80..4493f7c 100644 --- 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 @@ -1,9 +1,9 @@ package com.caplin.integration.datasourcex.reactive.core internal sealed interface InternalContainerEvent { - val subject: String + val path: String - data class Inserted(override val subject: String) : InternalContainerEvent + data class Inserted(override val path: String) : InternalContainerEvent - data class Removed(override val subject: String) : InternalContainerEvent + data class Removed(override val path: String) : InternalContainerEvent } diff --git a/reactive/datasource/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 index c044370..f7256ce 100644 --- a/reactive/datasource/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/datasource/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 index 887fbbb..693e4e2 100644 --- a/reactive/datasource/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/spring/src/main/kotlin/com/caplin/integration/datasourcex/spring/internal/DataSourceServerBootstrap.kt b/spring/src/main/kotlin/com/caplin/integration/datasourcex/spring/internal/DataSourceServerBootstrap.kt index 048891a..f13d361 100644 --- a/spring/src/main/kotlin/com/caplin/integration/datasourcex/spring/internal/DataSourceServerBootstrap.kt +++ b/spring/src/main/kotlin/com/caplin/integration/datasourcex/spring/internal/DataSourceServerBootstrap.kt @@ -162,10 +162,10 @@ internal class DataSourceServerBootstrap( when (dataSourceRequestType) { is RequestType.Channel -> { - fun createFlow(subject: String, receiveFlow: Flow): Flow = + fun createFlow(path: String, receiveFlow: Flow): Flow = flow { val sendFlow = AtomicReference>() - val route = routeMatcher.parseRoute(subject) + val route = routeMatcher.parseRoute(path) handleMessage( MessageBuilder.createMessage( receiveFlow, @@ -180,12 +180,12 @@ internal class DataSourceServerBootstrap( } .onStart { logger.info { - "$subject matching ${destinationMessageCondition.patterns} started" + "$path matching ${destinationMessageCondition.patterns} started" } } .onCompletion { logger.info { - "$subject matching ${destinationMessageCondition.patterns} completed" + "$path matching ${destinationMessageCondition.patterns} completed" } } @@ -235,10 +235,10 @@ internal class DataSourceServerBootstrap( } is RequestType.Stream -> { - fun createFlow(subject: String) = + fun createFlow(path: String) = flow { val sendFlow = AtomicReference>() - val route = routeMatcher.parseRoute(subject) + val route = routeMatcher.parseRoute(path) handleMessage( MessageBuilder.createMessage( emptyPayload, @@ -252,12 +252,12 @@ internal class DataSourceServerBootstrap( } .onStart { logger.info { - "$subject matching ${destinationMessageCondition.patterns} started" + "$path matching ${destinationMessageCondition.patterns} started" } } .onCompletion { logger.info { - "$subject matching ${destinationMessageCondition.patterns} completed" + "$path matching ${destinationMessageCondition.patterns} completed" } } diff --git a/util/src/main/kotlin/com/caplin/integration/datasourcex/util/AntPatternNamespace.kt b/util/src/main/kotlin/com/caplin/integration/datasourcex/util/AntPatternNamespace.kt index 3235cae..3dd233d 100644 --- a/util/src/main/kotlin/com/caplin/integration/datasourcex/util/AntPatternNamespace.kt +++ b/util/src/main/kotlin/com/caplin/integration/datasourcex/util/AntPatternNamespace.kt @@ -103,23 +103,21 @@ constructor(pattern: String, val rawPathVariables: Set = emptySet()) : N private val matcher = AntRegexPathMatcher(pattern) - override fun match(subject: String): Boolean = - matcher.regex.matchEntire(subject.pathPortion) != null + override fun match(path: String): Boolean = matcher.regex.matchEntire(path.pathPortion) != null /** * Extracts path variables from a matching subject. Values are URL-decoded, except for any listed * in [rawPathVariables] which are returned verbatim. Any trailing `?query` is ignored, so a * subject that carries query parameters yields the same path variables as the query-less subject. * - * @param subject The subject to extract variables from. Its path portion must match the - * [pattern]. + * @param path The subject to extract variables from. Its path portion must match the [pattern]. * @return A map of path variable names to their extracted values, in pattern order. * @throws IllegalStateException If the subject does not match the pattern. */ - fun extractPathVariables(subject: String): LinkedHashMap { + fun extractPathVariables(path: String): LinkedHashMap { val groups = - checkNotNull(matcher.regex.matchEntire(subject.pathPortion)) { - "Subject $subject does not match pattern $pattern" + checkNotNull(matcher.regex.matchEntire(path.pathPortion)) { + "Subject $path does not match pattern $pattern" } .groups @@ -136,21 +134,21 @@ constructor(pattern: String, val rawPathVariables: Set = emptySet()) : N * Splits the path portion of a subject into its URL-decoded segments, in order. Any trailing * `?query` is ignored. * - * @param subject The subject to extract path parameters from. + * @param path The subject to extract path parameters from. * @return The subject's path parts, decoded. */ - fun extractPathParameters(subject: String): List = - subject.pathPortion.split('/').filter(String::isNotEmpty).map(::urlDecode) + fun extractPathParameters(path: String): List = + path.pathPortion.split('/').filter(String::isNotEmpty).map(::urlDecode) /** * Extracts the query parameters from the optional trailing `?a=b&c=d` portion of a subject. * - * @param subject The subject to extract query parameters from. + * @param path The subject to extract query parameters from. * @return A map of query parameter names to values, or an empty map when the subject has no * query. */ - fun extractQueryParameters(subject: String): SortedMap = - parseQueryString(subject.substringAfter('?', "")) + fun extractQueryParameters(path: String): SortedMap = + parseQueryString(path.substringAfter('?', "")) /** Returns a copy of this namespace with [rawPathVariables] replaced. */ fun copy(rawPathVariables: Set = this.rawPathVariables): AntPatternNamespace = From 9adf3dcbe2cf34fd066610b16a0a0c2a88cb0e52 Mon Sep 17 00:00:00 2001 From: Ross Anderson Date: Tue, 14 Jul 2026 16:01:53 +0100 Subject: [PATCH 03/10] Import sl4jx as reactive/streamlink/kotlin (sl4jx-kotlin), unify on util.Subject Bring the StreamLink client extensions in as a new published module, sl4jx-kotlin, at reactive/streamlink/kotlin. Imported as a plain copy on this repo's toolchain (common-library convention, Kotlin 2.2.21, Kotest 6.2.2, Jackson 3 / slf4j from the Spring Boot BOM). - Add exact-pinned streamlink/keymaster and kotlin-logging catalog entries. - Delete streamlinkx.Subject and retarget every reference to the shared com.caplin.integration.datasourcex.util.Subject; query parameters now render sorted. - Carry the PSL-889 publish-time POM hack (strip transitive junit:junit from StreamLinkJava) verbatim. - Register the module in settings, datasourcex-version-catalog, api-docs (Dokka) and root Kover; commit its .api baseline. - Correct the stale AGENTS.md claim that datasourcex-util has no DataSource SDK dependency. Closes #68 --- AGENTS.md | 8 +- api-docs/build.gradle.kts | 1 + build.gradle.kts | 1 + gradle/libs.versions.toml | 10 + .../streamlink/kotlin/api/sl4jx-kotlin.api | 115 ++++ reactive/streamlink/kotlin/build.gradle.kts | 57 ++ .../DefaultSubscriptionListener.kt | 88 +++ .../streamlinkx/StreamLinkConnection.kt | 93 ++++ .../StreamLinkConnectionFactory.kt | 506 ++++++++++++++++++ .../caplin/integration/streamlinkx/Types.kt | 167 ++++++ .../StreamLinkConnectionFactoryTest.kt | 31 ++ settings.gradle.kts | 6 + version-catalog/build.gradle.kts | 6 + 13 files changed, 1087 insertions(+), 2 deletions(-) create mode 100644 reactive/streamlink/kotlin/api/sl4jx-kotlin.api create mode 100644 reactive/streamlink/kotlin/build.gradle.kts create mode 100644 reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/DefaultSubscriptionListener.kt create mode 100644 reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnection.kt create mode 100644 reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt create mode 100644 reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/Types.kt create mode 100644 reactive/streamlink/kotlin/src/test/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactoryTest.kt diff --git a/AGENTS.md b/AGENTS.md index 132fd5b..944f94f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,14 +42,18 @@ reactive/datasource/kotlin ← Bind DSL for Flow } each 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. + 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 five reactive modules sit under `reactive/datasource/` on disk so the StreamLink SDK family can sit beside them under `reactive/` later. 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`, etc. — set by the `.name` overrides), and the published Maven coordinates are all unchanged. Published coordinates use the renamed name (`datasourcex-kotlin`, `spring-boot-starter-datasource`, etc.), not the include path. +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) diff --git a/api-docs/build.gradle.kts b/api-docs/build.gradle.kts index f2e4bc3..2163bc9 100644 --- a/api-docs/build.gradle.kts +++ b/api-docs/build.gradle.kts @@ -7,6 +7,7 @@ dependencies { dokka(project(":reactive:datasourcex-java-flow")) dokka(project(":reactive:datasourcex-kotlin")) dokka(project(":reactive:datasourcex-reactivestreams")) + dokka(project(":reactive:sl4jx-kotlin")) 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/gradle/libs.versions.toml b/gradle/libs.versions.toml index 65ee58a..21e38d9 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -11,6 +11,11 @@ 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" @@ -39,6 +44,11 @@ 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" } diff --git a/reactive/streamlink/kotlin/api/sl4jx-kotlin.api b/reactive/streamlink/kotlin/api/sl4jx-kotlin.api new file mode 100644 index 0000000..3d8a31d --- /dev/null +++ b/reactive/streamlink/kotlin/api/sl4jx-kotlin.api @@ -0,0 +1,115 @@ +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 final fun component1 ()I + public final fun component2 ()Ljava/lang/String; + public final fun copy (ILjava/lang/String;)Lcom/caplin/integration/streamlinkx/ContainerChange$RowChange$Added; + public static synthetic fun copy$default (Lcom/caplin/integration/streamlinkx/ContainerChange$RowChange$Added;ILjava/lang/String;ILjava/lang/Object;)Lcom/caplin/integration/streamlinkx/ContainerChange$RowChange$Added; + 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 final fun component1 ()I + public final fun component2 ()Ljava/lang/String; + public final fun copy (ILjava/lang/String;)Lcom/caplin/integration/streamlinkx/ContainerChange$RowChange$Removed; + public static synthetic fun copy$default (Lcom/caplin/integration/streamlinkx/ContainerChange$RowChange$Removed;ILjava/lang/String;ILjava/lang/Object;)Lcom/caplin/integration/streamlinkx/ContainerChange$RowChange$Removed; + 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..56474db --- /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 [ContainerEvent] 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>): Flow + + /** + * Opens a JSON channel on [subject], sending each item from [send] to the server and emitting the + * server's updates deserialised to [type]. + */ + fun getChannel(subject: Subject, send: Flow, type: KClass): Flow> + + /** Subscribes to the record at [subject], emitting a [RecordEvent] for each field update. */ + fun getSubject(subject: Subject): Flow + + /** Subscribes to the JSON record at [subject], deserialising each update to [type]. */ + fun getSubject(subject: Subject, type: KClass): Flow> +} diff --git a/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt new file mode 100644 index 0000000..d60e687 --- /dev/null +++ b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt @@ -0,0 +1,506 @@ +@file:OptIn(ExperimentalAtomicApi::class) + +package com.caplin.integration.streamlinkx + +import com.caplin.integration.datasourcex.util.Subject +import com.caplin.integration.datasourcex.util.Subject.Companion.path +import com.caplin.integration.streamlinkx.ContainerChange.Clear +import com.caplin.integration.streamlinkx.ContainerChange.RowChange.Added +import com.caplin.integration.streamlinkx.ContainerChange.RowChange.Removed +import com.caplin.keymaster.AuthenticationParameters +import com.caplin.keymaster.IKeyMasterConfiguration +import com.caplin.keymaster.KeyMaster +import com.caplin.keymaster.KeyMasterHashingAlgorithm +import com.caplin.keymaster.KeyMasterHashingAlgorithm.SHA256 +import com.caplin.keymaster.PEMPKCS8KeyMasterConfiguration +import com.caplin.keymaster.StandardFormatter +import com.caplin.keymaster.permissioning.Permission +import com.caplin.keymaster.permissioning.UserPermissions +import com.caplin.streamlink.BaseConnectionListener +import com.caplin.streamlink.ChannelListener +import com.caplin.streamlink.ConnectionStatusEvent +import com.caplin.streamlink.ContainerElement +import com.caplin.streamlink.ContainerModel +import com.caplin.streamlink.CredentialsProvider +import com.caplin.streamlink.CredentialsReceiver +import com.caplin.streamlink.JsonChannel +import com.caplin.streamlink.JsonChannelListener +import com.caplin.streamlink.JsonEvent +import com.caplin.streamlink.JsonHandler +import com.caplin.streamlink.RecordType1Event +import com.caplin.streamlink.StreamLink +import com.caplin.streamlink.StreamLinkFactory +import com.caplin.streamlink.Subscription +import com.caplin.streamlink.SubscriptionErrorEvent +import com.caplin.streamlink.SubscriptionListener +import com.caplin.streamlink.SubscriptionStatusEvent +import com.caplin.streamlink.impl.credentials.CredentialsImpl +import com.flipkart.zjsonpatch.Jackson3JsonPatch +import io.github.oshai.kotlinlogging.KotlinLogging +import java.io.InputStream +import java.security.PrivateKey +import kotlin.concurrent.atomics.AtomicBoolean +import kotlin.concurrent.atomics.ExperimentalAtomicApi +import kotlin.io.encoding.Base64 +import kotlin.reflect.KClass +import kotlinx.collections.immutable.persistentListOf +import kotlinx.collections.immutable.toPersistentMap +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineExceptionHandler +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.SendChannel +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.channels.onFailure +import kotlinx.coroutines.channels.trySendBlocking +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.SharingStarted.Companion.Lazily +import kotlinx.coroutines.flow.callbackFlow +import kotlinx.coroutines.flow.consumeAsFlow +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.dropWhile +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onCompletion +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.onStart +import kotlinx.coroutines.flow.shareIn +import tools.jackson.databind.JsonNode +import tools.jackson.databind.ObjectMapper +import tools.jackson.module.kotlin.jacksonObjectMapper +import tools.jackson.module.kotlin.readValue + +/** + * Creates authenticated [StreamLinkConnection]s to a Liberator. + * + * Construct one via the `invoke` factory operators in the companion object, supplying the Liberator + * URL and a KeyMaster credential (a [PrivateKey], a PEM [InputStream], or a full + * [IKeyMasterConfiguration]). Each call to [connect] mints a KeyMaster token for a user and opens a + * fresh connection. + * + * @param liberator the Liberator URL to connect to (e.g. `rttp://host:port`). + * @param keymasterConfiguration KeyMaster configuration used to sign authentication tokens. + * @param objectMapper Jackson mapper used to (de)serialise JSON channel and record payloads. + */ +class StreamLinkConnectionFactory +private constructor( + private val liberator: String, + keymasterConfiguration: IKeyMasterConfiguration, + private val objectMapper: ObjectMapper = jacksonObjectMapper(), +) { + companion object { + private const val CHUNK_SIZE = 64 + + /** + * Characters that must not appear in a username. Usernames are injected verbatim into subjects + * (as a raw `%u`/`%U` path variable) and would corrupt subject parsing: `/` is the Ant path + * separator and `?` starts the query portion. + */ + private val UNSAFE_USERNAME_CHARS = setOf('/', '?') + + /** + * Encodes these DER-encoded key bytes as a PEM string, wrapping the Base64 body in + * `-----BEGIN/END [keyType]-----` markers. + */ + fun ByteArray.convertDerToPem(keyType: String = "PRIVATE KEY"): String { + val base64Encoded = Base64.encode(this) + val formattedBase64 = base64Encoded.chunked(CHUNK_SIZE).joinToString("\n") + return "-----BEGIN $keyType-----\n$formattedBase64\n-----END $keyType-----\n" + } + + /** + * Creates a factory using a KeyMaster [PrivateKey], which is converted to PEM and hashed with + * [hashingAlgorithm] when signing tokens. + */ + operator fun invoke( + liberator: String, + keymasterKey: PrivateKey, + hashingAlgorithm: KeyMasterHashingAlgorithm = SHA256, + objectMapper: ObjectMapper = jacksonObjectMapper(), + ) = + StreamLinkConnectionFactory( + liberator, + keymasterKey.encoded.convertDerToPem().byteInputStream(), + hashingAlgorithm, + objectMapper, + ) + + /** + * Creates a factory reading a PKCS#8 PEM KeyMaster key from [keymasterKey], hashed with + * [hashingAlgorithm] when signing tokens. + */ + operator fun invoke( + liberator: String, + keymasterKey: InputStream, + hashingAlgorithm: KeyMasterHashingAlgorithm = SHA256, + objectMapper: ObjectMapper = jacksonObjectMapper(), + ) = + StreamLinkConnectionFactory( + liberator, + PEMPKCS8KeyMasterConfiguration( + keymasterKey, + hashingAlgorithm, + null, + ), + objectMapper, + ) + + /** Creates a factory from a fully-formed [keymasterConfiguration]. */ + operator fun invoke( + liberator: String, + keymasterConfiguration: IKeyMasterConfiguration, + objectMapper: ObjectMapper = jacksonObjectMapper(), + ) = StreamLinkConnectionFactory(liberator, keymasterConfiguration, objectMapper) + } + + private val kLogger = KotlinLogging.logger {} + + private val scope = + CoroutineScope( + SupervisorJob() + + Dispatchers.Default + + CoroutineExceptionHandler { _, throwable -> + kLogger.error(throwable) { "Uncaught exception in StreamLink coroutine scope" } + }, + ) + + private val keymaster = KeyMaster(keymasterConfiguration) + + /** + * Opens a new [StreamLinkConnection] authenticated as [username]. + * + * A KeyMaster token granting full read/write permissions is generated for the user, and a + * StreamLink client is created and connected to the configured Liberator. The returned connection + * is single-use — [StreamLinkConnection.connect] on it will fail; create another connection to + * reconnect. + * + * @throws IllegalArgumentException if [username] contains a character that would corrupt subject + * parsing (`/` or `?`). + */ + fun connect(username: String): StreamLinkConnection { + require(username.none { it in UNSAFE_USERNAME_CHARS }) { + "Username '$username' contains characters that are not URL-safe; '/' and '?' are not allowed." + } + + val connected = AtomicBoolean(false) + + val token = + keymaster + .generateToken( + AuthenticationParameters( + username, + UserPermissions().apply { addPermission(Permission(".*", true, true, false)) }, + ), + StandardFormatter(), + ) + .let { + val tokenPrefix = "token=" + it.lines().single { line -> line.startsWith(tokenPrefix) }.removePrefix(tokenPrefix) + } + + val sl = + StreamLinkFactory.create( + StreamLinkFactory.createConfiguration().apply { + setLiberatorUrlProvider { liberator } + setJsonHandler( + object : JsonHandler { + override fun parse(value: String): Any = objectMapper.readValue(value) + + override fun patch(source: Any, diff: String): Any = + Jackson3JsonPatch.apply( + objectMapper.readValue(diff), + source as JsonNode, + ) + + override fun format(value: Any): String = objectMapper.writeValueAsString(value) + }, + ) + }, + object : CredentialsProvider { + override fun onCredentialsRequired(receiver: CredentialsReceiver) { + receiver.provideCredentials(CredentialsImpl(username, token)) + } + + override fun cancel() { + // no-op + } + }, + ) + + sl.connect() + + val state = + callbackFlow { + val listener = + object : BaseConnectionListener() { + override fun onConnectionStatusChange(event: ConnectionStatusEvent) { + trySendBlocking(event).onFailure { + if (it !is CancellationException) + kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + } + + sl.addConnectionListener(listener) + awaitClose { sl.removeConnectionListener(listener) } + } + .onStart { emit(sl.connectionCurrentState.connectionStatus) } + .distinctUntilChanged() + .shareIn(scope, Lazily, 1) + + return object : StreamLinkConnection, StreamLink by sl { + + override val username = username + + override val state = state + + override fun getChannel( + subject: Subject, + send: Flow>, + ): Flow { + val path = subject.path + kLogger.info { "Opening channel $path for $username" } + val receive = Channel() + + val channel = + createChannel( + path, + object : ChannelListener { + override fun onChannelData( + channel: com.caplin.streamlink.Channel, + event: RecordType1Event, + ) { + val event = UpdateEvent(event.fields.toPersistentMap()) + receive.trySendBlocking(event).onFailure { + if (it !is CancellationException) + kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + + override fun onChannelStatus( + channel: com.caplin.streamlink.Channel, + event: SubscriptionStatusEvent, + ) { + val event = + StatusEvent( + event.status, + event.statusMessage.orEmpty(), + event.fields.toPersistentMap(), + ) + receive.trySendBlocking(event).onFailure { + if (it !is CancellationException) + kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + + override fun onChannelError( + channel: com.caplin.streamlink.Channel, + event: SubscriptionErrorEvent, + ) { + val event = ErrorEvent(event.error, event.errorReason) + receive.trySendBlocking(event).onFailure { + if (it !is CancellationException) + kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + }, + null, + ) + + send + .onEach { channel.send(it, null) } + .onCompletion { + channel.closeChannel() + receive.close() + kLogger.info { "Channel closed $path for $username" } + } + .launchIn(scope) + + return receive.consumeAsFlow().dropWhile { it is UpdateEvent && it.payload.isEmpty() } + } + + override fun getChannel( + subject: Subject, + send: Flow, + type: KClass, + ): Flow> { + val path = subject.path + kLogger.info { "Opening channel $path for $username" } + val receive = Channel>() + + val channel = + createJsonChannel( + path, + object : JsonChannelListener { + + override fun onChannelData(channel: JsonChannel, event: JsonEvent) { + val event = + UpdateEvent( + objectMapper.convertValue( + event.json as JsonNode, + type.java, + ), + ) + receive.trySendBlocking(event).onFailure { + if (it !is CancellationException) + kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + + override fun onChannelStatus( + channel: JsonChannel, + event: SubscriptionStatusEvent, + ) { + val event = + StatusEvent( + event.status, + event.statusMessage.orEmpty(), + event.fields.toPersistentMap(), + ) + receive.trySendBlocking(event).onFailure { + if (it !is CancellationException) + kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + + override fun onChannelError(channel: JsonChannel, event: SubscriptionErrorEvent) { + val event = ErrorEvent(event.error, event.errorReason) + receive.trySendBlocking(event).onFailure { + if (it !is CancellationException) + kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + }, + null, + ) + + send + .onEach { event -> channel.send(event, null) } + .onCompletion { + channel.closeChannel() + receive.close() + kLogger.info { "Channel closed $path for $username" } + } + .launchIn(scope) + + return receive.consumeAsFlow() + } + + override fun getContainer(subject: Subject): Flow = + getSubject(subject.path) { channel -> + object : SubscriptionListener by DefaultSubscriptionListener(channel) { + + override fun onContainerUpdate( + subscription: Subscription, + event: com.caplin.streamlink.ContainerEvent, + ) { + var changes = persistentListOf() + event.updateModel( + object : ContainerModel { + + override fun clear() { + changes = changes.add(Clear) + } + + override fun insert(index: Int, element: ContainerElement) { + changes = changes.add(Added(index, element.subject)) + } + + override fun remove(index: Int, element: ContainerElement) { + changes = changes.add(Removed(index, element.subject)) + } + + override fun move(fromIndex: Int, toIndex: Int, element: ContainerElement) { + remove(fromIndex, element) + insert(toIndex, element) + } + }, + ) + changes.forEach { change -> + val event = UpdateEvent(change) + channel.trySendBlocking(event).onFailure { + if (it !is CancellationException) kLogger.error(it) { "Failed to send $event" } + if (it != null && it !is CancellationException) throw it + } + } + } + } + } + + override fun getSubject(subject: Subject): Flow = + getSubject(subject.path) { channel -> + object : SubscriptionListener by DefaultSubscriptionListener(channel) { + + override fun onRecordUpdate(subscription: Subscription, event: RecordType1Event) { + val event = UpdateEvent(event.fields.toPersistentMap()) + channel.trySendBlocking(event).onFailure { + if (it !is CancellationException) kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + } + } + + override fun getSubject(subject: Subject, type: KClass) = + getSubject(subject.path) { channel -> + object : SubscriptionListener by DefaultSubscriptionListener(channel) { + + override fun onJsonUpdate(subscription: Subscription, event: JsonEvent) { + val event = + UpdateEvent( + objectMapper.convertValue( + event.json as JsonNode, + type.java, + ), + ) + channel.trySendBlocking(event).onFailure { + if (it !is CancellationException) kLogger.error(it) { "Failed to send $event" } + if (it != null) throw it + } + } + } + } + + private fun getSubject( + path: String, + block: (SendChannel>) -> SubscriptionListener, + ): Flow> = callbackFlow { + kLogger.info { "Subscribing to subject $path for $username" } + + val handle = + subscribe( + path, + block(channel), + ) + + awaitClose { + handle.unsubscribe() + kLogger.info { "Unsubscribed from subject $path for $username" } + } + } + + override fun connect() { + check(!connected.load()) { "Create a new connection to reconnect." } + } + + override fun disconnect() { + if (connected.exchange(false)) { + kLogger.info { "Disconnecting as $username" } + sl.disconnect() + } + } + + override fun close() { + disconnect() + } + } + } +} diff --git a/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/Types.kt b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/Types.kt new file mode 100644 index 0000000..7268fb1 --- /dev/null +++ b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/Types.kt @@ -0,0 +1,167 @@ +package com.caplin.integration.streamlinkx + +import com.caplin.integration.streamlinkx.ContainerChange.Clear +import com.caplin.integration.streamlinkx.ContainerChange.RowChange.Added +import com.caplin.integration.streamlinkx.ContainerChange.RowChange.Removed +import com.caplin.streamlink.ErrorReason +import com.caplin.streamlink.SubscriptionErrorType +import com.caplin.streamlink.SubscriptionStatusType +import com.caplin.streamlink.SubscriptionStatusType.STATUS_OK +import kotlinx.collections.immutable.PersistentList +import kotlinx.collections.immutable.persistentListOf +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.filterIsInstance +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.runningFold + +/** + * An event emitted from a subscription or channel [Flow]. + * + * One of [UpdateEvent] (carrying a [T] payload), [StatusEvent] (a subscription status change) or + * [ErrorEvent] (a subscription error). + */ +sealed interface Event + +/** + * An [Event] stream carrying [ContainerChange]s, as produced by + * [StreamLinkConnection.getContainer]. + */ +typealias ContainerEvent = Event + +/** + * An [Event] stream carrying record field maps, as produced by [StreamLinkConnection.getSubject]. + */ +typealias RecordEvent = Event> + +/** A single change to the membership of a container subscription. */ +sealed interface ContainerChange { + + /** All rows were removed from the container. */ + data object Clear : ContainerChange + + /** A change affecting a single row at a given [index]. */ + sealed interface RowChange : ContainerChange { + /** The row position the change applies to. */ + val index: Int + /** The subject of the row that was added or removed. */ + val path: String + + /** A row for [path] was inserted at [index]. */ + data class Added( + override val index: Int, + override val path: String, + ) : RowChange + + /** The row for [path] at [index] was removed. */ + data class Removed( + override val index: Int, + override val path: String, + ) : RowChange + } +} + +/** + * An [Event] carrying a data [payload] — a record update, container change or deserialised value. + */ +class UpdateEvent(val payload: T) : Event { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (javaClass != other?.javaClass) return false + + other as UpdateEvent<*> + + return payload == other.payload + } + + override fun hashCode(): Int = payload?.hashCode() ?: 0 + + override fun toString(): String = "UpdateEvent(payload=$payload)" +} + +/** + * An [Event] reporting a subscription status change. + * + * @property type the status type reported by the server (e.g. [STATUS_OK]). + * @property message the accompanying status message, or empty if none. + * @property fields any fields delivered alongside the status. + */ +class StatusEvent( + val type: SubscriptionStatusType, + val message: String, + val fields: Map, +) : Event { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (javaClass != other?.javaClass) return false + + other as StatusEvent + + if (type != other.type) return false + if (message != other.message) return false + if (fields != other.fields) return false + + return true + } + + override fun hashCode(): Int { + var result = type.hashCode() + result = 31 * result + message.hashCode() + result = 31 * result + fields.hashCode() + return result + } + + override fun toString(): String = "StatusEvent(type=$type, message='$message', fields=$fields)" +} + +/** + * An [Event] reporting a subscription error, terminating the stream. + * + * @property type the category of error that occurred. + * @property reason the detailed reason for the error. + */ +class ErrorEvent( + val type: SubscriptionErrorType, + val reason: ErrorReason, +) : Event { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (javaClass != other?.javaClass) return false + + other as ErrorEvent + + if (type != other.type) return false + if (reason != other.reason) return false + + return true + } + + override fun hashCode(): Int { + var result = type.hashCode() + result = 31 * result + reason.hashCode() + return result + } + + override fun toString(): String = "ErrorEvent(type=$type, reason=$reason)" +} + +/** + * Folds a stream of [ContainerChange]s into successive snapshots of the container's row subjects, + * emitting the updated list after each change. + */ +fun Flow.runningFold(): Flow> = + runningFold(persistentListOf()) { list, event -> + when (event) { + is Added -> list.add(event.path) + is Removed -> list.remove(event.path) + Clear -> list.clear() + } + } + +/** Suspends until the stream emits a [StatusEvent] with type [STATUS_OK], returning it. */ +suspend fun Flow>.awaitStatusOk(): StatusEvent = + filterIsInstance().first { it.type == STATUS_OK } + +/** Keeps only [UpdateEvent]s from the stream, unwrapping each to its payload. */ +fun Flow>.filterIsUpdate(): Flow = + filterIsInstance>().map { it.payload } diff --git a/reactive/streamlink/kotlin/src/test/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactoryTest.kt b/reactive/streamlink/kotlin/src/test/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactoryTest.kt new file mode 100644 index 0000000..e19b607 --- /dev/null +++ b/reactive/streamlink/kotlin/src/test/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactoryTest.kt @@ -0,0 +1,31 @@ +package com.caplin.integration.streamlinkx + +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.core.spec.style.FunSpec +import io.kotest.datatest.withData +import io.kotest.matchers.string.shouldContain +import java.security.KeyPairGenerator + +class StreamLinkConnectionFactoryTest : + FunSpec( + { + val factory = + StreamLinkConnectionFactory( + "rttp://localhost:8080", + KeyPairGenerator.getInstance("RSA").generateKeyPair().private, + ) + + context("connect rejects usernames that would corrupt subject parsing") { + withData( + "user/name", + "user?name", + "/leading", + "trailing?", + "a/b?c", + ) { username -> + val exception = shouldThrow { factory.connect(username) } + exception.message shouldContain "not URL-safe" + } + } + }, + ) diff --git a/settings.gradle.kts b/settings.gradle.kts index 13a7eaf..8482e0b 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -59,6 +59,12 @@ project(":reactive:reactivestreams").projectDir = file("reactive/datasource/reac project(":reactive:reactivestreams").name = "datasourcex-reactivestreams" +include("reactive:streamlink") + +project(":reactive:streamlink").projectDir = file("reactive/streamlink/kotlin") + +project(":reactive:streamlink").name = "sl4jx-kotlin" + include("spring") project(":spring").name = "spring-boot-starter-datasource" diff --git a/version-catalog/build.gradle.kts b/version-catalog/build.gradle.kts index 81364ce..89803d9 100644 --- a/version-catalog/build.gradle.kts +++ b/version-catalog/build.gradle.kts @@ -15,6 +15,7 @@ catalog { val javaFlow = project(":reactive:datasourcex-java-flow") val kotlin = project(":reactive:datasourcex-kotlin") val reactiveStreams = project(":reactive:datasourcex-reactivestreams") + val streamlinkKotlin = project(":reactive:sl4jx-kotlin") versionCatalog { version("spring-boot", libs.versions.springBoot.get()) version(rootProject.name, util.version.toString()) @@ -44,6 +45,11 @@ catalog { project.group.toString(), reactiveStreams.name, ).versionRef(rootProject.name) + library( + streamlinkKotlin.name, + project.group.toString(), + streamlinkKotlin.name, + ).versionRef(rootProject.name) library( springBootStarterDataSource.name, project.group.toString(), From f811eb4025581f84a78af24113856e7f56edf9da Mon Sep 17 00:00:00 2001 From: Ross Anderson Date: Tue, 14 Jul 2026 16:03:06 +0100 Subject: [PATCH 04/10] Rethrow CancellationException in getContainer send failure Align getContainer's onFailure handler with every other trySendBlocking block, which rethrow CancellationException rather than swallowing it. --- .../integration/streamlinkx/StreamLinkConnectionFactory.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt index d60e687..2d3a0c5 100644 --- a/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt +++ b/reactive/streamlink/kotlin/src/main/kotlin/com/caplin/integration/streamlinkx/StreamLinkConnectionFactory.kt @@ -428,7 +428,7 @@ private constructor( val event = UpdateEvent(change) channel.trySendBlocking(event).onFailure { if (it !is CancellationException) kLogger.error(it) { "Failed to send $event" } - if (it != null && it !is CancellationException) throw it + if (it != null) throw it } } } From 317a30998c70c08fedc6e654dfaf779a37a15ff8 Mon Sep 17 00:00:00 2001 From: Ross Anderson Date: Tue, 14 Jul 2026 16:35:08 +0100 Subject: [PATCH 05/10] Import datasourcex-kotest test-support module, repackaged to com.caplin.integration.kotest Closes #69 --- AGENTS.md | 4 + gradle/libs.versions.toml | 6 + kotest/api/datasourcex-kotest.api | 99 ++++++++ kotest/build.gradle.kts | 34 +++ .../integration/streamlinkx/StreamLinkTest.kt | 29 +++ .../caplin/integration/kotest/Extensions.kt | 15 ++ .../kotest/LiberatorContainerExtension.kt | 215 ++++++++++++++++++ .../resources/default-liberator/Dockerfile | 37 +++ .../resources/default-liberator/liberator.sh | 49 ++++ settings.gradle.kts | 4 + version-catalog/build.gradle.kts | 6 + 11 files changed, 498 insertions(+) create mode 100644 kotest/api/datasourcex-kotest.api create mode 100644 kotest/build.gradle.kts create mode 100644 kotest/src/integrationTest/kotlin/com/caplin/integration/streamlinkx/StreamLinkTest.kt create mode 100644 kotest/src/main/kotlin/com/caplin/integration/kotest/Extensions.kt create mode 100644 kotest/src/main/kotlin/com/caplin/integration/kotest/LiberatorContainerExtension.kt create mode 100644 kotest/src/main/resources/default-liberator/Dockerfile create mode 100644 kotest/src/main/resources/default-liberator/liberator.sh diff --git a/AGENTS.md b/AGENTS.md index 944f94f..42eb718 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -45,6 +45,10 @@ reactive/datasource/reactivestreams ← Bind DSL for org.reactivestreams.Publish 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, Subject, Fory serialization helpers. `api`-depends on the DataSource diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 21e38d9..2b87acb 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -22,6 +22,9 @@ 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 @@ -54,6 +57,9 @@ kotlin-logging = { module = "io.github.oshai:kotlin-logging-jvm", version.ref = 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..512ddbf --- /dev/null +++ b/kotest/api/datasourcex-kotest.api @@ -0,0 +1,99 @@ +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 final fun component1 ()Lcom/caplin/keymaster/IKeyMasterConfiguration; + public final fun component2 ()I + public final fun component3 ()I + public final fun component4 ()Lorg/testcontainers/containers/GenericContainer; + public final fun copy (Lcom/caplin/keymaster/IKeyMasterConfiguration;IILorg/testcontainers/containers/GenericContainer;)Lcom/caplin/integration/kotest/LiberatorContainerExtension$CustomContainerConfig; + public static synthetic fun copy$default (Lcom/caplin/integration/kotest/LiberatorContainerExtension$CustomContainerConfig;Lcom/caplin/keymaster/IKeyMasterConfiguration;IILorg/testcontainers/containers/GenericContainer;ILjava/lang/Object;)Lcom/caplin/integration/kotest/LiberatorContainerExtension$CustomContainerConfig; + public fun equals (Ljava/lang/Object;)Z + public fun getContainer ()Lorg/testcontainers/containers/GenericContainer; + public fun getDataSourcePort ()I + public fun getHttpPort ()I + public fun getKeymasterConfiguration ()Lcom/caplin/keymaster/IKeyMasterConfiguration; + public fun hashCode ()I + public fun toString ()Ljava/lang/String; +} + +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 component1 ()Ljava/io/InputStream; + public final fun component2 ()Ljava/io/InputStream; + public final fun component3 ()Ljava/util/Map; + public final fun component4 ()Ljava/lang/String; + public final fun copy (Ljava/io/InputStream;Ljava/io/InputStream;Ljava/util/Map;Ljava/lang/String;)Lcom/caplin/integration/kotest/LiberatorContainerExtension$DefaultContainerConfig; + public static synthetic fun copy$default (Lcom/caplin/integration/kotest/LiberatorContainerExtension$DefaultContainerConfig;Ljava/io/InputStream;Ljava/io/InputStream;Ljava/util/Map;Ljava/lang/String;ILjava/lang/Object;)Lcom/caplin/integration/kotest/LiberatorContainerExtension$DefaultContainerConfig; + public fun equals (Ljava/lang/Object;)Z + 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 fun hashCode ()I + public fun toString ()Ljava/lang/String; +} + +public final class com/caplin/integration/kotest/LiberatorContainerExtension$Liberator { + public fun (Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;IILorg/testcontainers/containers/ContainerState;)V + public final fun component1 ()Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory; + public final fun component2 ()I + public final fun component3 ()I + public final fun component4 ()Lorg/testcontainers/containers/ContainerState; + public final fun connect (Ljava/lang/String;)Lcom/caplin/integration/streamlinkx/StreamLinkConnection; + public final fun copy (Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;IILorg/testcontainers/containers/ContainerState;)Lcom/caplin/integration/kotest/LiberatorContainerExtension$Liberator; + public static synthetic fun copy$default (Lcom/caplin/integration/kotest/LiberatorContainerExtension$Liberator;Lcom/caplin/integration/streamlinkx/StreamLinkConnectionFactory;IILorg/testcontainers/containers/ContainerState;ILjava/lang/Object;)Lcom/caplin/integration/kotest/LiberatorContainerExtension$Liberator; + public fun equals (Ljava/lang/Object;)Z + 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; + public fun hashCode ()I + public fun toString ()Ljava/lang/String; +} + 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..001cab3 --- /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. + */ + data 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]. + */ + data 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. + */ + data 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/settings.gradle.kts b/settings.gradle.kts index 8482e0b..4c794b1 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -65,6 +65,10 @@ project(":reactive:streamlink").projectDir = file("reactive/streamlink/kotlin") project(":reactive:streamlink").name = "sl4jx-kotlin" +include("kotest") + +project(":kotest").name = "datasourcex-kotest" + include("spring") project(":spring").name = "spring-boot-starter-datasource" diff --git a/version-catalog/build.gradle.kts b/version-catalog/build.gradle.kts index 89803d9..dc9d2a4 100644 --- a/version-catalog/build.gradle.kts +++ b/version-catalog/build.gradle.kts @@ -16,6 +16,7 @@ catalog { val kotlin = project(":reactive:datasourcex-kotlin") val reactiveStreams = project(":reactive:datasourcex-reactivestreams") val streamlinkKotlin = project(":reactive:sl4jx-kotlin") + val kotest = project(":datasourcex-kotest") versionCatalog { version("spring-boot", libs.versions.springBoot.get()) version(rootProject.name, util.version.toString()) @@ -50,6 +51,11 @@ catalog { project.group.toString(), streamlinkKotlin.name, ).versionRef(rootProject.name) + library( + kotest.name, + project.group.toString(), + kotest.name, + ).versionRef(rootProject.name) library( springBootStarterDataSource.name, project.group.toString(), From 36bfa3579e9c0be2acb3fc784550cc6f0eb3d9fc Mon Sep 17 00:00:00 2001 From: Ross Anderson Date: Tue, 14 Jul 2026 17:09:24 +0100 Subject: [PATCH 06/10] Add docker-backed integration test to examples/spring-kotlin Closes #70 --- examples/spring-kotlin/build.gradle.kts | 23 ++++ .../example/SpringKotlinIntegrationTest.kt | 128 ++++++++++++++++++ 2 files changed, 151 insertions(+) create mode 100644 examples/spring-kotlin/src/integrationTest/kotlin/example/SpringKotlinIntegrationTest.kt 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..d650ff2 --- /dev/null +++ b/examples/spring-kotlin/src/integrationTest/kotlin/example/SpringKotlinIntegrationTest.kt @@ -0,0 +1,128 @@ +package example + +import app.cash.turbine.test +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.milliseconds +import kotlin.time.Duration.Companion.seconds +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.firstOrNull +import kotlinx.coroutines.withTimeout +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 streamLink = liberator.connect("alice") + autoClose(streamLink) + + beforeSpec { + streamLink.awaitConnected() + // Gate on the adapter being discovered end-to-end: the first public update only arrives + // once the adapter has connected out to Liberator and registered its namespaces. Retry + // across the startup race — a subscription before the adapter registers errors out. + withTimeout(60.seconds) { + while ( + streamLink + .getSubject(Subject("public", "stream", "ready", "0")) + .filterIsUpdate() + .firstOrNull() == null + ) { + delay(500.milliseconds) + } + } + } + + 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() + } + } + } + }, + ) From e87376801fc67be6f4bddc74653bf68efbe08559 Mon Sep 17 00:00:00 2001 From: Ross Anderson Date: Tue, 14 Jul 2026 19:01:43 +0100 Subject: [PATCH 07/10] Replace `SimpleDataSource` with `LifecycleDataSource` to unify handling of connection lifecycle. Update factories, configurations, tests, and path handling APIs for consistency. --- .idea/compiler.xml | 1 + .idea/gradle.xml | 12 ++++--- .../example/SpringKotlinIntegrationTest.kt | 5 +++ .../internal/DataSourceAutoConfiguration.kt | 12 ++++++- .../DataSourceServerAutoConfiguration.kt | 4 +-- .../internal/DataSourceServerBootstrap.kt | 4 +-- util/api/datasourcex-util.api | 35 ++++++++++--------- ...leDataSource.kt => LifecycleDataSource.kt} | 6 ++-- .../util/SimpleDataSourceFactory.kt | 4 +-- .../integration/datasourcex/util/Subject.kt | 15 ++++++-- 10 files changed, 64 insertions(+), 34 deletions(-) rename util/src/main/kotlin/com/caplin/integration/datasourcex/util/{SimpleDataSource.kt => LifecycleDataSource.kt} (91%) 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 @@