redis

Documentation

Index

• Constants
• type RedisMessageWrapper
• type RedisPubSubProvider
  • func (i *RedisPubSubProvider) ClaimMessage(topic string, group string, minIdle time.Duration, msgID string) (*redis.XMessage, bool)
  • func (i *RedisPubSubProvider) ClaimMessageLoop(topic string, handler v1alpha2.EventHandler)
  • func (i *RedisPubSubProvider) ClaimWorker(msgID string) bool
  • func (v *RedisPubSubProvider) ID() string
  • func (i *RedisPubSubProvider) Init(config providers.IProviderConfig) error
  • func (i *RedisPubSubProvider) InitWithMap(properties map[string]string) error
  • func (i *RedisPubSubProvider) Publish(topic string, event v1alpha2.Event) error
  • func (i *RedisPubSubProvider) ReleaseWorker(msgID string)
  • func (i *RedisPubSubProvider) ResetIdleTimeLoop(topic string, group string, msgID string, stopCh chan struct{})
  • func (s *RedisPubSubProvider) SetContext(ctx *contexts.ManagerContext)
  • func (i *RedisPubSubProvider) Subscribe(topic string, handler v1alpha2.EventHandler) error
  • func (i *RedisPubSubProvider) WaitForIdleWorkers(msgID string, timeout time.Duration) bool
• type RedisPubSubProviderConfig
  • func RedisPubSubProviderConfigFromMap(properties map[string]string) (RedisPubSubProviderConfig, error)

Constants

const (
	// ResetIdleTimeInterval
	ResetIdleTimeInterval = 5 * time.Second
	// ClaimPendingMessage
	ClaimMessageInterval = 10 * time.Second
	// ClaimPendingMessageIdleTime
	ClaimMessageIdleTime = 30 * time.Second

	DefaultNumberOfWorkers = 20
)

Types

type RedisMessageWrapper

type RedisMessageWrapper struct {
	MessageID string
	Topic     string
	Message   interface{}
	Handler   v1alpha2.EventHandler
}

type RedisPubSubProvider

type RedisPubSubProvider struct {
	Config      RedisPubSubProviderConfig          `json:"config"`
	Subscribers map[string][]v1alpha2.EventHandler `json:"subscribers"`
	Client      *redis.Client
	Ctx         context.Context
	Cancel      context.CancelFunc
	Context     *contexts.ManagerContext
	WorkerLock  *sync.Mutex
	IdleWorkers int
}

func (*RedisPubSubProvider) ClaimMessage

func (i *RedisPubSubProvider) ClaimMessage(topic string, group string, minIdle time.Duration, msgID string) (*redis.XMessage, bool)

func (*RedisPubSubProvider) ClaimMessageLoop

func (i *RedisPubSubProvider) ClaimMessageLoop(topic string, handler v1alpha2.EventHandler)

func (*RedisPubSubProvider) ClaimWorker

func (i *RedisPubSubProvider) ClaimWorker(msgID string) bool

func (*RedisPubSubProvider) ID

func (v *RedisPubSubProvider) ID() string

func (*RedisPubSubProvider) Init

func (i *RedisPubSubProvider) Init(config providers.IProviderConfig) error

func (*RedisPubSubProvider) InitWithMap

func (i *RedisPubSubProvider) InitWithMap(properties map[string]string) error

func (*RedisPubSubProvider) Publish

func (i *RedisPubSubProvider) Publish(topic string, event v1alpha2.Event) error

func (*RedisPubSubProvider) ReleaseWorker

func (i *RedisPubSubProvider) ReleaseWorker(msgID string)

func (*RedisPubSubProvider) ResetIdleTimeLoop

func (i *RedisPubSubProvider) ResetIdleTimeLoop(topic string, group string, msgID string, stopCh chan struct{})

func (*RedisPubSubProvider) SetContext

func (s *RedisPubSubProvider) SetContext(ctx *contexts.ManagerContext)

func (*RedisPubSubProvider) Subscribe

func (i *RedisPubSubProvider) Subscribe(topic string, handler v1alpha2.EventHandler) error

func (*RedisPubSubProvider) WaitForIdleWorkers

func (i *RedisPubSubProvider) WaitForIdleWorkers(msgID string, timeout time.Duration) bool

type RedisPubSubProviderConfig

type RedisPubSubProviderConfig struct {
	Name            string `json:"name"`
	Host            string `json:"host"`
	Password        string `json:"password,omitempty"`
	RequiresTLS     bool   `json:"requiresTLS,omitempty"`
	NumberOfWorkers int    `json:"numberOfWorkers,omitempty"`
	ConsumerID      string `json:"consumerID"`
}

func RedisPubSubProviderConfigFromMap

func RedisPubSubProviderConfigFromMap(properties map[string]string) (RedisPubSubProviderConfig, error)