feat:xiugai

client.go

@@ -245,24 +245,58 @@ func RpcCall(address, serviceMethod string, args any, reply any) error {
 }
 
 // grpc call
-func GrpcCall(address string, args *entity.Args) (pb.ServiceClient, context.Context, error) {
+func GrpcCall(address string, timeout int64) (pb.ServiceClient, context.Context, func(), error) {
 	var err error
 	GrpcCoonOnce.Do(func() {
 		GrpcCoon, err = grpc.Dial(address, grpc.WithInsecure())
 	})
+	var cancel context.CancelFunc
+	if err != nil {
+		log.Println(err)
+		return nil, nil, cancel, err
+	}
 	var ctx context.Context
-	if args.Timeout == -1 {
+	if timeout < 0 {
 		ctx = context.Background()
-	} else {
-		thisCtx, cancel := context.WithTimeout(context.TODO(), args.Timeout*time.Second)
-		ctx = thisCtx
-		defer cancel()
+	} else if timeout == 0 {
+		timeout = 5
 	}
+	if timeout > 0 {
+		ctx, cancel = context.WithTimeout(context.TODO(), time.Duration(timeout)*time.Second)
+	}
+	return pb.NewServiceClient(GrpcCoon), ctx, cancel, nil
+}
+
+/* grpc方式上传
+ * @param address 域名，例如：192.168.3.206:8110
+ * @param args 参数
+ * @param args.BucketID 文件名
+ * @param args.objectName 对象名称
+ * @param args.Stream 文件流
+ * @param args.Gzip 是否压缩
+ * @return {"error_code":0,"error_msg":"上传成功"}
+ * @return error 错误信息
+ */
+func UpLoadByGrpc(address string, argsFunc func() (*pb.UploadRequest, bool), timeout int64) (*pb.Response, error) {
+	serviceClient, ctx, cancel, err := GrpcCall(address, timeout)
+	defer cancel()
+	result := &pb.Response{}
+	if err != nil {
+		return result, err
+	}
+	stream, err := serviceClient.Upload(ctx)
 	if err != nil {
 		log.Println(err)
-		return nil, ctx, err
+		return nil, err
+	}
+	for {
+		if args, ok := argsFunc(); ok {
+			stream.Send(args)
+		} else {
+			break
+		}
 	}
-	return pb.NewServiceClient(GrpcCoon), ctx, nil
+	return result, err
 }
 
 /*
@@ -274,13 +308,14 @@ func GrpcCall(address string, args *entity.Args) (pb.ServiceClient, context.Cont
  * @return []byte
  * @return error 错误信息
  */
-func DownloadByGrpc(address string, args *entity.Args) (*pb.Response, error) {
-	serviceClient, ctx, err := GrpcCall(address, args)
+func DownloadByGrpc(address string, args *entity.Args, timeout int64) (*pb.Response, error) {
+	serviceClient, ctx, cancel, err := GrpcCall(address, timeout)
+	defer cancel()
 	result := &pb.Response{}
 	if err != nil {
 		return result, err
 	}
-	resp, err := serviceClient.Download(ctx, &pb.Request{BucketID: args.BucketID, ObjectName: args.ObjectName})
+	resp, err := serviceClient.Download(ctx, &pb.DownloadRequest{BucketID: args.BucketID, ObjectName: args.ObjectName})
 	if err != nil {
 		log.Println(err)
 		return nil, err
@@ -312,7 +347,7 @@ func DownloadByGrpc(address string, args *entity.Args) (*pb.Response, error) {
  * @return error 错误信息
  */
 func GetBidDetailByGrpc(address string, args *entity.Args) (*api.Result, error) {
-	result, err := DownloadByGrpc(address, args)
+	result, err := DownloadByGrpc(address, args, 0)
 	if err != nil {
 		return &api.Result{}, err
 	}

client_grpc_test.go

@@ -3,13 +3,17 @@ package ossClient
 import (
 	"jygit.jydev.jianyu360.cn/BaseService/ossClient/entity"
 	"testing"
+	"time"
 )
 
-// rpc方式上传附件
+// rpc方式下载附件
 func TestDownloadByGrpc(t *testing.T) {
-	b, err := GetBidDetailByGrpc("127.0.0.1:8012", &entity.Args{
-		BucketID:   "58b9393361a0721f1516be41",
-		ObjectName: imgFileName,
-	})
-	t.Log(err, b)
+	for {
+		b, err := GetBidDetailByGrpc("127.0.0.1:8012", &entity.Args{
+			BucketID:   "58b9393361a0721f1516be41",
+			ObjectName: imgFileName,
+		})
+		t.Log(err, b)
+		time.Sleep(3 * time.Second)
+	}
 }

entity/entity.go

@@ -1,7 +1,5 @@
 package entity
 
-import "time"
-
 // RPC相关结构体
 type UploadArgs struct {
 	Stream     []byte // 客户端将文件数据传递过来
@@ -11,7 +9,6 @@ type UploadArgs struct {
 }
 
 type Args struct {
-	BucketID   string        //桶id
-	ObjectName string        //对象名称
-	Timeout    time.Duration //超时时间
+	BucketID   string //桶id
+	ObjectName string //对象名称
 }

pb/service.pb.go

@@ -21,7 +21,7 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-type Request struct {
+type DownloadRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	BucketID      string                 `protobuf:"bytes,1,opt,name=bucketID,proto3" json:"bucketID,omitempty"`
 	ObjectName    string                 `protobuf:"bytes,2,opt,name=objectName,proto3" json:"objectName,omitempty"`
@@ -29,20 +29,20 @@ type Request struct {
 	sizeCache     protoimpl.SizeCache
 }
 
-func (x *Request) Reset() {
-	*x = Request{}
+func (x *DownloadRequest) Reset() {
+	*x = DownloadRequest{}
 	mi := &file_service_proto_msgTypes[0]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
 
-func (x *Request) String() string {
+func (x *DownloadRequest) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*Request) ProtoMessage() {}
+func (*DownloadRequest) ProtoMessage() {}
 
-func (x *Request) ProtoReflect() protoreflect.Message {
+func (x *DownloadRequest) ProtoReflect() protoreflect.Message {
 	mi := &file_service_proto_msgTypes[0]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -54,19 +54,87 @@ func (x *Request) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use Request.ProtoReflect.Descriptor instead.
-func (*Request) Descriptor() ([]byte, []int) {
+// Deprecated: Use DownloadRequest.ProtoReflect.Descriptor instead.
+func (*DownloadRequest) Descriptor() ([]byte, []int) {
 	return file_service_proto_rawDescGZIP(), []int{0}
 }
 
-func (x *Request) GetBucketID() string {
+func (x *DownloadRequest) GetBucketID() string {
 	if x != nil {
 		return x.BucketID
 	}
 	return ""
 }
 
-func (x *Request) GetObjectName() string {
+func (x *DownloadRequest) GetObjectName() string {
+	if x != nil {
+		return x.ObjectName
+	}
+	return ""
+}
+
+type UploadRequest struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Stream        []byte                 `protobuf:"bytes,1,opt,name=stream,proto3" json:"stream,omitempty"`         // 客户端将文件数据传递过来
+	Gzip          bool                   `protobuf:"varint,2,opt,name=gzip,proto3" json:"gzip,omitempty"`            //是否压缩
+	BucketID      string                 `protobuf:"bytes,3,opt,name=bucketID,proto3" json:"bucketID,omitempty"`     //桶id
+	ObjectName    string                 `protobuf:"bytes,4,opt,name=objectName,proto3" json:"objectName,omitempty"` //对象名称
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *UploadRequest) Reset() {
+	*x = UploadRequest{}
+	mi := &file_service_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *UploadRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*UploadRequest) ProtoMessage() {}
+
+func (x *UploadRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_service_proto_msgTypes[1]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use UploadRequest.ProtoReflect.Descriptor instead.
+func (*UploadRequest) Descriptor() ([]byte, []int) {
+	return file_service_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *UploadRequest) GetStream() []byte {
+	if x != nil {
+		return x.Stream
+	}
+	return nil
+}
+
+func (x *UploadRequest) GetGzip() bool {
+	if x != nil {
+		return x.Gzip
+	}
+	return false
+}
+
+func (x *UploadRequest) GetBucketID() string {
+	if x != nil {
+		return x.BucketID
+	}
+	return ""
+}
+
+func (x *UploadRequest) GetObjectName() string {
 	if x != nil {
 		return x.ObjectName
 	}
@@ -84,7 +152,7 @@ type Response struct {
 
 func (x *Response) Reset() {
 	*x = Response{}
-	mi := &file_service_proto_msgTypes[1]
+	mi := &file_service_proto_msgTypes[2]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -96,7 +164,7 @@ func (x *Response) String() string {
 func (*Response) ProtoMessage() {}
 
 func (x *Response) ProtoReflect() protoreflect.Message {
-	mi := &file_service_proto_msgTypes[1]
+	mi := &file_service_proto_msgTypes[2]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -109,7 +177,7 @@ func (x *Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Response.ProtoReflect.Descriptor instead.
 func (*Response) Descriptor() ([]byte, []int) {
-	return file_service_proto_rawDescGZIP(), []int{1}
+	return file_service_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *Response) GetData() []byte {
@@ -137,18 +205,27 @@ var File_service_proto protoreflect.FileDescriptor
 
 const file_service_proto_rawDesc = "" +
 	"\n" +
-	"\rservice.proto\x12\x02pb\"E\n" +
-	"\aRequest\x12\x1a\n" +
+	"\rservice.proto\x12\x02pb\"M\n" +
+	"\x0fDownloadRequest\x12\x1a\n" +
 	"\bbucketID\x18\x01 \x01(\tR\bbucketID\x12\x1e\n" +
 	"\n" +
 	"objectName\x18\x02 \x01(\tR\n" +
+	"objectName\"w\n" +
+	"\rUploadRequest\x12\x16\n" +
+	"\x06stream\x18\x01 \x01(\fR\x06stream\x12\x12\n" +
+	"\x04gzip\x18\x02 \x01(\bR\x04gzip\x12\x1a\n" +
+	"\bbucketID\x18\x03 \x01(\tR\bbucketID\x12\x1e\n" +
+	"\n" +
+	"objectName\x18\x04 \x01(\tR\n" +
 	"objectName\"X\n" +
 	"\bResponse\x12\x12\n" +
 	"\x04data\x18\x01 \x01(\fR\x04data\x12\x1c\n" +
 	"\terrorCode\x18\x02 \x01(\x03R\terrorCode\x12\x1a\n" +
-	"\berrorMsg\x18\x03 \x01(\tR\berrorMsg22\n" +
-	"\aService\x12'\n" +
-	"\bDownload\x12\v.pb.Request\x1a\f.pb.Response0\x01B\aZ\x05../pbb\x06proto3"
+	"\berrorMsg\x18\x03 \x01(\tR\berrorMsg2\x9c\x01\n" +
+	"\aService\x12/\n" +
+	"\bDownload\x12\x13.pb.DownloadRequest\x1a\f.pb.Response0\x01\x12+\n" +
+	"\x06Upload\x12\x11.pb.UploadRequest\x1a\f.pb.Response(\x01\x123\n" +
+	"\fGetBidDetail\x12\x13.pb.DownloadRequest\x1a\f.pb.Response0\x01B\aZ\x05../pbb\x06proto3"
 
 var (
 	file_service_proto_rawDescOnce sync.Once
@@ -162,16 +239,21 @@ func file_service_proto_rawDescGZIP() []byte {
 	return file_service_proto_rawDescData
 }
 
-var file_service_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_service_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
 var file_service_proto_goTypes = []any{
-	(*Request)(nil),  // 0: pb.Request
-	(*Response)(nil), // 1: pb.Response
+	(*DownloadRequest)(nil), // 0: pb.DownloadRequest
+	(*UploadRequest)(nil),   // 1: pb.UploadRequest
+	(*Response)(nil),        // 2: pb.Response
 }
 var file_service_proto_depIdxs = []int32{
-	0, // 0: pb.Service.Download:input_type -> pb.Request
-	1, // 1: pb.Service.Download:output_type -> pb.Response
-	1, // [1:2] is the sub-list for method output_type
-	0, // [0:1] is the sub-list for method input_type
+	0, // 0: pb.Service.Download:input_type -> pb.DownloadRequest
+	1, // 1: pb.Service.Upload:input_type -> pb.UploadRequest
+	0, // 2: pb.Service.GetBidDetail:input_type -> pb.DownloadRequest
+	2, // 3: pb.Service.Download:output_type -> pb.Response
+	2, // 4: pb.Service.Upload:output_type -> pb.Response
+	2, // 5: pb.Service.GetBidDetail:output_type -> pb.Response
+	3, // [3:6] is the sub-list for method output_type
+	0, // [0:3] is the sub-list for method input_type
 	0, // [0:0] is the sub-list for extension type_name
 	0, // [0:0] is the sub-list for extension extendee
 	0, // [0:0] is the sub-list for field type_name
@@ -188,7 +270,7 @@ func file_service_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_service_proto_rawDesc), len(file_service_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   2,
+			NumMessages:   3,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

pb/service.proto

@@ -5,14 +5,23 @@ package pb;
 option go_package = "../pb";
 
 service Service {
-  rpc Download (Request) returns (stream Response);
+  rpc Download (DownloadRequest) returns (stream Response);
+  rpc Upload (stream UploadRequest) returns (Response);
+  rpc GetBidDetail (DownloadRequest) returns (stream Response);
 }
 
-message Request {
+message DownloadRequest {
   string bucketID = 1;
   string objectName = 2;
 }
 
+message UploadRequest {
+  bytes stream = 1; // 客户端将文件数据传递过来
+  bool gzip = 2; //是否压缩
+  string bucketID = 3; //桶id
+  string objectName = 4; //对象名称
+}
+
 message Response {
   bytes data = 1;
   int64 errorCode = 2;

pb/service_grpc.pb.go

@@ -19,14 +19,18 @@ import (
 const _ = grpc.SupportPackageIsVersion9
 
 const (
-	Service_Download_FullMethodName = "/pb.Service/Download"
+	Service_Download_FullMethodName     = "/pb.Service/Download"
+	Service_Upload_FullMethodName       = "/pb.Service/Upload"
+	Service_GetBidDetail_FullMethodName = "/pb.Service/GetBidDetail"
 )
 
 // ServiceClient is the client API for Service service.
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type ServiceClient interface {
-	Download(ctx context.Context, in *Request, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Response], error)
+	Download(ctx context.Context, in *DownloadRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Response], error)
+	Upload(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[UploadRequest, Response], error)
+	GetBidDetail(ctx context.Context, in *DownloadRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Response], error)
 }
 
 type serviceClient struct {
@@ -37,13 +41,13 @@ func NewServiceClient(cc grpc.ClientConnInterface) ServiceClient {
 	return &serviceClient{cc}
 }
 
-func (c *serviceClient) Download(ctx context.Context, in *Request, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Response], error) {
+func (c *serviceClient) Download(ctx context.Context, in *DownloadRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Response], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
 	stream, err := c.cc.NewStream(ctx, &Service_ServiceDesc.Streams[0], Service_Download_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
-	x := &grpc.GenericClientStream[Request, Response]{ClientStream: stream}
+	x := &grpc.GenericClientStream[DownloadRequest, Response]{ClientStream: stream}
 	if err := x.ClientStream.SendMsg(in); err != nil {
 		return nil, err
 	}
@@ -56,11 +60,45 @@ func (c *serviceClient) Download(ctx context.Context, in *Request, opts ...grpc.
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type Service_DownloadClient = grpc.ServerStreamingClient[Response]
 
+func (c *serviceClient) Upload(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[UploadRequest, Response], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &Service_ServiceDesc.Streams[1], Service_Upload_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[UploadRequest, Response]{ClientStream: stream}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Service_UploadClient = grpc.ClientStreamingClient[UploadRequest, Response]
+
+func (c *serviceClient) GetBidDetail(ctx context.Context, in *DownloadRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Response], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &Service_ServiceDesc.Streams[2], Service_GetBidDetail_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[DownloadRequest, Response]{ClientStream: stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Service_GetBidDetailClient = grpc.ServerStreamingClient[Response]
+
 // ServiceServer is the server API for Service service.
 // All implementations must embed UnimplementedServiceServer
 // for forward compatibility.
 type ServiceServer interface {
-	Download(*Request, grpc.ServerStreamingServer[Response]) error
+	Download(*DownloadRequest, grpc.ServerStreamingServer[Response]) error
+	Upload(grpc.ClientStreamingServer[UploadRequest, Response]) error
+	GetBidDetail(*DownloadRequest, grpc.ServerStreamingServer[Response]) error
 	mustEmbedUnimplementedServiceServer()
 }
 
@@ -71,9 +109,15 @@ type ServiceServer interface {
 // pointer dereference when methods are called.
 type UnimplementedServiceServer struct{}
 
-func (UnimplementedServiceServer) Download(*Request, grpc.ServerStreamingServer[Response]) error {
+func (UnimplementedServiceServer) Download(*DownloadRequest, grpc.ServerStreamingServer[Response]) error {
 	return status.Errorf(codes.Unimplemented, "method Download not implemented")
 }
+func (UnimplementedServiceServer) Upload(grpc.ClientStreamingServer[UploadRequest, Response]) error {
+	return status.Errorf(codes.Unimplemented, "method Upload not implemented")
+}
+func (UnimplementedServiceServer) GetBidDetail(*DownloadRequest, grpc.ServerStreamingServer[Response]) error {
+	return status.Errorf(codes.Unimplemented, "method GetBidDetail not implemented")
+}
 func (UnimplementedServiceServer) mustEmbedUnimplementedServiceServer() {}
 func (UnimplementedServiceServer) testEmbeddedByValue()                 {}
 
@@ -96,16 +140,34 @@ func RegisterServiceServer(s grpc.ServiceRegistrar, srv ServiceServer) {
 }
 
 func _Service_Download_Handler(srv interface{}, stream grpc.ServerStream) error {
-	m := new(Request)
+	m := new(DownloadRequest)
 	if err := stream.RecvMsg(m); err != nil {
 		return err
 	}
-	return srv.(ServiceServer).Download(m, &grpc.GenericServerStream[Request, Response]{ServerStream: stream})
+	return srv.(ServiceServer).Download(m, &grpc.GenericServerStream[DownloadRequest, Response]{ServerStream: stream})
 }
 
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type Service_DownloadServer = grpc.ServerStreamingServer[Response]
 
+func _Service_Upload_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(ServiceServer).Upload(&grpc.GenericServerStream[UploadRequest, Response]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Service_UploadServer = grpc.ClientStreamingServer[UploadRequest, Response]
+
+func _Service_GetBidDetail_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(DownloadRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(ServiceServer).GetBidDetail(m, &grpc.GenericServerStream[DownloadRequest, Response]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Service_GetBidDetailServer = grpc.ServerStreamingServer[Response]
+
 // Service_ServiceDesc is the grpc.ServiceDesc for Service service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -119,6 +181,16 @@ var Service_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _Service_Download_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "Upload",
+			Handler:       _Service_Upload_Handler,
+			ClientStreams: true,
+		},
+		{
+			StreamName:    "GetBidDetail",
+			Handler:       _Service_GetBidDetail_Handler,
+			ServerStreams: true,
+		},
 	},
 	Metadata: "service.proto",
 }