totole/services
1
0
Fork 0
mirror of https://github.com/kevin-DL/services.git synced 2026-01-11 19:04:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix mq

Browse Source
This commit is contained in:
Asim Aslam
2021-11-02 14:02:58 +00:00
parent 2646f96c42
commit ec440bee04
5 changed files with 52 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
mq/handler/mq.go
View File

@@ -14,9 +14,9 @@ import (
"google.golang.org/protobuf/types/known/structpb" "google.golang.org/protobuf/types/known/structpb"
) )
type MQ struct{} type Mq struct{}
func (mq *MQ) Publish(ctx context.Context, req *pb.PublishRequest, rsp *pb.PublishResponse) error { func (mq *Mq) Publish(ctx context.Context, req *pb.PublishRequest, rsp *pb.PublishResponse) error {
if len(req.Topic) == 0 { if len(req.Topic) == 0 {
return errors.BadRequest("mq.publish", "topic is blank") return errors.BadRequest("mq.publish", "topic is blank")
} }
@@ -41,7 +41,7 @@ func (mq *MQ) Publish(ctx context.Context, req *pb.PublishRequest, rsp *pb.Publi
return nil return nil
} }
func (mq *MQ) Subscribe(ctx context.Context, req *pb.SubscribeRequest, stream pb.MQ_SubscribeStream) error { func (mq *Mq) Subscribe(ctx context.Context, req *pb.SubscribeRequest, stream pb.Mq_SubscribeStream) error {
if len(req.Topic) == 0 { if len(req.Topic) == 0 {
return errors.BadRequest("mq.publish", "topic is blank") return errors.BadRequest("mq.publish", "topic is blank")
} }

2
mq/main.go
View File

@@ -15,7 +15,7 @@ func main() {
) )
// Register handler // Register handler
pb.RegisterMQHandler(srv.Server(), new(handler.MQ)) pb.RegisterMqHandler(srv.Server(), new(handler.Mq))
// Run service // Run service
if err := srv.Run(); err != nil { if err := srv.Run(); err != nil {

10
mq/proto/mq.pb.go
View File

@@ -245,7 +245,7 @@ var file_proto_mq_proto_rawDesc = []byte{
0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x31, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x31, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e,
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x53, 0x74, 0x72, 0x75, 0x63, 0x74, 0x52, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x53, 0x74, 0x72, 0x75, 0x63, 0x74, 0x52,
0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x32, 0x78, 0x0a, 0x02, 0x4d, 0x51, 0x12, 0x34, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x32, 0x78, 0x0a, 0x02, 0x4d, 0x71, 0x12, 0x34,
0x0a, 0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x12, 0x2e, 0x6d, 0x71, 0x2e, 0x50, 0x0a, 0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x12, 0x2e, 0x6d, 0x71, 0x2e, 0x50,
0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e,
0x6d, 0x71, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x6d, 0x71, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
@@ -280,10 +280,10 @@ var file_proto_mq_proto_goTypes = []interface{}{
var file_proto_mq_proto_depIdxs = []int32{ var file_proto_mq_proto_depIdxs = []int32{
4, // 0: mq.PublishRequest.message:type_name -> google.protobuf.Struct 4, // 0: mq.PublishRequest.message:type_name -> google.protobuf.Struct
4, // 1: mq.SubscribeResponse.message:type_name -> google.protobuf.Struct 4, // 1: mq.SubscribeResponse.message:type_name -> google.protobuf.Struct
0, // 2: mq.MQ.Publish:input_type -> mq.PublishRequest 0, // 2: mq.Mq.Publish:input_type -> mq.PublishRequest
2, // 3: mq.MQ.Subscribe:input_type -> mq.SubscribeRequest 2, // 3: mq.Mq.Subscribe:input_type -> mq.SubscribeRequest
1, // 4: mq.MQ.Publish:output_type -> mq.PublishResponse 1, // 4: mq.Mq.Publish:output_type -> mq.PublishResponse
3, // 5: mq.MQ.Subscribe:output_type -> mq.SubscribeResponse 3, // 5: mq.Mq.Subscribe:output_type -> mq.SubscribeResponse
4, // [4:6] is the sub-list for method output_type 4, // [4:6] is the sub-list for method output_type
2, // [2:4] is the sub-list for method input_type 2, // [2:4] is the sub-list for method input_type
2, // [2:2] is the sub-list for extension type_name 2, // [2:2] is the sub-list for extension type_name

84
mq/proto/mq.pb.micro.go
View File

@@ -34,33 +34,33 @@ var _ context.Context
var _ client.Option var _ client.Option
var _ server.Option var _ server.Option
// Api Endpoints for MQ service // Api Endpoints for Mq service
func NewMQEndpoints() []*api.Endpoint { func NewMqEndpoints() []*api.Endpoint {
return []*api.Endpoint{} return []*api.Endpoint{}
} }
// Client API for MQ service // Client API for Mq service
type MQService interface { type MqService interface {
Publish(ctx context.Context, in *PublishRequest, opts ...client.CallOption) (*PublishResponse, error) Publish(ctx context.Context, in *PublishRequest, opts ...client.CallOption) (*PublishResponse, error)
Subscribe(ctx context.Context, in *SubscribeRequest, opts ...client.CallOption) (MQ_SubscribeService, error) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...client.CallOption) (Mq_SubscribeService, error)
} }
type mQService struct { type mqService struct {
c client.Client c client.Client
name string name string
} }
func NewMQService(name string, c client.Client) MQService { func NewMqService(name string, c client.Client) MqService {
return &mQService{ return &mqService{
c: c, c: c,
name: name, name: name,
} }
} }
func (c *mQService) Publish(ctx context.Context, in *PublishRequest, opts ...client.CallOption) (*PublishResponse, error) { func (c *mqService) Publish(ctx context.Context, in *PublishRequest, opts ...client.CallOption) (*PublishResponse, error) {
req := c.c.NewRequest(c.name, "MQ.Publish", in) req := c.c.NewRequest(c.name, "Mq.Publish", in)
out := new(PublishResponse) out := new(PublishResponse)
err := c.c.Call(ctx, req, out, opts...) err := c.c.Call(ctx, req, out, opts...)
if err != nil { if err != nil {
@@ -69,8 +69,8 @@ func (c *mQService) Publish(ctx context.Context, in *PublishRequest, opts ...cli
return out, nil return out, nil
} }
func (c *mQService) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...client.CallOption) (MQ_SubscribeService, error) { func (c *mqService) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...client.CallOption) (Mq_SubscribeService, error) {
req := c.c.NewRequest(c.name, "MQ.Subscribe", &SubscribeRequest{}) req := c.c.NewRequest(c.name, "Mq.Subscribe", &SubscribeRequest{})
stream, err := c.c.Stream(ctx, req, opts...) stream, err := c.c.Stream(ctx, req, opts...)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -78,10 +78,10 @@ func (c *mQService) Subscribe(ctx context.Context, in *SubscribeRequest, opts ..
if err := stream.Send(in); err != nil { if err := stream.Send(in); err != nil {
return nil, err return nil, err
} }
return &mQServiceSubscribe{stream}, nil return &mqServiceSubscribe{stream}, nil
} }
type MQ_SubscribeService interface { type Mq_SubscribeService interface {
Context() context.Context Context() context.Context
SendMsg(interface{}) error SendMsg(interface{}) error
RecvMsg(interface{}) error RecvMsg(interface{}) error
@@ -89,27 +89,27 @@ type MQ_SubscribeService interface {
Recv() (*SubscribeResponse, error) Recv() (*SubscribeResponse, error)
} }
type mQServiceSubscribe struct { type mqServiceSubscribe struct {
stream client.Stream stream client.Stream
} }
func (x *mQServiceSubscribe) Close() error { func (x *mqServiceSubscribe) Close() error {
return x.stream.Close() return x.stream.Close()
} }
func (x *mQServiceSubscribe) Context() context.Context { func (x *mqServiceSubscribe) Context() context.Context {
return x.stream.Context() return x.stream.Context()
} }
func (x *mQServiceSubscribe) SendMsg(m interface{}) error { func (x *mqServiceSubscribe) SendMsg(m interface{}) error {
return x.stream.Send(m) return x.stream.Send(m)
} }
func (x *mQServiceSubscribe) RecvMsg(m interface{}) error { func (x *mqServiceSubscribe) RecvMsg(m interface{}) error {
return x.stream.Recv(m) return x.stream.Recv(m)
} }
func (x *mQServiceSubscribe) Recv() (*SubscribeResponse, error) { func (x *mqServiceSubscribe) Recv() (*SubscribeResponse, error) {
m := new(SubscribeResponse) m := new(SubscribeResponse)
err := x.stream.Recv(m) err := x.stream.Recv(m)
if err != nil { if err != nil {
@@ -118,42 +118,42 @@ func (x *mQServiceSubscribe) Recv() (*SubscribeResponse, error) {
return m, nil return m, nil
} }
// Server API for MQ service // Server API for Mq service
type MQHandler interface { type MqHandler interface {
Publish(context.Context, *PublishRequest, *PublishResponse) error Publish(context.Context, *PublishRequest, *PublishResponse) error
Subscribe(context.Context, *SubscribeRequest, MQ_SubscribeStream) error Subscribe(context.Context, *SubscribeRequest, Mq_SubscribeStream) error
} }
func RegisterMQHandler(s server.Server, hdlr MQHandler, opts ...server.HandlerOption) error { func RegisterMqHandler(s server.Server, hdlr MqHandler, opts ...server.HandlerOption) error {
type mQ interface { type mq interface {
Publish(ctx context.Context, in *PublishRequest, out *PublishResponse) error Publish(ctx context.Context, in *PublishRequest, out *PublishResponse) error
Subscribe(ctx context.Context, stream server.Stream) error Subscribe(ctx context.Context, stream server.Stream) error
} }
type MQ struct { type Mq struct {
mQ mq
} }
h := &mQHandler{hdlr} h := &mqHandler{hdlr}
return s.Handle(s.NewHandler(&MQ{h}, opts...)) return s.Handle(s.NewHandler(&Mq{h}, opts...))
} }
type mQHandler struct { type mqHandler struct {
MQHandler MqHandler
} }
func (h *mQHandler) Publish(ctx context.Context, in *PublishRequest, out *PublishResponse) error { func (h *mqHandler) Publish(ctx context.Context, in *PublishRequest, out *PublishResponse) error {
return h.MQHandler.Publish(ctx, in, out) return h.MqHandler.Publish(ctx, in, out)
} }
func (h *mQHandler) Subscribe(ctx context.Context, stream server.Stream) error { func (h *mqHandler) Subscribe(ctx context.Context, stream server.Stream) error {
m := new(SubscribeRequest) m := new(SubscribeRequest)
if err := stream.Recv(m); err != nil { if err := stream.Recv(m); err != nil {
return err return err
} }
return h.MQHandler.Subscribe(ctx, m, &mQSubscribeStream{stream}) return h.MqHandler.Subscribe(ctx, m, &mqSubscribeStream{stream})
} }
type MQ_SubscribeStream interface { type Mq_SubscribeStream interface {
Context() context.Context Context() context.Context
SendMsg(interface{}) error SendMsg(interface{}) error
RecvMsg(interface{}) error RecvMsg(interface{}) error
@@ -161,26 +161,26 @@ type MQ_SubscribeStream interface {
Send(*SubscribeResponse) error Send(*SubscribeResponse) error
} }
type mQSubscribeStream struct { type mqSubscribeStream struct {
stream server.Stream stream server.Stream
} }
func (x *mQSubscribeStream) Close() error { func (x *mqSubscribeStream) Close() error {
return x.stream.Close() return x.stream.Close()
} }
func (x *mQSubscribeStream) Context() context.Context { func (x *mqSubscribeStream) Context() context.Context {
return x.stream.Context() return x.stream.Context()
} }
func (x *mQSubscribeStream) SendMsg(m interface{}) error { func (x *mqSubscribeStream) SendMsg(m interface{}) error {
return x.stream.Send(m) return x.stream.Send(m)
} }
func (x *mQSubscribeStream) RecvMsg(m interface{}) error { func (x *mqSubscribeStream) RecvMsg(m interface{}) error {
return x.stream.Recv(m) return x.stream.Recv(m)
} }
func (x *mQSubscribeStream) Send(m *SubscribeResponse) error { func (x *mqSubscribeStream) Send(m *SubscribeResponse) error {
return x.stream.Send(m) return x.stream.Send(m)
} }

2
mq/proto/mq.proto
View File

@@ -5,7 +5,7 @@ package mq;
option go_package = "./proto;mq"; option go_package = "./proto;mq";
import "google/protobuf/struct.proto"; import "google/protobuf/struct.proto";
service MQ { service Mq {
rpc Publish(PublishRequest) returns (PublishResponse) {} rpc Publish(PublishRequest) returns (PublishResponse) {}
rpc Subscribe(SubscribeRequest) returns (stream SubscribeResponse) {} rpc Subscribe(SubscribeRequest) returns (stream SubscribeResponse) {}
} }