grpc

README

grpc

grpc client for evenemesh, support multiple type of loadbalancer

how to use

send message

subscribe message

setup with option

1. setup the logger

if you want to rewrite the log to other place, such as some promethus and so on, you need to implements the log.Logger interface, and set it with GRPCOption for example:

type selfLogger struct{
}

func (s *selfLogger) Infof(template string, args ...interface{}) {
	// todo
}
// ...
// other methods

cli, err := grpc.New(&conf.GRPCConfig{}, []Option{WithLogger(&selfLogger{})})

2. setup the idgen

in grpc client, we provide two kinds of id generator, uuid/flake, you can refers to commom/id for details. and if you want to implement it yourself, just implement the id.Interface api, for example:

type selfIdg struct{}
func (s* selfIdg) Next() string {
	return "uniq id"
}

cli, err := grpc.New(&conf.GRPCConfig{}, []Option{WithID(&selfIdg{})})

TODO

use etcd as service discovery

Documentation

Index

• Variables
• func CreateHeader(cfg *conf.GRPCConfig) *proto.RequestHeader
• func GetTTLWithDefault(msg *proto.SimpleMessage, def time.Duration) time.Duration
• type GRPCOption
  • func WithIDG(i id.Interface) GRPCOption
  • func WithLogger(l log.Logger) GRPCOption
  • func WithSeq(i seq.Interface) GRPCOption
• type Interface
  • func New(cfg *conf.GRPCConfig, opts ...GRPCOption) (Interface, error)
• type OnMessage
• type SimpleMessageBuilder
  • func NewMessageBuilder() *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithContent(content string) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithHeader(h *proto.RequestHeader) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithProducerGroup(grp string) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithProperties(props map[string]string) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithSeqNO(no string) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithTTL(ttl time.Duration) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithTag(tag string) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithTopic(topic string) *SimpleMessageBuilder
  • func (m *SimpleMessageBuilder) WithUniqueID(id string) *SimpleMessageBuilder

Variables

var (
	// ErrSubscribeResponse subscribe response code not ok
	ErrSubscribeResponse = fmt.Errorf("subscribe response code err")
	// ErrUnSupportResponse only support reflect.String, reflect.Struct, reflect.Ptr, reflect.Map
	ErrUnSupportResponse = fmt.Errorf("un support response msg type")
)

var (
	ErrNoMeshServer = fmt.Errorf("no event mesh server provided")
	ErrLoadBalancer = fmt.Errorf("can not peek status server from loadbalancer")

	// LoadBalancerInput key set in the context, used to store the parameter
	// into context, such as clientIP in IPHash
	LoadBalancerInput = "LoadBalancerInput"
)

var (
	// EventmeshMessage eventmesh message protocol type
	EventmeshMessage = "eventmeshmessage"

	// OpenMessage open message support later
	OpenMessage = "openmessage"
)

var EVENTMESH_MESSAGE_CONST_TTL = "ttl"

EVENTMESH_MESSAGE_CONST_TTL msg ttl store in the properties

var (
	// ErrHeartbeatResp err in sent heartbeat msg to mesh server, response not success
	ErrHeartbeatResp = fmt.Errorf("heartbeat response err")
)

var (
	// ErrTopicDispatcherExist repeated dispatcher for topic
	ErrTopicDispatcherExist = fmt.Errorf("already exist dispatcher for given topic")
)

var GRPC_ID_KEY = "GRPC_ID_KEY"

GRPC_ID_KEY key to indicate the uniq id

var Success = "0"

Success grpc code success

Functions

func CreateHeader

func CreateHeader(cfg *conf.GRPCConfig) *proto.RequestHeader

CreateHeader create msg header

func GetTTLWithDefault

func GetTTLWithDefault(msg *proto.SimpleMessage, def time.Duration) time.Duration

GetTTLWithDefault return the ttl for the given msg, if err occurred return default

Types

type GRPCOption

type GRPCOption func(*eventMeshGRPCClient)

GRPCOption option to set up the option for grpc client

func WithIDG

func WithIDG(i id.Interface) GRPCOption

WithIDG setup the id generate api

func WithLogger

func WithLogger(l log.Logger) GRPCOption

WithLogger set the logger for client, replace with the default

func WithSeq

func WithSeq(i seq.Interface) GRPCOption

type Interface

type Interface interface {
	// Publish send message to eventmesh, without wait the response from other client
	Publish(ctx context.Context, msg *proto.SimpleMessage, opts ...grpc.CallOption) (*proto.Response, error)

	// RequestReply send message to eventmesh, and wait for the response
	RequestReply(ctx context.Context, msg *proto.SimpleMessage, opts ...grpc.CallOption) (*proto.SimpleMessage, error)

	// BatchPublish send batch message to eventmesh
	BatchPublish(ctx context.Context, msg *proto.BatchMessage, opts ...grpc.CallOption) (*proto.Response, error)

	// SubscribeWebhook consumer message in webhook, and OnMessage invoked when new message arrived
	SubscribeWebhook(item conf.SubscribeItem, callbackURL string) error

	// SubscribeStream stream subscribe the message
	SubscribeStream(item conf.SubscribeItem, handler OnMessage) error

	// UnSubscribe unsubcribe topic, and don't subscribe msg anymore
	UnSubscribe() error

	// Close release all resources in the client
	Close() error
}

Interface grpc client to producer and consumer message

func New

func New(cfg *conf.GRPCConfig, opts ...GRPCOption) (Interface, error)

New create new eventmesh grpc client

type OnMessage

type OnMessage func(*proto.SimpleMessage) interface{}

OnMessage on receive message from eventmesh, used in subscribe message

type SimpleMessageBuilder

type SimpleMessageBuilder struct {
	*proto.SimpleMessage
}

SimpleMessageBuilder used to build the simple message

func NewMessageBuilder

func NewMessageBuilder() *SimpleMessageBuilder

NewMessageBuilder

func (*SimpleMessageBuilder) WithContent

func (m *SimpleMessageBuilder) WithContent(content string) *SimpleMessageBuilder

WithContent set the content to message

func (*SimpleMessageBuilder) WithHeader

func (m *SimpleMessageBuilder) WithHeader(h *proto.RequestHeader) *SimpleMessageBuilder

WithHeader set the header for message

func (*SimpleMessageBuilder) WithProducerGroup

func (m *SimpleMessageBuilder) WithProducerGroup(grp string) *SimpleMessageBuilder

WithProducerGroup set the message producer group

func (*SimpleMessageBuilder) WithProperties

func (m *SimpleMessageBuilder) WithProperties(props map[string]string) *SimpleMessageBuilder

WithProperties set the properties for message

func (*SimpleMessageBuilder) WithSeqNO

func (m *SimpleMessageBuilder) WithSeqNO(no string) *SimpleMessageBuilder

WithSeqNO set the sequence no for message

func (*SimpleMessageBuilder) WithTTL

func (m *SimpleMessageBuilder) WithTTL(ttl time.Duration) *SimpleMessageBuilder

WithTTL set the message ttl

func (*SimpleMessageBuilder) WithTag

func (m *SimpleMessageBuilder) WithTag(tag string) *SimpleMessageBuilder

WithTag set the tag for message

func (*SimpleMessageBuilder) WithTopic

func (m *SimpleMessageBuilder) WithTopic(topic string) *SimpleMessageBuilder

WithTopic set the topic

func (*SimpleMessageBuilder) WithUniqueID

func (m *SimpleMessageBuilder) WithUniqueID(id string) *SimpleMessageBuilder

WithUniqueID set the uniq id for message