Copilot commented on code in PR #1113: URL: https://github.com/apache/incubator-seata-go/pull/1113#discussion_r3132205673
########## pkg/remoting/grpc/e2e_test.go: ########## @@ -0,0 +1,201 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package grpc + +import ( + "context" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "seata.apache.org/seata-go/v2/pkg/protocol/message" + "seata.apache.org/seata-go/v2/pkg/remoting/grpc/pb" +) + +// mockSeataServer implements SeataServiceServer for e2e testing. +// It echoes back a response with the same ID so the client Future can be resolved. +type mockSeataServer struct { + pb.UnimplementedSeataServiceServer + received chan *pb.GrpcMessageProto +} + +func (s *mockSeataServer) SendRequest(stream pb.SeataService_SendRequestServer) error { + for { + msg, err := stream.Recv() + if err != nil { + return err + } + s.received <- msg + + // Echo back a GlobalBeginResponse with the same ID so the client Future resolves. + resp := buildGlobalBeginResponse(msg.Id) + _ = stream.Send(resp) + } +} + +func buildGlobalBeginResponse(id int32) *pb.GrpcMessageProto { + respMsg := message.RpcMessage{ + ID: id, + Type: message.RequestType(pb.MessageTypeProto_TYPE_GLOBAL_BEGIN_RESULT), + Body: &pb.GlobalBeginResponseProto{ + AbstractTransactionResponse: &pb.AbstractTransactionResponseProto{ + AbstractResultMessage: &pb.AbstractResultMessageProto{ + AbstractMessage: &pb.AbstractMessageProto{ + MessageType: pb.MessageTypeProto_TYPE_GLOBAL_BEGIN_RESULT, + }, + ResultCode: pb.ResultCodeProto_Success, + }, + }, + }, + } + proto, _ := Encode(respMsg) + return proto +} + +// startMockServer starts a real gRPC server and returns its address and the received-messages channel. +func startMockServer(t *testing.T) (addr string, received chan *pb.GrpcMessageProto, stop func()) { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + mock := &mockSeataServer{received: make(chan *pb.GrpcMessageProto, 16)} + srv := grpc.NewServer() + pb.RegisterSeataServiceServer(srv, mock) + + go srv.Serve(lis) + return lis.Addr().String(), mock.received, srv.Stop +} + +// newTestChannel dials the mock server and returns a ready Channel. +func newTestChannel(t *testing.T, addr string) *Channel { + t.Helper() + conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock(), + grpc.WithTimeout(3*time.Second)) Review Comment: `grpc.WithTimeout` is deprecated in gRPC-Go. Prefer `grpc.DialContext` with a `context.WithTimeout` to enforce dial deadlines (and to avoid relying on deprecated options). ########## pkg/remoting/getty/session_manager.go: ########## @@ -158,7 +158,7 @@ func (g *SessionManager) newSession(session getty.Session) error { } func (g *SessionManager) selectSession(msg interface{}) getty.Session { - session := loadbalance.Select(config.GetSeataConfig().LoadBalanceType, &g.allSessions, g.getXid(msg)) + session := loadbalance.Select(loadbalance.GetLoadBalanceConfig().Type, &g.allSessions, g.getXid(msg)).(getty.Session) if session != nil { Review Comment: This type assertion can panic when `loadbalance.Select(...)` returns `nil` (no available sessions). Since `Select` now returns `connection.Connection`, a `nil` result has no dynamic type and cannot be asserted to `getty.Session`. Use a safe assertion (`sess, ok := ....(getty.Session)`) and handle `!ok`/`sess == nil` by returning nil or falling back. ########## pkg/remoting/config/transport_config.go: ########## @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package config + +import ( + "flag" + "time" +) + +var ( + transportConfig *TransportConfig +) + +type ShutdownConfig struct { + Wait time.Duration `yaml:"wait" json:"wait" konaf:"wait"` Review Comment: Struct tag uses `konaf:\"wait\"` instead of `koanf:\"wait\"`, so this field won't be unmarshaled from config when using koanf. Fix the tag key to `koanf` to ensure shutdown wait time is read correctly. ########## pkg/remoting/grpc/grpc_client.go: ########## @@ -0,0 +1,130 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package grpc + +import ( + "fmt" + "sync" + + gxtime "github.com/dubbogo/gost/time" + "go.uber.org/atomic" + + "seata.apache.org/seata-go/v2/pkg/protocol/codec" + "seata.apache.org/seata-go/v2/pkg/protocol/message" + "seata.apache.org/seata-go/v2/pkg/util/log" +) + +var ( + grpcRemotingClient *GrpcRemotingClient + onceGrpcRemotingClient = &sync.Once{} +) + +type GrpcRemotingClient struct { + idGenerator *atomic.Uint32 + grpcRemoting *GrpcRemoting +} + +func GetGrpcRemotingClient() *GrpcRemotingClient { + if grpcRemotingClient == nil { + onceGrpcRemotingClient.Do(func() { + grpcRemotingClient = &GrpcRemotingClient{ + idGenerator: &atomic.Uint32{}, + grpcRemoting: newGrpcRemoting(), + } + }) + } + return grpcRemotingClient +} + +func (client *GrpcRemotingClient) SendAsyncRequest(msg interface{}) error { + var msgType message.RequestType + if _, ok := msg.(message.HeartBeatMessage); ok { + msgType = message.RequestTypeHeartbeatRequest + } else { + msgType = message.RequestTypeRequestOneway + } Review Comment: gRPC heartbeat messages use `*pb.HeartbeatMessageProto`, which will not satisfy `message.HeartBeatMessage` (the Seata/getty heartbeat type). As a result, callers sending a gRPC heartbeat via `SendAsyncRequest` will incorrectly mark it as `RequestOneway` instead of `HeartbeatRequest`. Detect the gRPC heartbeat type (or introduce a transport-agnostic heartbeat marker) to ensure correct request typing. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
