流生命周期控制最佳实践

// ctx 一般来源于 handler
ctx, cancel := context.WithCancel(ctx)
defer cancel()
cliSt, err := cli.InvokeStreaming(ctx, req)
if err != nil {
    // 打日志或执行其它操作
    return
}

for {
    resp, err := cliSt.Recv(cliSt.Context())
    if err != nil {
        if err == io.EOF {
            // 正常结束
            return
        }
        // 打日志或执行其它操作
        // 异常结束
        return
    }
    // 判断是否是业务上特殊的 resp，例如 resp 中定义了特殊 flag，表示结束
    if isBizSpecialResp(resp) {
        // cancel 下游 Stream
        cancel()
        return
    }
}

import (
    "github.com/cloudwego/kitex/pkg/kerrors"
)

func (impl *ServiceImpl) InvokeStreaming(ctx context.Context, stream Service_InvokeStreamingServer) (err error) {
    // 下游不停发送数据，仅做展示
    for {
        if err = stream.Send(ctx, resp); err != nil {
            if errors.Is(kerrors.ErrStreamingCanceled, err) {
                // 上游 cancel
            }
            // 打日志或执行其它操作
            return
        }
        time.Sleep(100 * time.Millisecond)
    }
}