Documentation ¶
Overview ¶
Package proto contains protocol buffers for gRPC messagebus event ingestion and delivery system. # Using gRPC client ## Publish example:
package main import ( "context" "fmt" "os" "time" "google.golang.org/grpc" "google.golang.org/grpc/keepalive" pubsub_proto "github.com/vardius/pubsub/proto" ) func main() { host:= "0.0.0.0" port:= 9090 ctx := context.Background() opts := []grpc.DialOption{ grpc.WithInsecure(), grpc.WithKeepaliveParams(keepalive.ClientParameters{ Time: 10 * time.Second, // send pings every 10 seconds if there is no activity Timeout: 20 * time.Second, // wait 20 second for ping ack before considering the connection dead PermitWithoutStream: true, // send pings even without active streams }), } conn, err := grpc.DialContext(ctx, fmt.Sprintf("%s:%d", host, port), opts...) if err != nil { os.Exit(1) } defer conn.Close() client := pubsub_proto.NewMessageBusClient(pubsubConn) client.Publish(ctx, &pubsub_proto.PublishRequest{ Topic: "my-topic", Payload: []byte("Hello you!"), }) }
## Subscribe example:
package main import ( "context" "fmt" "os" "time" "google.golang.org/grpc" "google.golang.org/grpc/keepalive" pubsub_proto "github.com/vardius/pubsub/proto" ) func main() { host:= "0.0.0.0" port:= 9090 ctx := context.Background() opts := []grpc.DialOption{ grpc.WithInsecure(), grpc.WithKeepaliveParams(keepalive.ClientParameters{ Time: 10 * time.Second, // send pings every 10 seconds if there is no activity Timeout: 20 * time.Second, // wait 20 second for ping ack before considering the connection dead PermitWithoutStream: true, // send pings even without active streams }), } conn, err := grpc.DialContext(ctx, fmt.Sprintf("%s:%d", host, port), opts...) if err != nil { os.Exit(1) } defer conn.Close() client := pubsub_proto.NewMessageBusClient(pubsubConn) stream, err := client.Subscribe(ctx, &pubsub_proto.SubscribeRequest{ Topic: "my-topic", }) if err != nil { os.Exit(1) } for { resp, err := stream.Recv() if err != nil { os.Exit(1) // stream closed or error } fmt.Println(resp.GetPayload()) } }
Index ¶
- func RegisterMessageBusServer(s *grpc.Server, srv MessageBusServer)
- type MessageBusClient
- type MessageBusServer
- type MessageBus_SubscribeClient
- type MessageBus_SubscribeServer
- type PublishRequest
- func (*PublishRequest) Descriptor() ([]byte, []int)
- func (m *PublishRequest) GetPayload() []byte
- func (m *PublishRequest) GetTopic() string
- func (*PublishRequest) ProtoMessage()
- func (m *PublishRequest) Reset()
- func (m *PublishRequest) String() string
- func (m *PublishRequest) XXX_DiscardUnknown()
- func (m *PublishRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
- func (m *PublishRequest) XXX_Merge(src proto.Message)
- func (m *PublishRequest) XXX_Size() int
- func (m *PublishRequest) XXX_Unmarshal(b []byte) error
- type SubscribeRequest
- func (*SubscribeRequest) Descriptor() ([]byte, []int)
- func (m *SubscribeRequest) GetTopic() string
- func (*SubscribeRequest) ProtoMessage()
- func (m *SubscribeRequest) Reset()
- func (m *SubscribeRequest) String() string
- func (m *SubscribeRequest) XXX_DiscardUnknown()
- func (m *SubscribeRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
- func (m *SubscribeRequest) XXX_Merge(src proto.Message)
- func (m *SubscribeRequest) XXX_Size() int
- func (m *SubscribeRequest) XXX_Unmarshal(b []byte) error
- type SubscribeResponse
- func (*SubscribeResponse) Descriptor() ([]byte, []int)
- func (m *SubscribeResponse) GetPayload() []byte
- func (*SubscribeResponse) ProtoMessage()
- func (m *SubscribeResponse) Reset()
- func (m *SubscribeResponse) String() string
- func (m *SubscribeResponse) XXX_DiscardUnknown()
- func (m *SubscribeResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
- func (m *SubscribeResponse) XXX_Merge(src proto.Message)
- func (m *SubscribeResponse) XXX_Size() int
- func (m *SubscribeResponse) XXX_Unmarshal(b []byte) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RegisterMessageBusServer ¶
func RegisterMessageBusServer(s *grpc.Server, srv MessageBusServer)
Types ¶
type MessageBusClient ¶
type MessageBusClient interface { Publish(ctx context.Context, in *PublishRequest, opts ...grpc.CallOption) (*empty.Empty, error) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (MessageBus_SubscribeClient, error) }
MessageBusClient is the client API for MessageBus service.
For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
func NewMessageBusClient ¶
func NewMessageBusClient(cc *grpc.ClientConn) MessageBusClient
type MessageBusServer ¶
type MessageBusServer interface { Publish(context.Context, *PublishRequest) (*empty.Empty, error) Subscribe(*SubscribeRequest, MessageBus_SubscribeServer) error }
MessageBusServer is the server API for MessageBus service.
type MessageBus_SubscribeClient ¶
type MessageBus_SubscribeClient interface { Recv() (*SubscribeResponse, error) grpc.ClientStream }
type MessageBus_SubscribeServer ¶
type MessageBus_SubscribeServer interface { Send(*SubscribeResponse) error grpc.ServerStream }
type PublishRequest ¶
type PublishRequest struct { Topic string `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"` Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` }
PublishRequest is passed when publishing
func (*PublishRequest) Descriptor ¶
func (*PublishRequest) Descriptor() ([]byte, []int)
func (*PublishRequest) GetPayload ¶
func (m *PublishRequest) GetPayload() []byte
func (*PublishRequest) GetTopic ¶
func (m *PublishRequest) GetTopic() string
func (*PublishRequest) ProtoMessage ¶
func (*PublishRequest) ProtoMessage()
func (*PublishRequest) Reset ¶
func (m *PublishRequest) Reset()
func (*PublishRequest) String ¶
func (m *PublishRequest) String() string
func (*PublishRequest) XXX_DiscardUnknown ¶
func (m *PublishRequest) XXX_DiscardUnknown()
func (*PublishRequest) XXX_Marshal ¶
func (m *PublishRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
func (*PublishRequest) XXX_Merge ¶
func (m *PublishRequest) XXX_Merge(src proto.Message)
func (*PublishRequest) XXX_Size ¶
func (m *PublishRequest) XXX_Size() int
func (*PublishRequest) XXX_Unmarshal ¶
func (m *PublishRequest) XXX_Unmarshal(b []byte) error
type SubscribeRequest ¶
type SubscribeRequest struct { Topic string `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` }
SubscribeRequest is passed when subscribing
func (*SubscribeRequest) Descriptor ¶
func (*SubscribeRequest) Descriptor() ([]byte, []int)
func (*SubscribeRequest) GetTopic ¶
func (m *SubscribeRequest) GetTopic() string
func (*SubscribeRequest) ProtoMessage ¶
func (*SubscribeRequest) ProtoMessage()
func (*SubscribeRequest) Reset ¶
func (m *SubscribeRequest) Reset()
func (*SubscribeRequest) String ¶
func (m *SubscribeRequest) String() string
func (*SubscribeRequest) XXX_DiscardUnknown ¶
func (m *SubscribeRequest) XXX_DiscardUnknown()
func (*SubscribeRequest) XXX_Marshal ¶
func (m *SubscribeRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
func (*SubscribeRequest) XXX_Merge ¶
func (m *SubscribeRequest) XXX_Merge(src proto.Message)
func (*SubscribeRequest) XXX_Size ¶
func (m *SubscribeRequest) XXX_Size() int
func (*SubscribeRequest) XXX_Unmarshal ¶
func (m *SubscribeRequest) XXX_Unmarshal(b []byte) error
type SubscribeResponse ¶
type SubscribeResponse struct { Payload []byte `protobuf:"bytes,1,opt,name=payload,proto3" json:"payload,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` }
SubscribeResponse object
func (*SubscribeResponse) Descriptor ¶
func (*SubscribeResponse) Descriptor() ([]byte, []int)
func (*SubscribeResponse) GetPayload ¶
func (m *SubscribeResponse) GetPayload() []byte
func (*SubscribeResponse) ProtoMessage ¶
func (*SubscribeResponse) ProtoMessage()
func (*SubscribeResponse) Reset ¶
func (m *SubscribeResponse) Reset()
func (*SubscribeResponse) String ¶
func (m *SubscribeResponse) String() string
func (*SubscribeResponse) XXX_DiscardUnknown ¶
func (m *SubscribeResponse) XXX_DiscardUnknown()
func (*SubscribeResponse) XXX_Marshal ¶
func (m *SubscribeResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
func (*SubscribeResponse) XXX_Merge ¶
func (m *SubscribeResponse) XXX_Merge(src proto.Message)
func (*SubscribeResponse) XXX_Size ¶
func (m *SubscribeResponse) XXX_Size() int
func (*SubscribeResponse) XXX_Unmarshal ¶
func (m *SubscribeResponse) XXX_Unmarshal(b []byte) error
Click to show internal directories.
Click to hide internal directories.