kubemq

package module
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Dec 16, 2018 License: Apache-2.0 Imports: 7 Imported by: 207

README

KubeMQ - Client library for Go

KubeMQ is an enterprise grade message broker for containers, designed for any type of workload and architecture running in Kubernetes. This library is Go implementation of KubeMQ client connection.

Installation

$ go get -u github.com/kubemq-io/kubemq-go

Examples

Please find usage examples on the examples folders.

KubeMQ server

Please visit https://kubemq.io, create an account, get KubeMQ token and follow the instructions to run the KubeMQ docker container in your environment.

Core Concepts

KubeMQ messaging broker has 4 messaging patterns:

  • Events - real time pub/sub pattern
  • Events Store - pub/sub with persistence pattern
  • Commands - the Command part of CQRS pattern, which send a commands with response for executed or not (with proper error messaging)
  • Queries - the Query part of CQRS pattern, which send a query and get response with the relevant query result back

For each one of the patterns we can distinguish between the senders and the receivers.

For events and events store the KubeMQ supports both rpc and upstream calls.

the data model is almost identical between all the pattern with some data added related to the specific patter.

The common part of all the patters are:

  • Id - the sender can set the Id for each type of message, or the Id will be generate UUID Id for him.
  • Metadata - a string field that can hold any metadata related to the message
  • Body - a Bytes array which hold the actual payload to be send from the sender to the receiver

The KubeMQ core transport is based on gRPC and the library is a wrapper around the client side of gRPC complied protobuf hence leveraging the gRPC benefits and advantages.

Before any transactions to be perform with KubeMQ server the Client should connect and dial KubeMQ server and obtain Client connection.

With the Client connection object the user can perform all transactions to and from KubeMQ server.

A Client connection object is thread safe and can be shared between all process needed to communicate with KubeMQ.

IMPORTANT - it's the user responsibility to close the Client connection when no further communication with KubeMQ is needed.

Connection

Connecting to KubeMQ server can be done like that:

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
client, err := kubemq.NewClient(ctx,
kubemq.WithAddress("localhost", 50000),
kubemq.WithClientId("test-event-client-id"))
if err != nil {
	log.Fatal(err)
}
defer client.Close()

List of connection options:

  • WithAddress - set host and port address of KubeMQ server
  • WithCredentials - set secured TLS credentials from the input certificate file for client.
  • WithToken - set KubeMQ token to be used for KubeMQ connection - not mandatory, only if enforced by the KubeMQ server
  • WithClientId - set client id to be used in all functions call with this client - mandatory
  • WithReceiveBufferSize - set length of buffered channel to be set in all subscriptions
  • WithDefaultChannel - set default channel for any outbound requests
  • WithDefaultCacheTTL - set default cache time to live for any query requests with any CacheKey set value

Events

Sending Events
Single Event
err := client.E().
		SetId("some-id").
		SetChannel(channel).
		SetMetadata("some-metadata").
		SetBody([]byte("hello kubemq - sending single event")).
		Send(ctx)
if err != nil {
	log.Fatal(err)
}
Stream Events
eventStreamCh := make(chan *kubemq.Event, 1)
errStreamCh := make(chan error, 1)
go client.StreamEvents(ctx, eventStreamCh, errStreamCh)
event := client.E().SetId("some-event-id").
	SetChannel("some_channel").
	SetMetadata("some-metadata").
	SetBody([]byte("hello kubemq - sending stream event"))
for {
	select {
	case err := <-errStreamCh:
		log.Println(err)
		return
	case eventStreamCh <- event:
		return
	}
}
Receiving Events

First you should subscribe to Events and get a channel:

channelName := "testing_event_channel"
errCh := make(chan error)
eventsCh, err := client.SubscribeToEvents(ctx, channelName, "", errCh)
if err != nil {
 	log.Fatal(err)
}

Then you can loop over the channel of events:

for {
	select {
	case err := <-errCh:
		log.Fatal(err)
	case event := <-eventsCh:
		log.Printf("Event Recevied:\nEventID: %s\nChannel: %s\nMetadata: %s\nBody: %s\n", event.Id, event.Channel, event.Metadata, event.Body)
	}
}

Events Store

Sending Events Store
Single Event to Store
//sending 10 single events to store
for i := 0; i < 10; i++ {
	result, err := client.ES().
	    SetId(fmt.Sprintf("event-store-%d", i)).
		SetChannel(channelName).
		SetMetadata("some-metadata").
		SetBody([]byte("hello kubemq - sending single event to store")).
		Send(ctx)
	if err != nil {
			log.Fatal(err)
	}
	log.Printf("Sending event #%d: Result: %t", i, result.Sent)
}
Stream Events Store
// sending addtional events to store
eventsStoreStreamCh := make(chan *kubemq.EventStore, 1)
eventsStoreSResultCh := make(chan *kubemq.EventStoreResult, 1)
errStreamCh := make(chan error, 1)
go client.StreamEventsStore(ctx, eventsStoreStreamCh, eventsStoreSResultCh, errStreamCh)
for i := 0; i < 10; i++ {
    event := client.ES().
    SetId(fmt.Sprintf("event-store-%d", i)).
    SetChannel(channelName).
    SetMetadata("some-metadata").
    SetBody([]byte("hello kubemq - sending stream event to store"))
    eventsStoreStreamCh <- event
    select {
        case err := <-errStreamCh:
    		log.Println(err)
    		return
   		case result := <-eventsStoreSResultCh:
   			log.Printf("Sending event #%d: Result: %t", i, result.Sent)
  		}
}
Receiving Events Store

First you should subscribe to Events Store and get a channel:

eventsCh, err := client.SubscribeToEventsStore(ctx, channelName, "", errCh, kubemq.StartFromFirstEvent())
if err != nil {
 	log.Fatal(err)
 }
Subscription Options

KubeMQ supports 6 type of subscriptions:

  • StartFromNewEvents - start event store subscription with only new events
  • StartFromFirstEvent - replay all the stored events from the first available sequence and continue stream new events from this point
  • StartFromLastEvent - replay last event and continue stream new events from this point
  • StartFromSequence - replay events from specific event sequence number and continue stream new events from this point
  • StartFromTime - replay events from specific time continue stream new events from this point
  • StartFromTimeDelta - replay events from specific current time - delta duration in seconds, continue stream new events from this point

Then you can loop over the channel of events:

for {
	select {
	case err := <-errCh:
		log.Fatal(err)
	case event := <-eventsCh:
		log.Printf("Receive EventStore\nSequence: %d\nTime: %s\nBody: %s\n", event.Sequence, event.Timestamp, event.Body)
	}
}

Commands

Concept

Commands implements synchronous messaging pattern which the sender send a request and wait for specific amount of time to get a response.

The response can be successful or not. This is the responsibility of the responder to return with the result of the command within the time the sender set in the request.

Sending Command Requests

In this example, the responder should send his response withing one second, otherwise an error will be return as timout.

response, err := client.C().
	SetId("some-command-id").
	SetChannel(channelName).
	SetMetadata("some-metadata").
	SetBody([]byte("hello kubemq - sending command, please reply")).
	SetTimeout(time.Second).
	Send(ctx)
if err != nil {
	log.Fatal(err)
}
Receiving Commands Requests

First get a channel of commands:

errCh := make(chan error)
commandsCh, err := client.SubscribeToCommands(ctx, channelName, "", errCh)
if err != nil {
		log.Fatal(err)
    }

Then a loop over the channel will get the requests from the senders.

for {
	select {
	case err := <-errCh:
		log.Fatal(err)
        return
	case command := <-commandsCh:
		log.Printf("Command Recevied:\nId %s\nChannel: %s\nMetadata: %s\nBody: %s\n", command.Id, command.Channel, command.Metadata, command.Body)
	case <-ctx.Done():
		return
	}
}
Sending a Command Response

When sending response there are two important things to remember:

  • Set the relevant requestId which you response to
  • Set the ResponseTo string to the value of the request ResponseTo field
err := client.R().
    SetRequestId(command.Id).
	SetResponseTo(command.ResponseTo).
	SetExecutedAt(time.Now()).
	Send(ctx)
if err != nil {
	log.Fatal(err)
}

Queries

Concept

Queries implements synchronous messaging pattern which the sender send a request and wait for specific amount of time to get a response.

The response must includes metadata or body together with indication of successful or not operation. This is the responsibility of the responder to return with the result of the query within the time the sender set in the request.

Sending Query Requests

In this example, the responder should send his response withing one second, otherwise an error will be return as timout.

response, err := client.Q().
    SetId("some-query-id").
    SetChannel(channel).
	SetMetadata("some-metadata").
	SetBody([]byte("hello kubemq - sending a query, please reply")).
	SetTimeout(time.Second).
	Send(ctx)
if err != nil {
	log.Fatal(err)
}
Receiving Query Requests

First get a channel of queries:

errCh := make(chan error)
queriesCh, err := client.SubscribeToQueries(ctx, channelName, "", errCh)
if err != nil {
	log.Fatal(err)
}

Then a loop over the channel will get the requests from the senders.

for {
	select {
	case err := <-errCh:
		log.Fatal(err)
		return
	case query := <-queriesCh:
		log.Printf("Query Recevied:\nId %s\nChannel: %s\nMetadata: %s\nBody: %s\n", query.Id, query.Channel, query.Metadata, query.Body)
	case <-ctx.Done():
		return
   	}
}
Sending a Query Response

When sending response there are two important things to remember:

  • Set the relevant requestId which you response to
  • Set the ResponseTo string to the value of the request ResponseTo field
err := client.R().
    SetRequestId(query.Id).
  	SetResponseTo(query.ResponseTo).
  	SetExecutedAt(time.Now()).
  	SetMetadata("this is a response").
  	SetBody([]byte("got your query, you are good to go")).
  	Send(ctx)
  	
if err != nil {
	log.Fatal(err)
}

Support

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

func NewClient 

func NewClient(ctx context.Context, op ...Option) (*Client, error)

NewClient - create client instance to be use to communicate with KubeMQ server

func (*Client) C 

func (c *Client) C() *Command

C - create an empty command object

func (*Client) Close 

func (c *Client) Close() error

Close - closing client connection. any on going transactions will be aborted

func (*Client) E 

func (c *Client) E() *Event

E - create an empty event object

func (*Client) ES 

func (c *Client) ES() *EventStore

ES - create an empty event store object

func (*Client) Q 

func (c *Client) Q() *Query

Q - create an empty query object

func (*Client) R 

func (c *Client) R() *Response

R - create an empty response object for command or query responses

func (*Client) StreamEvents 

func (c *Client) StreamEvents(ctx context.Context, eventsCh chan *Event, errCh chan error)

StreamEvents - send stream of events in a single call

func (*Client) StreamEventsStore 

func (c *Client) StreamEventsStore(ctx context.Context, eventsCh chan *EventStore, eventsResultCh chan *EventStoreResult, errCh chan error)

StreamEventsStore - send stream of events store in a single call

func (*Client) SubscribeToCommands 

func (c *Client) SubscribeToCommands(ctx context.Context, channel, group string, errCh chan error) (<-chan *CommandReceive, error)

SubscribeToCommands - subscribe to commands requests by channel and group. return channel of CommandReceived or en error

func (*Client) SubscribeToEvents 

func (c *Client) SubscribeToEvents(ctx context.Context, channel, group string, errCh chan error) (<-chan *Event, error)

SubscribeToEvents - subscribe to events by channel and group. return channel of events or en error

func (*Client) SubscribeToEventsStore 

func (c *Client) SubscribeToEventsStore(ctx context.Context, channel, group string, errCh chan error, opt SubscriptionOption) (<-chan *EventStoreReceive, error)

SubscribeToEventsStore - subscribe to events store by channel and group with subscription option. return channel of events or en error

func (*Client) SubscribeToQueries 

func (c *Client) SubscribeToQueries(ctx context.Context, channel, group string, errCh chan error) (<-chan *QueryReceive, error)

SubscribeToQueries - subscribe to queries requests by channel and group. return channel of QueryReceived or en error

type Command 

type Command struct {
	Id       string
	Channel  string
	Metadata string
	Body     []byte
	Timeout  time.Duration
	ClientId string
	// contains filtered or unexported fields
}

func (*Command) Send 

func (c *Command) Send(ctx context.Context) (*CommandResponse, error)

Send - sending command , waiting for response or timeout

func (*Command) SetBody 

func (c *Command) SetBody(body []byte) *Command

SetBody - set command body - mandatory if metadata field is empty

func (*Command) SetChannel 

func (c *Command) SetChannel(channel string) *Command

SetChannel - set command channel - mandatory if default channel was not set

func (*Command) SetClientId 

func (c *Command) SetClientId(clientId string) *Command

SetClientId - set command ClientId - mandatory if default client was not set

func (*Command) SetId 

func (c *Command) SetId(id string) *Command

SetId - set command requestId, otherwise new random uuid will be set

func (*Command) SetMetadata 

func (c *Command) SetMetadata(metadata string) *Command

SetMetadata - set command metadata - mandatory if body field is empty

func (*Command) SetTimeout 

func (c *Command) SetTimeout(timeout time.Duration) *Command

SetTimeout - set timeout for command to be returned. if timeout expired , send command will result with an error

type CommandReceive 

type CommandReceive struct {
	Id         string
	Channel    string
	Metadata   string
	Body       []byte
	ResponseTo string
}

type CommandResponse 

type CommandResponse struct {
	CommandId        string
	ResponseClientId string
	Executed         bool
	ExecutedAt       time.Time
	Error            string
}

type Event 

type Event struct {
	Id       string
	Channel  string
	Metadata string
	Body     []byte
	ClientId string
	// contains filtered or unexported fields
}

func (*Event) Send 

func (e *Event) Send(ctx context.Context) error

func (*Event) SetBody 

func (e *Event) SetBody(body []byte) *Event

SetBody - set event body - mandatory if metadata field was not set

func (*Event) SetChannel 

func (e *Event) SetChannel(channel string) *Event

SetChannel - set event channel - mandatory if default channel was not set

func (*Event) SetClientId 

func (e *Event) SetClientId(clientId string) *Event

SetClientId - set event ClientId - mandatory if default client was not set

func (*Event) SetId 

func (e *Event) SetId(id string) *Event

SetId - set event id otherwise new random uuid will be set

func (*Event) SetMetadata 

func (e *Event) SetMetadata(metadata string) *Event

SetMetadata - set event metadata - mandatory if body field was not set

type EventStore 

type EventStore struct {
	Id       string
	Channel  string
	Metadata string
	Body     []byte
	ClientId string
	// contains filtered or unexported fields
}

func (*EventStore) Send 

func (es *EventStore) Send(ctx context.Context) (*EventStoreResult, error)

func (*EventStore) SetBody 

func (es *EventStore) SetBody(body []byte) *EventStore

SetBody - set event store body - mandatory if metadata field was not set

func (*EventStore) SetChannel 

func (es *EventStore) SetChannel(channel string) *EventStore

SetChannel - set event store channel - mandatory if default channel was not set

func (*EventStore) SetClientId 

func (es *EventStore) SetClientId(clientId string) *EventStore

SetClientId - set event store ClientId - mandatory if default client was not set

func (*EventStore) SetId 

func (es *EventStore) SetId(id string) *EventStore

SetId - set event store id otherwise new random uuid will be set

func (*EventStore) SetMetadata 

func (es *EventStore) SetMetadata(metadata string) *EventStore

SetMetadata - set event store metadata - mandatory if body field was not set

type EventStoreReceive 

type EventStoreReceive struct {
	Id        string
	Sequence  uint64
	Timestamp time.Time
	Channel   string
	Metadata  string
	Body      []byte
	ClientId  string
}

type EventStoreResult 

type EventStoreResult struct {
	Id   string
	Sent bool
	Err  error
}

type Option 

type Option interface {
	// contains filtered or unexported methods
}

func WithAddress 

func WithAddress(host string, port int) Option

WithAddress - set host and port address of KubeMQ server

func WithClientId 

func WithClientId(id string) Option

WithClientId - set client id to be used in all functions call with this client - mandatory

func WithCredentials 

func WithCredentials(certFile, serverOverrideDomain string) Option

WithCredentials - set secured TLS credentials from the input certificate file for client. serverNameOverride is for testing only. If set to a non empty string, it will override the virtual host name of authority (e.g. :authority header field) in requests.

func WithDefaultCacheTTL 

func WithDefaultCacheTTL(ttl time.Duration) Option

WithDefaultCacheTTL - set default cache time to live for any query requests with any CacheKey set value

func WithDefualtChannel 

func WithDefualtChannel(channel string) Option

WithDefaultChannel - set default channel for any outbound requests

func WithReceiveBufferSize 

func WithReceiveBufferSize(size int) Option

WithReceiveBufferSize - set length of buffered channel to be set in all subscriptions

func WithToken 

func WithToken(token string) Option

WithToken - set KubeMQ token to be used for KubeMQ connection - not mandatory, only if enforced by the KubeMQ server

func WithTransportType 

func WithTransportType(transportType TransportType) Option

WithDefaultCacheTTL - set default cache time to live for any query requests with any CacheKey set value

type Options 

type Options struct {
	// contains filtered or unexported fields
}

func GetDefaultOptions 

func GetDefaultOptions() *Options

type Query 

type Query struct {
	Id       string
	Channel  string
	Metadata string
	Body     []byte
	Timeout  time.Duration
	ClientId string
	CacheKey string
	CacheTTL time.Duration
	// contains filtered or unexported fields
}

func (*Query) Send 

func (q *Query) Send(ctx context.Context) (*QueryResponse, error)

Send - sending query request , waiting for response or timeout

func (*Query) SetBody 

func (q *Query) SetBody(body []byte) *Query

SetBody - set query body - mandatory if metadata field is empty

func (*Query) SetCacheKey 

func (q *Query) SetCacheKey(cacheKey string) *Query

SetCacheKey - set cache key to retrieve already stored query response, otherwise the response for this query will be stored in cache for future query requests

func (*Query) SetCacheTTL 

func (q *Query) SetCacheTTL(ttl time.Duration) *Query

SetCacheTTL - set cache time to live for the this query cache key response to be retrieved already stored query response, if not set default cacheTTL will be set

func (*Query) SetChannel 

func (q *Query) SetChannel(channel string) *Query

SetChannel - set query channel - mandatory if default channel was not set

func (*Query) SetClientId 

func (q *Query) SetClientId(clientId string) *Query

SetClientId - set query ClientId - mandatory if default client was not set

func (*Query) SetId 

func (q *Query) SetId(id string) *Query

SetId - set query requestId, otherwise new random uuid will be set

func (*Query) SetMetadata 

func (q *Query) SetMetadata(metadata string) *Query

SetMetadata - set query metadata - mandatory if body field is empty

func (*Query) SetTimeout 

func (q *Query) SetTimeout(timeout time.Duration) *Query

SetTimeout - set timeout for query to be returned. if timeout expired , send query will result with an error

type QueryReceive 

type QueryReceive struct {
	Id         string
	Channel    string
	Metadata   string
	Body       []byte
	ResponseTo string
}

type QueryResponse 

type QueryResponse struct {
	QueryId          string
	Executed         bool
	ExecutedAt       time.Time
	Metadata         string
	ResponseClientId string
	Body             []byte
	CacheHit         bool
	Error            string
}

type Response 

type Response struct {
	RequestId  string
	ResponseTo string
	Metadata   string
	Body       []byte
	ClientId   string
	ExecutedAt time.Time
	Err        error
	// contains filtered or unexported fields
}

func (*Response) Send 

func (r *Response) Send(ctx context.Context) error

Send - sending response to command or query request

func (*Response) SetBody 

func (r *Response) SetBody(body []byte) *Response

SetMetadata - set body response, for query only

func (*Response) SetClientId 

func (r *Response) SetClientId(clientId string) *Response

SetClientID - set clientId response, if not set default clientId will be used

func (*Response) SetError 

func (r *Response) SetError(err error) *Response

SetError - set query or command execution error

func (*Response) SetExecutedAt 

func (r *Response) SetExecutedAt(executedAt time.Time) *Response

SetExecutedAt - set query or command execution time

func (*Response) SetMetadata 

func (r *Response) SetMetadata(metadata string) *Response

SetMetadata - set metadata response, for query only

func (*Response) SetRequestId 

func (r *Response) SetRequestId(id string) *Response

SetId - set response corresponded requestId - mandatory

func (*Response) SetResponseTo 

func (r *Response) SetResponseTo(channel string) *Response

SetResponseTo - set response channel as received in CommandReceived or QueryReceived object - mandatory

type SubscriptionOption 

type SubscriptionOption interface {
	// contains filtered or unexported methods
}

func StartFromFirstEvent 

func StartFromFirstEvent() SubscriptionOption

StartFromFirstEvent - replay all the stored events from the first available sequence and continue stream new events from this point

func StartFromLastEvent 

func StartFromLastEvent() SubscriptionOption

StartFromLastEvent - replay last event and continue stream new events from this point

func StartFromNewEvents 

func StartFromNewEvents() SubscriptionOption

StartFromNewEvents - start event store subscription with only new events

func StartFromSequence 

func StartFromSequence(sequence int) SubscriptionOption

StartFromSequence - replay events from specific event sequence number and continue stream new events from this point

func StartFromTime 

func StartFromTime(since time.Time) SubscriptionOption

StartFromTime - replay events from specific time continue stream new events from this point

func StartFromTimeDelta 

func StartFromTimeDelta(delta time.Duration) SubscriptionOption

StartFromTimeDelta - replay events from specific current time - delta duration in seconds, continue stream new events from this point

type Transport 

type Transport interface {
	SendEvent(ctx context.Context, event *Event) error
	StreamEvents(ctx context.Context, eventsCh chan *Event, errCh chan error)
	SubscribeToEvents(ctx context.Context, channel, group string, errCh chan error) (<-chan *Event, error)
	SendEventStore(ctx context.Context, eventStore *EventStore) (*EventStoreResult, error)
	StreamEventsStore(ctx context.Context, eventsCh chan *EventStore, eventsResultCh chan *EventStoreResult, errCh chan error)
	SubscribeToEventsStore(ctx context.Context, channel, group string, errCh chan error, opt SubscriptionOption) (<-chan *EventStoreReceive, error)
	SendCommand(ctx context.Context, command *Command) (*CommandResponse, error)
	SubscribeToCommands(ctx context.Context, channel, group string, errCh chan error) (<-chan *CommandReceive, error)
	SendQuery(ctx context.Context, query *Query) (*QueryResponse, error)
	SubscribeToQueries(ctx context.Context, channel, group string, errCh chan error) (<-chan *QueryReceive, error)
	SendResponse(ctx context.Context, response *Response) error
	Close() error
}

type TransportType 

type TransportType int
const TransportTypeGRPC TransportType = iota

Source Files

View all Source files

Directories

Path Synopsis
examples
command
events
events_store
query
v2
examples/pubsub/events/basic Module
examples/pubsub/events_store/basic Module

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.