正常关闭gRPC下游

Using the following proto buffer code :

syntax = "proto3";

package pb;

message SimpleRequest {
    int64 number = 1;
}

message SimpleResponse {
    int64 doubled = 1;
}

// All the calls in this serivce preform the action of doubling a number.
// The streams will continuously send the next double, eg. 1, 2, 4, 8, 16.
service Test {
    // This RPC streams from the server only.
    rpc Downstream(SimpleRequest) returns (stream SimpleResponse);
}

I'm able to successfully open a stream, and continuously get the next doubled number from the server.

My go code for running this looks like :

ctxDownstream, cancel := context.WithCancel(ctx)
downstream, err := testClient.Downstream(ctxDownstream, &pb.SimpleRequest{Number: 1})
for {
    responseDownstream, err := downstream.Recv()
    if err != io.EOF {
        println(fmt.Sprintf("downstream response: %d, error: %v", responseDownstream.Doubled, err))

        if responseDownstream.Doubled >= 32 {
            break
        }
    }
}
cancel() // !!This is not a graceful shutdown
println(fmt.Sprintf("%v", downstream.Trailer()))

The problem I'm having is using a context cancellation means my downstream.Trailer() response is empty. Is there a way to gracefully close this connection from the client side and receive downstream.Trailer().

Note: if I close the downstream connection from the server side, my trailers are populated. But I have no way of instructing my server side to close this particular stream. So there must be a way to gracefully close a stream client side.

Thanks.

As requested some server code :

func (b *binding) Downstream(req *pb.SimpleRequest, stream pb.Test_DownstreamServer) error {
    request := req

    r := make(chan *pb.SimpleResponse)
    e := make(chan error)
    ticker := time.NewTicker(200 * time.Millisecond)
    defer func() { ticker.Stop(); close(r); close(e) }()

    go func() {
        defer func() { recover() }()
        for {
            select {
            case <-ticker.C:
                response, err := b.Endpoint(stream.Context(), request)
                if err != nil {
                    e <- err
                }
                r <- response
            }
        }
    }()

    for {
        select {
        case err := <-e:
            return err
        case response := <-r:
            if err := stream.Send(response); err != nil {
                return err
            }
            request.Number = response.Doubled
        case <-stream.Context().Done():
            return nil
        }
    }
}

You will still need to populate the trailer with some information. I use the grpc.StreamServerInterceptor to do this.

According to the grpc go documentation

Trailer returns the trailer metadata from the server, if there is any. It must only be called after stream.CloseAndRecv has returned, or stream.Recv has returned a non-nil error (including io.EOF).

So if you want to read the trailer in client try something like this

ctxDownstream, cancel := context.WithCancel(ctx)
defer cancel()
for {
  ...
  // on error or EOF
  break;
}
println(fmt.Sprintf("%v", downstream.Trailer()))

Break from the infinate loop when there is a error and print the trailer. cancel will be called at the end of the function as it is deferred.

I can't find a reference that explains it clearly, but this doesn't appear to be possible.

On the wire, grpc-status is followed by the trailer metadata when the call completes normally (i.e. the server exits the call). When the client cancels the call, neither of these are sent.

Seems that gRPC treats call cancellation as a quick abort of the rpc, not much different than the socket being dropped.

Adding a "cancel message" via request streaming works; the server can pick this up and cancel the stream from its end and trailers will still get sent:

message SimpleRequest {
    oneof RequestType {
        int64 number = 1;
        bool cancel = 2;
    }
}
....
rpc Downstream(stream SimpleRequest) returns (stream SimpleResponse);

Although this does add a bit of complication to the code.