修改响应机构在连接拦截器中

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方法实现了此示例，所以我想看看是否可以使用 interceptor 要处理它，因此我不需要确保在所有单独的RPC方法实现中都对其进行处理。

由于protoc-gen-go编译器如何定义字段的getters，因此检查请求消息是否包含now_on_device字段可以通过定义接口并使用类型sostface轻松完成字段：

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
            })
        },
    )
}

我遇到的问题（如上所述）是如何修改响应。

我无法像为请求一样定义响应的接口，因为protoc-gen-go未定义设置器。然后，我认为我可以使用类型开关（todo评论）：

switch resMsg := res.Any().(type) {
case *livev1.StartResponse:
    resMsg.DeviceOffset = durationpb.New(deviceTimeOffset)
    return &connect.Response[livev1.StartResponse]{
        Msg: resMsg,
    }, err
case *livev1.StatusResponse:
    resMsg.DeviceOffset = durationpb.New(deviceTimeOffset)
    return &connect.Response[livev1.StatusResponse]{
        Msg: resMsg,
    }, err
}

这种方法有三个问题：

1. 我找不到复制标题/拖车的方法从旧的响应到这个新响应。 （我认为目前还没有设置它们，但我不知道它们是确定的。）
2. 使用类型断言要求我对每种类型重复几乎相同的代码块。
3. 这不再比单独实现此功能更简单。

是否有一种更简单的方法来使用拦截器修改响应中的字段？还是我应该这样做的其他方法？

Answer 1

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

我找不到将旧响应中的标题/拖车复制到新响应中的方法。 （我认为目前还没有设定它们，但我不确定。）

您不需要这样做。在您的示例中，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!"}