bamboo

package module

Go to main page

Versions in this module

v0

v0.0.3

Oct 9, 2023

Changes in this version

type MetricsEventHandler

+ OnDecrNumRunningWorkers func()

+ OnIncrNumRunningWorkers func()

v0.0.2

Oct 8, 2023

Changes in this version

+ var BambooLoggers map[ContextKey]*slog.Logger

+ var ErrContextCanceled = errors.New("ContextCanceled")

+ var LoggerNameKey = "bamboo_logger_name"

+ var RequestIDKey = "bamboo_request_id"

+ func GetLoggerFromContext(ctx context.Context, key ContextKey) *slog.Logger

+ func Init(ctx context.Context) context.Context

+ func MetricsServerProcess(ctx context.Context, port int, gracefulShutdownTimeSec int) error

+ func WithLoggerName(ctx context.Context, val ContextKey) context.Context

+ func WithValue(ctx context.Context, key ContextKey, val any) context.Context

+ type BambooHeartbeatPublisher interface

+ Ping func(ctx context.Context) error

+ Run func(ctx context.Context, resultChannel string, heartbeatIntervalMSec int, ...) error

+ func NewGoroutineBambooHeartbeatPublisher(pubsubMap GoroutineBambooPubSubMap) BambooHeartbeatPublisher

+ type BambooLogHandler struct

+ func (h *BambooLogHandler) Handle(ctx context.Context, record slog.Record) error

+ func (h *BambooLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler

+ func (h *BambooLogHandler) WithGroup(name string) slog.Handler

+ type BambooRequestConsumer interface

+ Close func(ctx context.Context) error

+ Consume func(ctx context.Context) (*pb.WorkerParameter, error)

+ Ping func(ctx context.Context) error

+ func NewGoroutineBambooRequestConsumer(queue <-chan []byte) BambooRequestConsumer

+ func NewKafkaBambooRequestConsumer(consumerOptions kafka.ReaderConfig, requestWaitTimeout time.Duration) BambooRequestConsumer

+ func NewRedisBambooRequestConsumer(consumerOptions *redis.UniversalOptions, consumerChannel string, ...) BambooRequestConsumer

type BambooRequestProducer

+ func NewGoroutineBambooRequestProducer(ctx context.Context, workerName string, queue chan<- []byte) BambooRequestProducer

+ type BambooResultPublisher interface

+ Ping func(ctx context.Context) error

+ Publish func(ctx context.Context, resultChannel string, result []byte) error

+ func NewGoroutineBambooResultPublisher(pubsubMap GoroutineBambooPubSubMap) BambooResultPublisher

+ func NewRedisBambooResultPublisher(publisherOptions *redis.UniversalOptions) BambooResultPublisher

type BambooResultSubscriber

+ func NewGoroutineBambooResultSubscriber(ctx context.Context, workerName string, pubsubMap GoroutineBambooPubSubMap) BambooResultSubscriber

+ func NewRedisBambooResultSubscriber(ctx context.Context, workerName string, ...) BambooResultSubscriber

+ OpenSubscribeConnection func(ctx context.Context, resultChannel string) (SubscribeFunc, CloseSubscribeConnectionFunc, error)

type BambooWorker

+ func NewBambooWorker(createRequestConsumerFunc CreateBambooRequestConsumerFunc, ...) (BambooWorker, error)

+ type BambooWorkerClient interface

+ Call func(ctx context.Context, heartbeatIntervalMSec int, jobTimeoutMSec int, ...) ([]byte, error)

+ Close func(ctx context.Context)

+ Ping func(ctx context.Context) error

+ func NewBambooWorkerClient(requestProducer BambooRequestProducer, resultSubscriber BambooResultSubscriber) BambooWorkerClient

+ type CloseSubscribeConnectionFunc func(ctx context.Context) error

+ type ContextKey string

+ const BambooHeartbeatPublisherLoggerContextKey

+ const BambooRequestConsumerLoggerContextKey

+ const BambooRequestProducerLoggerContextKey

+ const BambooResultPublisherLoggerContextKey

+ const BambooResultSubscriberLoggerContextKey

+ const BambooWorkerClientLoggerContextKey

+ const BambooWorkerJobLoggerContextKey

+ const BambooWorkerLoggerContextKey

+ const LoggerNameContextKey

+ const RequestIDContextKey

+ type CreateBambooRequestConsumerFunc func(ctx context.Context) BambooRequestConsumer

+ type GoroutineBambooPubSubMap interface

+ ClosePublishChannel func(channelName string) error

+ CloseSubscribeChannel func(channelName string) error

+ CreateChannel func(channelName string) chan []byte

+ GetChannel func(channelName string) (chan []byte, error)

+ func NewGoroutineBambooPubSubMap() GoroutineBambooPubSubMap

+ type MetricsEventHandler interface

+ OnInternalErrorJob func()

+ OnInvalidArgumentJob func()

+ OnReceiveRequest func()

+ OnSuccessJob func()

+ func NewEmptyEventHandler() MetricsEventHandler

+ func NewPrometheusEventHandler() MetricsEventHandler

+ type SubscribeFunc func(ctx context.Context) (*pb.WorkerResponse, error)

+ type WorkerJob interface

+ Run func(ctx context.Context) error

+ func NewWorkerJob(ctx context.Context, carrier propagation.MapCarrier, workerFunc WorkerFunc, ...) WorkerJob

v0.0.1

Sep 24, 2023

Changes in this version

+ var ErrAborted = errors.New("Aborted")

+ var ErrTimedout = errors.New("Timedout")

+ var File_bamboo_proto protoreflect.FileDescriptor

+ var ResponseType_name = map[int32]string

+ var ResponseType_value = map[string]int32

+ type BambooRequestProducer interface

+ Close func(ctx context.Context) error

+ Produce func(ctx context.Context, resultChannel string, heartbeatIntervalSec int, ...) error

+ func NewKafkaBambooRequestProducer(ctx context.Context, workerName string, kafkaWriter *kafka.Writer, ...) BambooRequestProducer

+ func NewRedisBambooRequestProducer(ctx context.Context, workerName string, producerOptions redis.UniversalOptions, ...) BambooRequestProducer

+ type BambooResultSubscriber interface

+ Ping func(ctx context.Context) error

+ Subscribe func(ctx context.Context, resultChannel string, heartbeatIntervalSec int, ...) ([]byte, error)

+ func NewRedisResultSubscriber(ctx context.Context, workerName string, ...) BambooResultSubscriber

+ type BambooWorker interface

+ Run func(ctx context.Context) error

+ func NewKafkaRedisBambooWorker(consumerOptions kafka.ReaderConfig, publisherOptions *redis.UniversalOptions, ...) BambooWorker

+ func NewRedisRedisBambooWorker(consumerOptions *redis.UniversalOptions, consumerChannel string, ...) BambooWorker

+ type ByteArreayResult struct

+ Error error

+ Value []byte

+ type LogConfigFunc func(ctx context.Context, headers map[string]string) context.Context

+ type RedisBambooHeartbeatPublisher interface

+ Run func(ctx context.Context)

+ func NewRedisBambooHeartbeatPublisher(publisherOptions *redis.UniversalOptions, resultChannel string, ...) RedisBambooHeartbeatPublisher

+ type RedisJob interface

+ Run func(ctx context.Context) error

+ func NewRedisJob(ctx context.Context, carrier propagation.MapCarrier, workerFunc WorkerFunc, ...) RedisJob

+ type RedisResultSubscriber struct

+ func (s *RedisResultSubscriber) Ping(ctx context.Context) error

+ func (s *RedisResultSubscriber) Subscribe(ctx context.Context, resultChannel string, heartbeatIntervalSec int, ...) ([]byte, error)

+ type ResponseType int32

+ const ResponseType_DATA

+ const ResponseType_HEARTBEAT

+ func (ResponseType) Descriptor() protoreflect.EnumDescriptor

+ func (ResponseType) EnumDescriptor() ([]byte, []int)

+ func (ResponseType) Type() protoreflect.EnumType

+ func (x ResponseType) Enum() *ResponseType

+ func (x ResponseType) Number() protoreflect.EnumNumber

+ func (x ResponseType) String() string

+ type WorkerFunc func(ctx context.Context, headers map[string]string, data []byte, ...) ([]byte, error)

+ type WorkerParameter struct

+ Carrier map[string]string

+ Data []byte

+ Headers map[string]string

+ HeartbeatIntervalSec int32

+ JobTimeoutSec int32

+ ResultChannel string

+ func (*WorkerParameter) Descriptor() ([]byte, []int)

+ func (*WorkerParameter) ProtoMessage()

+ func (x *WorkerParameter) GetCarrier() map[string]string

+ func (x *WorkerParameter) GetData() []byte

+ func (x *WorkerParameter) GetHeaders() map[string]string

+ func (x *WorkerParameter) GetHeartbeatIntervalSec() int32

+ func (x *WorkerParameter) GetJobTimeoutSec() int32

+ func (x *WorkerParameter) GetResultChannel() string

+ func (x *WorkerParameter) ProtoReflect() protoreflect.Message

+ func (x *WorkerParameter) Reset()

+ func (x *WorkerParameter) String() string

+ type WorkerResponse struct

+ Data []byte

+ Type ResponseType

+ func (*WorkerResponse) Descriptor() ([]byte, []int)

+ func (*WorkerResponse) ProtoMessage()

+ func (x *WorkerResponse) GetData() []byte

+ func (x *WorkerResponse) GetType() ResponseType

+ func (x *WorkerResponse) ProtoReflect() protoreflect.Message

+ func (x *WorkerResponse) Reset()

+ func (x *WorkerResponse) String() string

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL