修复实时流客户端断连异常

yeg

2026-08-30 30264a7c67d44989eb9f1de1f01f0081bd298e6c

修复实时流客户端断连异常

 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestLiveServiceImplTest.java

New file
@@ -0,0 +1,117 @@
package com.ruoyi.system.service.impl;

import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;

import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Test;
import org.springframework.web.context.request.async.AsyncRequestNotUsableException;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ITestLiveService.LiveState;

/** 实时流客户端断连时应结束连接而不触发全局异常处理。 */
class TTestLiveServiceImplTest
{
    @Test
    void initialWriteFailureShouldCompleteStreamWithoutErrorDispatch()
    {
        RecordingEmitter emitter = new RecordingEmitter(0);
        TTestLiveServiceImpl service = new TestableLiveService(emitter);

        try
        {
            service.stream(4L);
        }
        finally
        {
            service.shutdown();
        }

        assertTrue(emitter.completed);
        assertFalse(emitter.completedWithError);
    }

    @Test
    void scheduledWriteFailureShouldCompleteStreamWithoutErrorDispatch()
    {
        RecordingEmitter emitter = new RecordingEmitter(6);
        TTestLiveServiceImpl service = new TestableLiveService(emitter);

        try
        {
            service.stream(4L);
            Awaitility.await().atMost(Duration.ofSeconds(5)).until(() -> emitter.completed);
        }
        finally
        {
            service.shutdown();
        }

        assertFalse(emitter.completedWithError);
    }

    private static final class TestableLiveService extends TTestLiveServiceImpl
    {
        private final RecordingEmitter emitter;

        private TestableLiveService(RecordingEmitter emitter)
        {
            super(mock(ITTestService.class), mock(DeviceDiscoveryCandidateStore.class),
                    mock(BtsDevelopmentSimulator.class), mock(TTestMapper.class));
            this.emitter = emitter;
        }

        @Override
        protected SseEmitter createEmitter()
        {
            return emitter;
        }

        @Override
        public LiveState current(Long testId)
        {
            return new LiveState(testId, "RUNNING", "RECEIVING", Instant.now(), List.of(), List.of());
        }
    }

    private static final class RecordingEmitter extends SseEmitter
    {
        private final int successfulWrites;
        private int writes;
        private volatile boolean completed;
        private volatile boolean completedWithError;

        private RecordingEmitter(int successfulWrites)
        {
            this.successfulWrites = successfulWrites;
        }

        @Override
        public void send(SseEventBuilder builder) throws IOException
        {
            if (writes++ >= successfulWrites)
                throw new AsyncRequestNotUsableException("client disconnected", new IOException("connection closed"));
        }

        @Override
        public synchronized void complete()
        {
            completed = true;
        }

        @Override
        public synchronized void completeWithError(Throwable error)
        {
            completedWithError = true;
        }
    }
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java

@@ -98,12 +98,23 @@

    @Override public SseEmitter stream(Long testId)
    {
        SseEmitter emitter = new SseEmitter(0L);
        SseEmitter emitter = createEmitter();
        try { sendAll(emitter, testId); }
        catch (RuntimeException | IOException exception) { emitter.completeWithError(exception); return emitter; }
        catch (IOException exception)
        {
            // 客户端刷新或离开页面会断开连接，不能再交给全局异常处理器写 SSE JSON。
            emitter.complete();
            return emitter;
        }
        catch (RuntimeException exception) { emitter.completeWithError(exception); return emitter; }
        ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
            try { sendAll(emitter, testId); }
            catch (RuntimeException | IOException exception) { emitter.completeWithError(exception); }
            catch (IOException exception)
            {
                // 写出失败表示 SSE 客户端已断开，静默结束连接。
                emitter.complete();
            }
            catch (RuntimeException exception) { emitter.completeWithError(exception); }
        }, 3, 3, TimeUnit.SECONDS);
        emitter.onCompletion(() -> future.cancel(false));
        emitter.onTimeout(() -> future.cancel(false));
@@ -111,6 +122,12 @@
        return emitter;
    }

    /** 创建实时流发射器，供断连生命周期测试替换。 */
    protected SseEmitter createEmitter()
    {
        return new SseEmitter(0L);
    }

    private void sendAll(SseEmitter emitter, Long testId) throws IOException
    {
        LiveState state = current(testId);