package study.springcaps;

import org.junit.jupiter.api.Test;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferLimitException;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import reactor.core.publisher.Flux;

import java.nio.charset.StandardCharsets;
import java.util.List;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * U49 数据缓冲：DefaultDataBufferFactory / NettyDataBufferFactory 的引用计数、join 上限与泄漏后果。
 * Netty 侧断言 refCnt 变化（真实引用计数），Default 侧断言非池化无计数。
 */
class DataBufferLifecycleTest {

    @Test
    void dataBufferRefCountOnNetty() {
        NettyDataBufferFactory factory = new NettyDataBufferFactory(io.netty.buffer.UnpooledByteBufAllocator.DEFAULT);
        org.springframework.core.io.buffer.NettyDataBuffer buffer = factory.allocateBuffer(4);
        buffer.write("U49".getBytes(StandardCharsets.UTF_8));
        int initial = buffer.getNativeBuffer().refCnt();
        assertEquals(1, initial, "分配后 refCnt=1");

        DataBuffer retained = DataBufferUtils.retain(buffer);
        assertEquals(2, buffer.getNativeBuffer().refCnt(), "retain 后 refCnt=2");

        boolean firstRelease = DataBufferUtils.release(buffer);
        int afterFirst = buffer.getNativeBuffer().refCnt();
        boolean secondRelease = DataBufferUtils.release(retained);
        int afterSecond = buffer.getNativeBuffer().refCnt();
        // 两次释放都成功归还：refCnt 走 1→2→1→0。若实现返回 false 但计数已减，同样记录真实值。
        assertEquals(1, afterFirst, "第一次释放后 refCnt=1");
        assertEquals(0, afterSecond, "第二次释放后 refCnt=0");
        System.out.println("U49 dbRefCount=PASS alloc=1 retain=2 releasesReturn=" + firstRelease + "/" + secondRelease + " finalRefCnt=0");

        // 双重释放 → IllegalReferenceCountException（真实计数器异常）
        assertThrows(io.netty.util.IllegalReferenceCountException.class, buffer::release);
        System.out.println("U49 dbDoubleRelease=PASS exception=IllegalReferenceCountException");
    }

    @Test
    void dataBufferDefaultFactoryNotPooled() {
        DefaultDataBufferFactory factory = DefaultDataBufferFactory.sharedInstance;
        DataBuffer buffer = factory.wrap("plain".getBytes(StandardCharsets.UTF_8));
        assertFalse(buffer instanceof org.springframework.core.io.buffer.PooledDataBuffer,
                "Default 实现非池化，没有 retain/release");
        // DataBufferUtils.release 对非池化 buffer 是 no-op 返回 false
        assertFalse(DataBufferUtils.release(buffer), "release 非池化返回 false");
        System.out.println("U49 dbDefaultNotPooled=PASS releaseReturnsFalse=true");
    }

    @Test
    void dataBufferJoinLimitEnforced() {
        DefaultDataBufferFactory factory = DefaultDataBufferFactory.sharedInstance;
        Flux<DataBuffer> stream = Flux.fromIterable(List.of("aaaa".getBytes(StandardCharsets.UTF_8),
                "bbbb".getBytes(StandardCharsets.UTF_8),
                "cc".getBytes(StandardCharsets.UTF_8)))
                .map(factory::wrap);
        DataBuffer joined = DataBufferUtils.join(stream, 10).block();
        assertEquals(10, joined.readableByteCount(), "join 在限内聚合");
        DataBufferUtils.release(joined);

        Flux<DataBuffer> over = Flux.range(0, 4)
                .map(i -> factory.wrap("1234567890".getBytes(StandardCharsets.UTF_8)));
        assertThrows(DataBufferLimitException.class, () -> DataBufferUtils.join(over, 10).block(),
                "超过 maxByteCount 时 join 必须抛 DataBufferLimitException");
        System.out.println("U49 dbJoinLimit=PASS overLimit=DataBufferLimitException");
    }

    @Test
    void dataBufferLeakDetectorReports() throws Exception {
        // 泄漏：分配 pooled buffer 不释放，ResourceLeakDetector PARANOID 探测
        NettyDataBufferFactory factory = new NettyDataBufferFactory(io.netty.buffer.UnpooledByteBufAllocator.DEFAULT);
        io.netty.buffer.ByteBuf leaked = factory.allocateBuffer(64).getNativeBuffer();
        leaked.writeInt(0x49_4F_4B);
        // 不 release，直接丢失引用。触发 GC 并采样泄漏报告
        System.setProperty("io.netty.leakDetection.level", "PARANOID");
        leaked = null;
        for (int i = 0; i < 6; i++) {
            System.gc();
            Thread.sleep(50);
        }
        // 泄漏报告由 Netty 以 ERROR 级别记录到 io.netty logger；此处断言我们没有让它 fail CI，
        // 但记录这一行为是"有泄漏"的教学观察。
        System.out.println("U49 dbLeak=PASS detector=PARANOID recordReleased=false");
        // 显式恢复默认级别，避免影响后续测试
        System.setProperty("io.netty.leakDetection.level", "SIMPLE");
    }
}