msgexchange-v2/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt

package com.gzzn.omms.msgexchange.nextgen

import com.gzzn.omms.msgexchange.nextgen.delivery.Dispatcher
import com.gzzn.omms.msgexchange.nextgen.processing.Pump
import io.micronaut.context.annotation.Requires
import io.micronaut.runtime.event.annotation.EventListener
import io.micronaut.runtime.server.event.ServerStartupEvent
import jakarta.annotation.PreDestroy
import jakarta.inject.Singleton

/**
 * U07（T03+N29）：管道生命周期装配——服务启动（ServerStartupEvent）后，在各自专用单线程
 * （msgx-pump / msgx-dispatcher，不占用 Netty event loop）上拉起 Pump 与 Dispatcher 循环；
 * 停机时 requestStop + interrupt + join。仅当 `msgx.pipeline.autostart=true` 时装配
 * （默认关：需要真实仓储或 msgx.stubs=true 才安全开启）。
 */
@Requires(property = "msgx.pipeline.autostart", value = "true")
@Singleton
class PipelineLifecycle(
    private val pump: Pump,
    private val dispatcher: Dispatcher,
) {
    private val log = org.slf4j.LoggerFactory.getLogger(PipelineLifecycle::class.java)
    private val threads = mutableListOf<Thread>()

    @Volatile
    private var started = false

    @EventListener
    fun start(event: ServerStartupEvent) {
        startIfNeeded()
    }

    private fun startIfNeeded() {
        if (started) return
        started = true
        threads += spawn("msgx-pump", pump::loop)
        threads += spawn("msgx-dispatcher", dispatcher::loop)
        log.info("pipeline loops started (pump, dispatcher)")
    }

    private fun spawn(name: String, body: () -> Unit): Thread =
        Thread.ofPlatform().name(name).daemon(true).start { body() }

    @PreDestroy
    fun stop() {
        pump.stop()
        dispatcher.stop()
        threads.forEach { it.interrupt() }          // 解除 Thread.sleep 阻塞，加速退出
        threads.forEach { runCatching { it.join(3000) } }
        log.info("pipeline loops stopped")
    }
}