internal

package

Go to main page

Versions in this module

v0

v0.3.1

Nov 29, 2024

Changes in this version

+ var GlobalMessageDelegateHelper MessageDelegateHelper = MessageDelegateHelper

+ var GlobalNoopMessageDelegate redis.MessageDelegate = NoopMessageDelegate(0)

+ var GlobalTracerManager *TracerManager

+ func IsMessageHandler(rv reflect.Value) bool

+ func IsMessageHandlerType(rt reflect.Type) bool

+ func StopRecursiveForwardMessageHandler(ctx *Context, msg *Message)

type Context

+ func (c *Context) CatchErr(err error)

+ func (c *Context) IsRecordingLog() bool

+ func (c *Context) RecordingLog(v bool)

+ func (c *Context) Status() StatusCode

type ContextHelper

+ func (ContextHelper) InjectReplyCodeSafe(ctx *Context, reply ReplyCode)

+ type MessageDelegateHelper struct

+ func (MessageDelegateHelper) IsRestricted(msg *Message) (bool, error)

+ func (MessageDelegateHelper) Restrict(msg *Message) error

+ func (MessageDelegateHelper) Unrestrict(msg *Message) error

+ type MessageHandleProc func(ctx *Context, message *Message)

+ func (proc MessageHandleProc) ProcessMessage(ctx *Context, message *Message)

type MessageHandler

+ func AsMessageHandler(rv reflect.Value) MessageHandler

+ type NoopMessageDelegate int

+ func (n NoopMessageDelegate) OnAck(msg *redis.Message)

+ func (n NoopMessageDelegate) OnDel(msg *redis.Message)

+ type StatusCode = ReplyCode

type TracerManager

+ func (m *TracerManager) UndefinedTracer() *trace.SeverityTracer

v0.3.0

May 14, 2024

Changes in this version

type Context

+ func (c *Context) Pause(streams ...string) error

+ func (c *Context) Resume(streams ...string) error

v0.2.2

Sep 7, 2023

Changes in this version

type RouteComponent

+ StreamSetting *StreamSetting

type Router

+ func (r Router) GetRouteComponent(stream string) RouteComponent

+ type StreamSetting struct

+ MessageStateKeyPrefix *string

+ func (s *StreamSetting) DecodeMessageContentOption() []redis.DecodeMessageContentOption

v0.2.2-alpha.20240430101832

Apr 30, 2024

v0.2.1

Aug 4, 2023

v0.2.0

Aug 4, 2023

Changes in this version

+ const LOGGER_PREFIX

+ const REDIS_BUSYGROUP_PREFIX

+ const RestrictedForwardMessage_InvalidOperation

+ const RestrictedForwardMessage_Recursive

+ var GlobalContextHelper ContextHelper = ContextHelper

+ var GlobalMessageHelper MessageHelper = MessageHelper

+ var GlobalRestrictedMessageDelegate redis.MessageDelegate = RestrictedMessageDelegate(0)

+ var RedisWorkerLogger *log.Logger = log.New(os.Stdout, LOGGER_PREFIX, log.LstdFlags | log.Lmsgprefix)

+ var RedisWorkerModuleInstance = RedisWorkerModule

+ func SetTracerManager(v *TracerManager)

+ type CompositeMessageObserver []MessageObserver

+ func (o CompositeMessageObserver) OnAck(ctx *Context, message *Message)

+ func (o CompositeMessageObserver) OnDel(ctx *Context, message *Message)

+ func (o CompositeMessageObserver) Type() reflect.Type

+ type Context struct

+ ConsumerGroup string

+ ConsumerName string

+ func (*Context) Deadline() (deadline time.Time, ok bool)

+ func (*Context) Done() <-chan struct{}

+ func (*Context) Err() error

+ func (c *Context) InvalidMessage(message *Message)

+ func (c *Context) Logger() *log.Logger

+ func (c *Context) SetValue(key interface{}, value interface{})

+ func (c *Context) Value(key any) any

+ type ContextBuilder struct

+ func NewContextBuilder() *ContextBuilder

+ func (b *ContextBuilder) Build() *Context

+ func (b *ContextBuilder) ConsumerGroup(v string) *ContextBuilder

+ func (b *ContextBuilder) ConsumerName(v string) *ContextBuilder

+ func (b *ContextBuilder) Context(v context.Context) *ContextBuilder

+ func (b *ContextBuilder) InvalidMessageHandler(v MessageHandler) *ContextBuilder

+ func (b *ContextBuilder) Logger(v *log.Logger) *ContextBuilder

+ type ContextHelper struct

+ func (ContextHelper) ExtractReplyCode(ctx *Context) ReplyCode

+ func (ContextHelper) InjectReplyCode(ctx *Context, reply ReplyCode)

+ type ContextMessageDelegate struct

+ func NewContextMessageDelegate(ctx *Context) *ContextMessageDelegate

+ func (d *ContextMessageDelegate) OnAck(msg *redis.Message)

+ func (d *ContextMessageDelegate) OnDel(msg *redis.Message)

+ type Defer struct

+ func (d *Defer) Do(do func(f Finalizer))

+ type ErrorHandler func(ctx *Context, message *Message, err interface{})

+ type Finalizer struct

+ func (f Finalizer) Add(actions ...func(err interface{}))

+ type Message = redis.Message

+ type MessageDispatcher struct

+ ErrorHandler ErrorHandler

+ InvalidMessageHandler MessageHandler

+ MessageHandleService *MessageHandleService

+ MessageObserverService *MessageObserverService

+ MessageTracerService *MessageTracerService

+ OnHostErrorProc OnHostErrorHandler

+ Router Router

+ StreamSet map[string]StreamOffset

+ func (d *MessageDispatcher) ProcessMessage(ctx *Context, message *Message)

+ func (d *MessageDispatcher) StreamOffsets() []StreamOffset

+ func (d *MessageDispatcher) Streams() []string

+ type MessageHandleModule interface

+ CanSetSuccessor func() bool

+ OnInitComplete func()

+ OnStart func(ctx context.Context) error

+ OnStop func(ctx context.Context) error

+ ProcessMessage func(ctx *Context, message *Message, state ProcessingState, recover *Recover)

+ SetSuccessor func(successor MessageHandleModule)

+ type MessageHandleService struct

+ func NewMessageHandleService() *MessageHandleService

+ func (s *MessageHandleService) ProcessMessage(ctx *Context, message *Message, state ProcessingState, recover *Recover)

+ func (s *MessageHandleService) Register(module MessageHandleModule)

+ type MessageHandler interface

+ ProcessMessage func(ctx *Context, message *Message)

+ type MessageHelper struct

+ func (MessageHelper) IsRestricted(msg *Message) (bool, error)

+ func (MessageHelper) Restrict(msg *Message) error

+ func (MessageHelper) Unrestrict(msg *Message) error

+ type MessageObserver interface

+ OnAck func(ctx *Context, message *Message)

+ OnDel func(ctx *Context, message *Message)

+ Type func() reflect.Type

+ type MessageObserverAffair interface

+ MessageObserverTypes func() []reflect.Type

+ type MessageObserverService struct

+ MessageObservers map[reflect.Type]MessageObserver

+ func (s *MessageObserverService) RegisterMessageObservers(msg *Message, handlerID string)

+ func (s *MessageObserverService) UnregisterAllMessageObservers(msg *Message)

+ type MessageTracerService struct

+ Enabled bool

+ InvalidMessageHandlerComponentID string

+ TracerManager *TracerManager

+ func NewMessageTracerService() *MessageTracerService

+ func (s *MessageTracerService) TextMapPropagator() propagation.TextMapPropagator

+ func (s *MessageTracerService) Tracer(id string) *trace.SeverityTracer

+ type OnHostErrorHandler func(err error) (disposed bool)

+ type ProcessingState struct

+ Span *trace.SeveritySpan

+ Stream string

+ Tracer *trace.SeverityTracer

+ type Recover struct

+ func (s *Recover) Defer(catch func(err interface{})) *Defer

+ type RedisWorker struct

+ AllowCreateGroup bool

+ ClaimMinIdleTime time.Duration

+ ClaimOccurrenceRate int32

+ ClaimSensitivity int

+ ConsumerGroup string

+ ConsumerName string

+ IdlingTimeout time.Duration

+ MaxInFlight int64

+ MaxPollingTimeout time.Duration

+ RedisOption *redis.UniversalOptions

+ func (w *RedisWorker) Logger() *log.Logger

+ func (w *RedisWorker) Start(ctx context.Context)

+ func (w *RedisWorker) Stop(ctx context.Context) error

+ type RedisWorkerModule struct

+ func (RedisWorkerModule) ConfigureLogger(logflags int, w io.Writer)

+ func (RedisWorkerModule) DescribeHostType() reflect.Type

+ func (RedisWorkerModule) Init(h host.Host, app *host.AppModule)

+ func (RedisWorkerModule) InitComplete(h host.Host, ctx *host.AppModule)

+ type RedisWorkerRegistrar struct

+ func NewRedisWorkerRegistrar(worker *RedisWorker) *RedisWorkerRegistrar

+ func (r *RedisWorkerRegistrar) AddRouter(stream string, handler MessageHandler, handlerComponentID string)

+ func (r *RedisWorkerRegistrar) EnableTracer(enabled bool)

+ func (r *RedisWorkerRegistrar) RegisterMessageHandleModule(module MessageHandleModule)

+ func (r *RedisWorkerRegistrar) RegisterMessageObserver(v MessageObserver)

+ func (r *RedisWorkerRegistrar) RegisterStream(streamOffset redis.StreamOffset)

+ func (r *RedisWorkerRegistrar) SetErrorHandler(handler ErrorHandler)

+ func (r *RedisWorkerRegistrar) SetInvalidMessageHandler(handler MessageHandler)

+ func (r *RedisWorkerRegistrar) SetMessageManager(messageManager interface{})

+ type ReplyCode int

+ const ABORT

+ const FAIL

+ const INVALID

+ const PASS

+ const UNSET

+ func (code ReplyCode) String() string

+ type RestrictedForwardMessageHandler int

+ func (h RestrictedForwardMessageHandler) ProcessMessage(ctx *Context, message *Message)

+ type RestrictedMessageDelegate int

+ func (RestrictedMessageDelegate) OnAck(*redis.Message)

+ func (RestrictedMessageDelegate) OnDel(*redis.Message)

+ type RestrictedOperationError string

+ func (e RestrictedOperationError) Error() string

+ type RouteComponent struct

+ HandlerComponentID string

+ MessageHandler MessageHandler

+ type Router map[string]RouteComponent

+ func (r Router) Add(stream string, handler MessageHandler, handlerComponentID string)

+ func (r Router) FindHandlerComponentID(stream string) string

+ func (r Router) Get(stream string) MessageHandler

+ func (r Router) Has(stream string) bool

+ func (r Router) Remove(stream string)

+ type StdInvalidMessageHandler struct

+ func (h *StdInvalidMessageHandler) ProcessMessage(ctx *Context, message *Message)

+ type StdMessageHandleModule struct

+ func NewStdMessageHandleModule(dispatcher *MessageDispatcher) *StdMessageHandleModule

+ func (*StdMessageHandleModule) CanSetSuccessor() bool

+ func (*StdMessageHandleModule) OnInitComplete()

+ func (*StdMessageHandleModule) OnStart(ctx context.Context) error

+ func (*StdMessageHandleModule) OnStop(ctx context.Context) error

+ func (*StdMessageHandleModule) SetSuccessor(successor MessageHandleModule)

+ func (m *StdMessageHandleModule) ProcessMessage(ctx *Context, message *Message, state ProcessingState, recover *Recover)

+ type StopError struct

+ func (e *StopError) Error() string

+ func (e *StopError) Unwrap() error

+ type StreamOffset = redis.StreamOffset

+ type StreamOffsetInfo = redis.StreamOffsetInfo

+ type TracerManager struct

+ TextMapPropagator propagation.TextMapPropagator

+ TracerProvider *trace.SeverityTracerProvider

+ func GetTracerManager() *TracerManager

+ func NewTraceManager() *TracerManager

+ func (m *TracerManager) GenerateManagedTracer(v interface{}) *trace.SeverityTracer

+ type UniversalClient = redis.UniversalClient

+ type UniversalOptions = redis.UniversalOptions

+ type XMessage = redis.XMessage

+ type XStream = redis.XStream

v0.1.1

Oct 2, 2022

v0.1.0

Jun 23, 2022

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