client

Documentation

Index

• Constants
• type Client
  • func NewGrpcClient(address string) Client
• type GrpcClient
  • func (c GrpcClient) Get(ctx context.Context, id string) chan events.LogResponse
• type LogResponse
• type NatsLogStream
  • func (c NatsLogStream) Get(ctx context.Context) (chan events.LogResponse, error)
  • func (c NatsLogStream) Init(ctx context.Context) (StreamMetadata, error)
  • func (c NatsLogStream) Push(ctx context.Context, chunk events.Log) error
  • func (c NatsLogStream) PushBytes(ctx context.Context, chunk []byte) error
  • func (c NatsLogStream) Start(ctx context.Context) (resp StreamResponse, err error)
  • func (c NatsLogStream) Stop(ctx context.Context) (resp StreamResponse, err error)
• type Stream
  • func NewNatsLogStream(nc *nats.Conn, id string) (Stream, error)
• type StreamConfigurer
• type StreamGetter
• type StreamInitializer
• type StreamMetadata
• type StreamPusher
• type StreamResponse
• type StreamTrigger

Constants

const (
	StreamPrefix = "log"

	StartSubject = "events.logs.start"
	StopSubject  = "events.logs.stop"
)

Types

type Client

type Client interface {
	Get(ctx context.Context, id string) chan events.LogResponse
}

func NewGrpcClient

func NewGrpcClient(address string) Client

type GrpcClient

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

func (GrpcClient) Get

func (c GrpcClient) Get(ctx context.Context, id string) chan events.LogResponse

Get returns channel with log stream chunks for given execution id connects through GRPC to log service

type LogResponse

type LogResponse struct {
	Log   events.Log
	Error error
}

type NatsLogStream

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

func (NatsLogStream) Get

func (c NatsLogStream) Get(ctx context.Context) (chan events.LogResponse, error)

Get returns channel with log stream chunks for given execution id connects through GRPC to log service

func (NatsLogStream) Init

func (c NatsLogStream) Init(ctx context.Context) (StreamMetadata, error)

func (NatsLogStream) Push

func (c NatsLogStream) Push(ctx context.Context, chunk events.Log) error

Push log chunk to NATS stream

func (NatsLogStream) PushBytes

func (c NatsLogStream) PushBytes(ctx context.Context, chunk []byte) error

Push log chunk to NATS stream TODO handle message repeat with backoff strategy on error

func (NatsLogStream) Start

func (c NatsLogStream) Start(ctx context.Context) (resp StreamResponse, err error)

Start emits start event to the stream - logs service will handle start and create new stream

func (NatsLogStream) Stop

func (c NatsLogStream) Stop(ctx context.Context) (resp StreamResponse, err error)

Stop emits stop event to the stream and waits for given stream to be stopped fully - logs service will handle stop and close stream and all subscribers

type Stream

type Stream interface {
	StreamInitializer
	StreamPusher
	StreamTrigger
	StreamGetter
}

func NewNatsLogStream

func NewNatsLogStream(nc *nats.Conn, id string) (Stream, error)

type StreamConfigurer

type StreamConfigurer interface {
	// Init creates or updates stream on demand
	WithAddress(address string) Stream
}

type StreamGetter

type StreamGetter interface {
	// Init creates or updates stream on demand
	Get(ctx context.Context) (chan events.LogResponse, error)
}

LogStream is a single log stream chunk with possible errors

type StreamInitializer

type StreamInitializer interface {
	// Init creates or updates stream on demand
	Init(ctx context.Context) (meta StreamMetadata, err error)
}

type StreamMetadata

type StreamMetadata struct {
	Name string
}

type StreamPusher

type StreamPusher interface {
	// Push sends logs to log stream
	Push(ctx context.Context, chunk events.Log) error
	// PushBytes sends RAW bytes to log stream, developer is responsible for marshaling valid data
	PushBytes(ctx context.Context, chunk []byte) error
}

type StreamResponse

type StreamResponse struct {
	Message []byte
	Error   bool
}

type StreamTrigger

type StreamTrigger interface {
	// Trigger start event
	Start(ctx context.Context) (StreamResponse, error)
	// Trigger stop event
	Stop(ctx context.Context) (StreamResponse, error)
}