在 connect-go 拦截器中修改 responsebody

Question

我正在使用 Buf 的connect-go库来实现 gRPC 服务器。许多 gRPC 调用对时间敏感，因此它们包含一个字段，客户端使用该字段发送其当前时间戳。服务器将客户端时间戳与本地时间戳进行比较，并返回它们之间的差异。这是.proto定义中的示例：service EventService {    // Start performing a task    rpc Start (StartRequest) returns (StartResponse);}message StartRequest {    int64 location_id = 1;    int64 task_id = 2;    Location user_latlng = 3;    google.protobuf.Timestamp now_on_device = 4;}message StartResponse {    TaskPerformanceInfo info = 1;    google.protobuf.Duration device_offset = 2;}因为我已经为几个 RPC 方法实现了这个，所以我想看看我是否可以使用拦截器来处理它，所以我不需要确保它在所有单独的 RPC 方法实现中都被处理。由于protoc-gen-go编译器为字段定义 getter 的方式，检查请求消息是否包含now_on_device字段很容易通过定义接口和使用类型断言来完成：type hasNowOnDevice interface {    GetNowOnDevice() *timestamppb.Timestamp}if reqWithNow, ok := req.Any().(hasNowOnDevice); ok {   // ...}这使得大部分拦截器非常容易编写：func MakeDeviceTimeInterceptor() func(connect.UnaryFunc) connect.UnaryFunc {    return connect.UnaryInterceptorFunc(        func(next connect.UnaryFunc) connect.UnaryFunc {            return connect.UnaryFunc(func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {                now := time.Now().UTC()                ctxa := context.WithValue(ctx, CurrentTimestampKey{}, now)                var deviceTimeOffset time.Duration                // If the protobuf message has a `NowOnDevice` field, use it                // to get the difference betweent the device time and server time.                if reqWithNow, ok := req.Any().(hasNowOnDevice); ok {                    deviceTime := reqWithNow.GetNowOnDevice().AsTime()                    deviceTimeOffset = now.Sub(deviceTime)                    ctxa = context.WithValue(ctxa, DeviceTimeDiffKey{}, deviceTimeOffset)                }                res, err := next(ctxa, req)                // TODO: How do I modify the response here?                return res, err            })        },    )}我遇到的问题（如上面评论中所述）是如何修改响应。

Answer 1

Deepankar 概述了一种解决方案，但我确实看到了将所有响应数据保存在模式定义的响应结构中的吸引力。protoc-gen-go如果生成的 setter 与 getter 一起使用，这肯定会更简单！

我找不到将旧响应中的标头/尾部复制到新响应中的方法。（我认为此时它们还没有真正确定，但我不确定。）

你不需要这样做。在您的示例中，res.Any()返回指向 protobuf 消息的指针 - 您可以就地修改它。您的类型开关可能如下所示：

switch resMsg := res.Any().(type) {

case *livev1.StartResponse:

    resMsg.DeviceOffset = durationpb.New(deviceTimeOffset)

case *livev1.StatusResponse:

    resMsg.DeviceOffset = durationpb.New(deviceTimeOffset)

}

return res, err

使用类型断言需要我为每种类型一遍又一遍地重复几乎相同的代码块。

不幸的是，你最好的选择可能是反思。您可以在标准 Go 反射或 protobuf 反射之间进行选择——两者都可以。使用 protobuf 反射，这样的事情应该可以解决问题：

res, err := next(ctx, req)

if err != nil {

    return nil, err

}

msg, ok := res.Any().(proto.Message)

if !ok {

    return res, nil

}

// Keep your logic to calculate offset!

var deviceTimeOffset time.Duration

// You could make this a global.

durationName := (*durationpb.Duration)(nil).ProtoReflect().Descriptor().FullName()

refMsg := msg.ProtoReflect()

offsetFD := refMsg.Descriptor().Fields().ByName("DeviceOffset")

if offsetFD != nil &&

    offsetFD.Message() != nil &&

    offsetFD.Message().FullName() == durationName {

    refOffset := durationpb.New(deviceTimeOffset).ProtoReflect()

    refMsg.Set(

        offsetFD, 

        protoreflect.ValueOf(refOffset),

    )

}

return res, nil

这取决于您是否认为这比重复类型切换更好或更差——它要复杂得多，但它确实让事情变得更干燥。

Answer 2

您是否有可能使用标题而不是正文。如果客户端可以NowOnDevice通过请求标头发送，那么您可以改为在响应标头中发回响应。Unix 时间戳可能是最好的方法。

func MakeDeviceTimeInterceptor() connect.UnaryInterceptorFunc {

    return func(next connect.UnaryFunc) connect.UnaryFunc {

        return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {

            now := time.Now().UTC()

            ctxa := context.WithValue(ctx, CurrentTimestampKey{}, now)

            var deviceTimeOffset time.Duration

            // Check the header message `now-on-device` field, instead of body

            reqWithNow := req.Header().Get("now-on-device")

            if reqWithNow != "" {

                val, err := strconv.Atoi(reqWithNow)

                if err != nil {

                    return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("invalid timestamp"))

                }

                deviceTime := time.Unix(int64(val), 0)

                deviceTimeOffset = now.Sub(deviceTime)

                ctxa = context.WithValue(ctxa, DeviceTimeDiffKey{}, deviceTimeOffset)

            }

            res, err := next(ctxa, req)

            // Set to response header if value is set

            if deviceTimeOffset != 0 {

                res.Header().Set("device-time-offset", fmt.Sprintf("%d", deviceTimeOffset))

            }

            return res, err

        }

    }

}

然后你有回应：

curl -v \

    --header "Content-Type: application/json" --header "now-on-device: 1656442814" \

    --data '{"name": "Jane"}' \

    http://localhost:8080/greet.v1.GreetService/Greet

*   Trying 127.0.0.1:8080...

* Connected to localhost (127.0.0.1) port 8080 (#0)

> POST /greet.v1.GreetService/Greet HTTP/1.1

> Host: localhost:8080

> User-Agent: curl/7.79.1

> Accept: */*

> Content-Type: application/json

> now-on-device: 1656442814

> Content-Length: 16

>

* Mark bundle as not supporting multiuse

< HTTP/1.1 200 OK

< Accept-Encoding: gzip

< Content-Type: application/json

< Device-Time-Offset: 7259524766000

< Greet-Version: v1

< Date: Tue, 28 Jun 2022 21:01:13 GMT

< Content-Length: 27

<

* Connection #0 to host localhost left intact

{"greeting":"Hello, Jane!"}