为什么调用SendAndClose stream不返回io.EOF错误

问题描述 投票：0回答：1

我试图使用客户端流来模拟客户端/服务器 gRPC，以测试从服务器调用的 SendAndClose。 该测试运行一个每 7 秒发送一次请求的客户端和一个接收请求并在接收请求的同一流上调用 SendAndClose 的服务器。 我希望看到 io.EOF 被调用，但它看起来根本没有错误...... 可能是什么原因？

客户：

func main() {
    conn, err := grpc.Dial("localhost:8080",
        grpc.WithTransportCredentials(insecure.NewCredentials()))
    defer conn.Close()
    c := v3.NewAccessLogServiceClient(conn)

    for i := 1; i <= 100; i++ {
        if err != nil {
            log.Fatalf("could not connect: %v", err)
        }
        var opts []grpc.CallOption
        if client, e := c.StreamAccessLogs(context.Background(), opts...); e != nil {
            log.Printf("Error: %s", e)
        } else {
            msg := v3.StreamAccessLogsMessage{
                LogEntries: &v3.StreamAccessLogsMessage_HttpLogs{
                    HttpLogs: &v3.StreamAccessLogsMessage_HTTPAccessLogEntries{
                        LogEntry: []*v31.HTTPAccessLogEntry{{
                           .....
                        }}}}}
            err2 := client.SendMsg(&msg)
            if err2 != nil {
                log.Fatalf("error calling StreamAccessLog: %v", err2)
            }
        }
        time.Sleep(7 * time.Second)
    }
}

服务员：

type AccessLogService struct {
}

func main() {
    lis, err := net.Listen("tcp", ":8080")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    go func() {
        log.Println(http.ListenAndServe("localhost:6060", nil))
    }()
    s := grpc.NewServer()
    als := AccessLogService{}
    v3.RegisterAccessLogServiceServer(s, &als)
    log.Printf("Server started")
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

func (svc *AccessLogService) StreamAccessLogs(stream v3.AccessLogService_StreamAccessLogsServer) error {
    var nodeID string
    for {
        msg, err := stream.Recv()
        log.Printf("Incoming request")
        if err != nil {
            log.Printf("Error: %s", err.Error())
            return nil
        }

        switch entries := msg.LogEntries.(type) {
        case *v3.StreamAccessLogsMessage_HttpLogs:
            for _, entry := range entries.HttpLogs.LogEntry {
                log.Printf("A new Entry arrived")
            }
        }

        // Close the stream
        stream.SendAndClose(&v3.StreamAccessLogsResponse{})
    }
}

当我查看服务器日志时，我没有看到任何错误被调用：

2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: read HEADERS flags=END_HEADERS stream=157 len=8
2023/05/16 02:43:17 http2: decoded hpack field header field ":method" = "POST"
2023/05/16 02:43:17 http2: decoded hpack field header field ":scheme" = "http"
2023/05/16 02:43:17 http2: decoded hpack field header field ":path" = "/proto2.envoy.AccessLogService/StreamAccessLogs"
2023/05/16 02:43:17 http2: decoded hpack field header field ":authority" = "localhost:8080"
2023/05/16 02:43:17 http2: decoded hpack field header field "content-type" = "application/grpc"
2023/05/16 02:43:17 http2: decoded hpack field header field "user-agent" = "grpc-go/1.54.0"
2023/05/16 02:43:17 http2: decoded hpack field header field "te" = "trailers"
2023/05/16 02:43:17 http2: decoded hpack field header field "grpc-accept-encoding" = "gzip"
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: read DATA stream=157 len=11 data="\x00\x00\x00\x00\x06\x12\x04\n\x02\n\x00"
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: wrote WINDOW_UPDATE len=4 (conn) incr=11
2023/05/16 02:43:17 Got a request
2023/05/16 02:43:17 A new Entry arrived
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: wrote PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: wrote HEADERS flags=END_HEADERS stream=157 len=2
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: wrote DATA stream=157 len=5 data="\x00\x00\x00\x00\x00"
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: read PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: read WINDOW_UPDATE len=4 (conn) incr=5
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: read PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:17 http2: Framer 0xc0006be2a0: wrote PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: read HEADERS flags=END_HEADERS stream=159 len=8
2023/05/16 02:43:24 http2: decoded hpack field header field ":method" = "POST"
2023/05/16 02:43:24 http2: decoded hpack field header field ":scheme" = "http"
2023/05/16 02:43:24 http2: decoded hpack field header field ":path" = "/proto2.envoy.AccessLogService/StreamAccessLogs"
2023/05/16 02:43:24 http2: decoded hpack field header field ":authority" = "localhost:8080"
2023/05/16 02:43:24 http2: decoded hpack field header field "content-type" = "application/grpc"
2023/05/16 02:43:24 http2: decoded hpack field header field "user-agent" = "grpc-go/1.54.0"
2023/05/16 02:43:24 http2: decoded hpack field header field "te" = "trailers"
2023/05/16 02:43:24 http2: decoded hpack field header field "grpc-accept-encoding" = "gzip"
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: read DATA stream=159 len=11 data="\x00\x00\x00\x00\x06\x12\x04\n\x02\n\x00"
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: wrote WINDOW_UPDATE len=4 (conn) incr=11
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: wrote PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:24 Got a request
2023/05/16 02:43:24 A new Entry arrived
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: wrote HEADERS flags=END_HEADERS stream=159 len=2
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: wrote DATA stream=159 len=5 data="\x00\x00\x00\x00\x00"
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: read PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: read WINDOW_UPDATE len=4 (conn) incr=5
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: read PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:24 http2: Framer 0xc0006be2a0: wrote PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: read HEADERS flags=END_HEADERS stream=161 len=8
2023/05/16 02:43:31 http2: decoded hpack field header field ":method" = "POST"
2023/05/16 02:43:31 http2: decoded hpack field header field ":scheme" = "http"
2023/05/16 02:43:31 http2: decoded hpack field header field ":path" = "/proto2.envoy.AccessLogService/StreamAccessLogs"
2023/05/16 02:43:31 http2: decoded hpack field header field ":authority" = "localhost:8080"
2023/05/16 02:43:31 http2: decoded hpack field header field "content-type" = "application/grpc"
2023/05/16 02:43:31 http2: decoded hpack field header field "user-agent" = "grpc-go/1.54.0"
2023/05/16 02:43:31 http2: decoded hpack field header field "te" = "trailers"
2023/05/16 02:43:31 http2: decoded hpack field header field "grpc-accept-encoding" = "gzip"
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: read DATA stream=161 len=11 data="\x00\x00\x00\x00\x06\x12\x04\n\x02\n\x00"
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: wrote WINDOW_UPDATE len=4 (conn) incr=11
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: wrote PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:31 Got a request
2023/05/16 02:43:31 A new Entry arrived
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: wrote HEADERS flags=END_HEADERS stream=161 len=2
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: wrote DATA stream=161 len=5 data="\x00\x00\x00\x00\x00"
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: read PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: read WINDOW_UPDATE len=4 (conn) incr=5
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: read PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:31 http2: Framer 0xc0006be2a0: wrote PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: read HEADERS flags=END_HEADERS stream=163 len=8
2023/05/16 02:43:38 http2: decoded hpack field header field ":method" = "POST"
2023/05/16 02:43:38 http2: decoded hpack field header field ":scheme" = "http"
2023/05/16 02:43:38 http2: decoded hpack field header field ":path" = "/proto2.envoy.AccessLogService/StreamAccessLogs"
2023/05/16 02:43:38 http2: decoded hpack field header field ":authority" = "localhost:8080"
2023/05/16 02:43:38 http2: decoded hpack field header field "content-type" = "application/grpc"
2023/05/16 02:43:38 http2: decoded hpack field header field "user-agent" = "grpc-go/1.54.0"
2023/05/16 02:43:38 http2: decoded hpack field header field "te" = "trailers"
2023/05/16 02:43:38 http2: decoded hpack field header field "grpc-accept-encoding" = "gzip"
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: read DATA stream=163 len=11 data="\x00\x00\x00\x00\x06\x12\x04\n\x02\n\x00"
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: wrote WINDOW_UPDATE len=4 (conn) incr=11
2023/05/16 02:43:38 Got a request
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: wrote PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:38 A new Entry arrived
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: wrote HEADERS flags=END_HEADERS stream=163 len=2
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: wrote DATA stream=163 len=5 data="\x00\x00\x00\x00\x00"
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: read PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: read WINDOW_UPDATE len=4 (conn) incr=5
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: read PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:38 http2: Framer 0xc0006be2a0: wrote PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: read HEADERS flags=END_HEADERS stream=165 len=8
2023/05/16 02:43:45 http2: decoded hpack field header field ":method" = "POST"
2023/05/16 02:43:45 http2: decoded hpack field header field ":scheme" = "http"
2023/05/16 02:43:45 http2: decoded hpack field header field ":path" = "/proto2.envoy.AccessLogService/StreamAccessLogs"
2023/05/16 02:43:45 http2: decoded hpack field header field ":authority" = "localhost:8080"
2023/05/16 02:43:45 http2: decoded hpack field header field "content-type" = "application/grpc"
2023/05/16 02:43:45 http2: decoded hpack field header field "user-agent" = "grpc-go/1.54.0"
2023/05/16 02:43:45 http2: decoded hpack field header field "te" = "trailers"
2023/05/16 02:43:45 http2: decoded hpack field header field "grpc-accept-encoding" = "gzip"
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: read DATA stream=165 len=11 data="\x00\x00\x00\x00\x06\x12\x04\n\x02\n\x00"
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: wrote WINDOW_UPDATE len=4 (conn) incr=11
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: wrote PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:45 Got a request
2023/05/16 02:43:45 A new Entry arrived
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: wrote HEADERS flags=END_HEADERS stream=165 len=2
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: wrote DATA stream=165 len=5 data="\x00\x00\x00\x00\x00"
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: read PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: read WINDOW_UPDATE len=4 (conn) incr=5
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: read PING len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"
2023/05/16 02:43:45 http2: Framer 0xc0006be2a0: wrote PING flags=ACK len=8 ping="\x02\x04\x10\x10\t\x0e\a\a"

知道我做错了什么吗？

grpc http2 grpc-go
1个回答
0
投票

  1. 您的客户端代码永远不会从流中接收。它发送（不检查错误）然后在进入下一个循环迭代时丢弃

    client
    。客户端需要接收才能得到RPC的状态

  2. 您的服务器方法处理程序永远不会退出。 

    SendAndClose
    实际上并没有做任何“关闭”，而只是 
    Send
    s 并且以这种方式命名以指示不应再使用该流，并且多个 
    Send
    操作无效，因为它是客户端流式 RPC。流上的后续
    Recv
    操作将接收客户端发送的内容或块，直到从客户端接收到消息或上下文被取消。

服务器发送状态并结束 RPC 的唯一方法是从方法处理程序返回。一旦

SendAndClose
被调用，你也应该尽快从方法处理程序中
return
。

客户端流式 RPC 的典型模式是：

  1. 服务器循环接收流中的数据，直到客户端执行

    CloseSend
    （即
    Recv
    返回
    io.EOF
    ）。

  2. 服务器处理完客户端的请求消息，并返回响应消息和状态码。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.