Merge remote-tracking branch 'origin/main' into springboot-4

# Conflicts:
#	spring/build.gradle.kts
#	util/src/main/kotlin/com/caplin/integration/datasourcex/util/SimpleDataSourceFactory.kt

Author: rossdanderson

MIGRATION.md

@@ -30,14 +30,7 @@ change is required — plain POJOs serialize as before.
 
 If you registered **custom Jackson 2 modules, serializers, or `ObjectMapper` customizers**, port them
 to Jackson 3. See the
-[Jackson 3 release notes](https://github.com/FasterXML/jackson/wiki/Jackson-Release-3.0); the main
-changes are:
-
-- group/package `com.fasterxml.jackson.*` → `tools.jackson.*` (annotations stay on
-  `com.fasterxml.jackson`),
-- `JsonSerializer` / `JsonDeserializer` → `ValueSerializer` / `ValueDeserializer`,
-  `SerializerProvider` → `SerializationContext`, `JsonMappingException` → `DatabindException`,
-- the `ObjectMapper` is immutable — build it via `JsonMapper.builder()…build()`.
+[Jackson 3 release notes](https://github.com/FasterXML/jackson/wiki/Jackson-Release-3.0).
 
 ### 3. Keeping Jackson 2 (optional)
 
@@ -65,8 +58,3 @@ Outside Spring, `SimpleDataSourceFactory.createDataSource(...)` now defaults to
 handler; pass `SimpleDataSourceFactory.defaultJackson2JsonHandler` (or your own) explicitly to keep
 Jackson 2. On the `3.x` line the Jackson 2 artifacts are `compileOnly` in `datasourcex-util`, so add
 them to your own classpath if you use the Jackson 2 helpers directly.
-
-### 4. Versioning note
-
-The published `.api` surface is unchanged between `2.x` and `3.x`; the breaking change is the Spring
-Boot 4 / Jackson 3 runtime baseline, which is why this is a major version bump.

spring/build.gradle.kts

@@ -27,9 +27,9 @@ dependencies {
   testImplementation(libs.kotest.assertions)
   testImplementation(libs.kotest.runner)
   testImplementation(libs.turbine)
-  // Spring-context test of the autoconfiguration (kotest SpringExtension + @SpringBootTest).
-  testImplementation("org.springframework.boot:spring-boot-test")
+  // End-to-end SpringExtension test of the autoconfiguration + publish path.
   testImplementation(libs.kotest.extensions.spring)
+  testImplementation("org.springframework.boot:spring-boot-test")
 }
 
 val prepareReadme =

spring/src/test/kotlin/com/caplin/integration/datasourcex/spring/internal/DataSourceEndToEndTest.kt

@@ -0,0 +1,185 @@
+package com.caplin.integration.datasourcex.spring.internal
+
+import com.caplin.datasource.DataSource
+import com.caplin.integration.datasourcex.reactive.api.ContainerEvent
+import com.caplin.integration.datasourcex.reactive.api.ContainerEvent.RowEvent.Upsert
+import com.caplin.integration.datasourcex.spring.annotations.DataMessageMapping
+import com.caplin.integration.datasourcex.spring.annotations.DataMessageMapping.Type.MAPPING
+import com.caplin.integration.datasourcex.spring.annotations.DataMessageMapping.Type.RECORD_GENERIC
+import com.caplin.integration.datasourcex.spring.annotations.DataService
+import com.caplin.integration.datasourcex.spring.annotations.IngressDestinationVariable
+import com.caplin.integration.datasourcex.spring.annotations.IngressToken.USER_ID
+import io.kotest.assertions.nondeterministic.eventually
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.extensions.spring.SpringExtension
+import io.kotest.matchers.collections.shouldContain
+import io.kotest.matchers.shouldBe
+import kotlin.time.Duration.Companion.seconds
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.map
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.messaging.handler.annotation.Payload
+import org.springframework.test.context.TestPropertySource
+import tools.jackson.databind.ObjectMapper
+import tools.jackson.module.kotlin.jacksonMapperBuilder
+
+/**
+ * End-to-end test of the Spring Boot starter: a real application context wires the
+ * autoconfiguration, which binds the `@DataService` controller's `@DataMessageMapping` methods onto
+ * a (faked) DataSource. Simulating peer requests drives the whole stack — message handler, argument
+ * resolution, the controller, and the return-value handler — across every supported subject shape.
+ */
+@SpringBootTest(
+    classes =
+        [DataSourceEndToEndTest.TestConfig::class, DataSourceEndToEndTest.TestController::class]
+)
+@ImportAutoConfiguration(
+    DataSourceAutoConfiguration::class,
+    DataSourceServerAutoConfiguration::class,
+)
+@TestPropertySource(properties = ["caplin.datasource.managed.discovery.hostname=localhost"])
+class DataSourceEndToEndTest : FunSpec() {
+
+  @Autowired private lateinit var fake: FakeDataSource
+
+  init {
+    extension(SpringExtension())
+
+    test("active JSON subject publishes the controller's flow") {
+      fake.request("/price/USD")
+      eventually(TIMEOUT) {
+        fake.publishedJson shouldContain ("/price/USD" to mapOf("ccy" to "USD", "bid" to "1.23"))
+      }
+    }
+
+    test("active generic-record subject publishes the controller's flow as record fields") {
+      fake.request("/record/EUR")
+      eventually(TIMEOUT) {
+        fake.publishedRecords shouldContain
+            ("/record/EUR" to mapOf("ccy" to "EUR", "bid" to "1.10"))
+      }
+    }
+
+    test("active mapping subject publishes the remapped subject") {
+      fake.request("/map/abc")
+      eventually(TIMEOUT) { fake.publishedMappings shouldContain ("/map/abc" to "/real/abc") }
+    }
+
+    test("a streaming subject publishes each update in order then NotFound on completion") {
+      fake.request("/ticks")
+      eventually(TIMEOUT) {
+        fake.publishedJson.filter { it.first == "/ticks" }.map { it.second } shouldBe
+            listOf(mapOf("seq" to "1"), mapOf("seq" to "2"), mapOf("seq" to "3"))
+        fake.erroredSubjects shouldContain "/ticks"
+      }
+    }
+
+    test("a failing flow publishes a SubjectError and nothing else") {
+      fake.request("/broken")
+      eventually(TIMEOUT) { fake.erroredSubjects shouldContain "/broken" }
+      fake.publishedJson.any { it.first == "/broken" } shouldBe false
+    }
+
+    test("a container subject publishes element structure and serves row data on request") {
+      fake.request("/book/X")
+      eventually(TIMEOUT) {
+        fake.publishedContainers shouldContain ("/book/X" to listOf("/book/X-items/1"))
+      }
+
+      fake.request("/book/X-items/1")
+      eventually(TIMEOUT) {
+        fake.publishedRecords shouldContain ("/book/X-items/1" to mapOf("px" to "100"))
+      }
+    }
+
+    test("a request-stream channel returns a response derived from the client's message") {
+      val responses = fake.openChannel("/echo/42")
+      fake.sendToChannel("/echo/42", mapOf("ping" to "1"))
+      eventually(TIMEOUT) { responses shouldContain mapOf("ping" to "1", "echoedBy" to "42") }
+    }
+
+    test("a bidirectional channel responds to each client message") {
+      val responses = fake.openChannel("/chat/7")
+      fake.sendToChannel("/chat/7", mapOf("m" to "a"))
+      fake.sendToChannel("/chat/7", mapOf("m" to "b"))
+      eventually(TIMEOUT) {
+        responses shouldContain mapOf("m" to "a", "by" to "7")
+        responses shouldContain mapOf("m" to "b", "by" to "7")
+      }
+    }
+
+    test("a fire-and-forget channel returns an OK acknowledgement") {
+      val responses = fake.openChannel("/submit")
+      fake.sendToChannel("/submit", mapOf("x" to "1"))
+      eventually(TIMEOUT) { responses shouldContain mapOf("Result" to "ok") }
+    }
+  }
+
+  @DataService
+  class TestController {
+    @DataMessageMapping("/price/{ccy}")
+    fun price(@IngressDestinationVariable(token = USER_ID, value = "ccy") ccy: String): Flow<Any> =
+        flowOf(mapOf("ccy" to ccy, "bid" to "1.23"))
+
+    @DataMessageMapping("/record/{ccy}", type = RECORD_GENERIC)
+    fun record(
+        @IngressDestinationVariable(token = USER_ID, value = "ccy") ccy: String
+    ): Flow<Map<String, String>> = flowOf(mapOf("ccy" to ccy, "bid" to "1.10"))
+
+    @DataMessageMapping("/map/{id}", type = MAPPING)
+    fun map(@IngressDestinationVariable(token = USER_ID, value = "id") id: String): Flow<String> =
+        flowOf("/real/$id")
+
+    @DataMessageMapping("/ticks")
+    fun ticks(): Flow<Any> = flowOf(mapOf("seq" to "1"), mapOf("seq" to "2"), mapOf("seq" to "3"))
+
+    @DataMessageMapping("/broken") fun broken(): Flow<Any> = flow { error("boom") }
+
+    @DataMessageMapping("/book/{id}", type = RECORD_GENERIC)
+    fun book(
+        @IngressDestinationVariable(token = USER_ID, value = "id") id: String
+    ): Flow<ContainerEvent<Map<String, String>>> = flow {
+      emit(Upsert("1", mapOf("px" to "100")))
+      awaitCancellation()
+    }
+
+    @DataMessageMapping("/echo/{id}")
+    fun echo(
+        @IngressDestinationVariable(token = USER_ID, value = "id") id: String,
+        @Payload request: Map<String, String>,
+    ): Flow<Map<String, String>> = flowOf(request + ("echoedBy" to id))
+
+    @DataMessageMapping("/chat/{id}")
+    fun chat(
+        @IngressDestinationVariable(token = USER_ID, value = "id") id: String,
+        @Payload messages: Flow<Map<String, String>>,
+    ): Flow<Map<String, String>> = messages.map { it + ("by" to id) }
+
+    @DataMessageMapping("/submit")
+    fun submit(@Payload request: Map<String, String>) {
+      // fire-and-forget: side effects only; the starter returns an OK acknowledgement
+    }
+  }
+
+  @Configuration(proxyBeanMethods = false)
+  class TestConfig {
+    private val fake = FakeDataSource()
+
+    @Bean fun fakeDataSource(): FakeDataSource = fake
+
+    @Bean fun dataSource(): DataSource = fake.dataSource
+
+    @Bean fun objectMapper(): ObjectMapper = jacksonMapperBuilder().build()
+  }
+
+  private companion object {
+    val TIMEOUT = 5.seconds
+  }
+}

spring/src/test/kotlin/com/caplin/integration/datasourcex/spring/internal/DataSourcePayloadReturnValueHandlerTest.kt

@@ -0,0 +1,59 @@
+package com.caplin.integration.datasourcex.spring.internal
+
+import com.caplin.integration.datasourcex.spring.internal.DataSourcePayloadReturnValueHandler.Companion.RESPONSE_HEADER
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.shouldBe
+import java.util.concurrent.atomic.AtomicReference
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.runBlocking
+import org.springframework.core.MethodParameter
+import org.springframework.core.ReactiveAdapterRegistry
+import org.springframework.messaging.support.MessageBuilder
+
+internal class DataSourcePayloadReturnValueHandlerTest :
+    FunSpec({
+      val handler = DataSourcePayloadReturnValueHandler(ReactiveAdapterRegistry.getSharedInstance())
+
+      fun returnType(methodName: String) =
+          MethodParameter(ReturnValues::class.java.getMethod(methodName), -1)
+
+      fun message(ref: AtomicReference<Flow<Any>>?) =
+          MessageBuilder.withPayload("ignored")
+              .apply { ref?.let { setHeader(RESPONSE_HEADER, it) } }
+              .build()
+
+      fun collect(ref: AtomicReference<Flow<Any>>) = runBlocking { ref.get().toList() }
+
+      test("a single value is exposed as a one-element flow") {
+        val ref = AtomicReference<Flow<Any>>()
+        handler.handleReturnValue("hello", returnType("value"), message(ref)).block()
+        collect(ref) shouldBe listOf("hello")
+      }
+
+      test("a null return value is exposed as an empty flow") {
+        val ref = AtomicReference<Flow<Any>>()
+        handler.handleReturnValue(null, returnType("value"), message(ref)).block()
+        collect(ref) shouldBe emptyList()
+      }
+
+      test("a returned Flow is exposed element by element") {
+        val ref = AtomicReference<Flow<Any>>()
+        handler.handleReturnValue(flowOf("a", "b", "c"), returnType("flow"), message(ref)).block()
+        collect(ref) shouldBe listOf("a", "b", "c")
+      }
+
+      test("a missing response-reference header fails") {
+        shouldThrow<IllegalStateException> {
+          handler.handleReturnValue("hello", returnType("value"), message(null)).block()
+        }
+      }
+    })
+
+private class ReturnValues {
+  fun value(): Any = "ignored"
+
+  fun flow(): Flow<Any> = flowOf()
+}