services

Documentation

Index

• type InhooksConfigService
  • func NewInhooksConfigService(logger *zap.Logger, appConf *lib.AppConfig) InhooksConfigService
• type MessageBuilder
  • func NewMessageBuilder(timeSvc TimeService) MessageBuilder
• type MessageEnqueuer
  • func NewMessageEnqueuer(redisStore RedisStore, timeSvc TimeService) MessageEnqueuer
• type MessageFetcher
  • func NewMessageFetcher(redisStore RedisStore, timeSvc TimeService) MessageFetcher
• type MessageProcessor
  • func NewMessageProcessor(httpClient *http.Client) MessageProcessor
• type ProcessingResultsService
  • func NewProcessingResultsService(timeSvc TimeService, redisStore RedisStore) ProcessingResultsService
• type RedisStore
  • func NewRedisStore(client *redis.Client, inhooksDBName string) (RedisStore, error)
• type SchedulerService
  • func NewSchedulerService(redisStore RedisStore, timeSvc TimeService) SchedulerService
• type TimeService
  • func NewTimeService() TimeService

Types

type InhooksConfigService

type InhooksConfigService interface {
	Load(path string) error
	FindFlowForSource(sourceSlug string) *models.Flow
	GetFlow(flowID string) *models.Flow
	GetFlows() map[string]*models.Flow
}

func NewInhooksConfigService

func NewInhooksConfigService(logger *zap.Logger, appConf *lib.AppConfig) InhooksConfigService

type MessageBuilder

type MessageBuilder interface {
	FromHttp(flow *models.Flow, r *http.Request, reqID string) ([]*models.Message, error)
}

func NewMessageBuilder

func NewMessageBuilder(timeSvc TimeService) MessageBuilder

type MessageEnqueuer

type MessageEnqueuer interface {
	Enqueue(ctx context.Context, messages []*models.Message) error
}

func NewMessageEnqueuer

func NewMessageEnqueuer(redisStore RedisStore, timeSvc TimeService) MessageEnqueuer

type MessageFetcher

type MessageFetcher interface {
	GetMessageForProcessing(ctx context.Context, timeout time.Duration, flowID string, sinkID string) (*models.Message, error)
}

func NewMessageFetcher

func NewMessageFetcher(redisStore RedisStore, timeSvc TimeService) MessageFetcher

type MessageProcessor

type MessageProcessor interface {
	Process(ctx context.Context, sink *models.Sink, m *models.Message) error
}

func NewMessageProcessor

func NewMessageProcessor(httpClient *http.Client) MessageProcessor

type ProcessingResultsService

type ProcessingResultsService interface {
	HandleFailed(ctx context.Context, sink *models.Sink, m *models.Message, processingErr error) (models.QueueStatus, error)
	HandleOK(ctx context.Context, m *models.Message) error
}

func NewProcessingResultsService

func NewProcessingResultsService(timeSvc TimeService, redisStore RedisStore) ProcessingResultsService

type RedisStore

type RedisStore interface {
	Get(ctx context.Context, messageKey string) ([]byte, error)
	SetAndEnqueue(ctx context.Context, messageKey string, value []byte, queueKey string, messageID string) error
	SetAndZAdd(ctx context.Context, messageKey string, value []byte, queueKey string, messageID string, score float64) error
	SetAndMove(ctx context.Context, messageKey string, value []byte, sourceQueueKey, destQueueKey string, messageID string) error
	SetLRemZAdd(ctx context.Context, messageKey string, value []byte, sourceQueueKey, destQueueKey string, messageID string, score float64) error
	Enqueue(ctx context.Context, key string, value []byte) error
	Dequeue(ctx context.Context, timeout time.Duration, key string) ([]byte, error)
	BLMove(ctx context.Context, timeout time.Duration, sourceQueueKey string, destQueueKey string) ([]byte, error)
	ZRangeBelowScore(ctx context.Context, queueKey string, score float64) ([]string, error)
	ZRemRpush(ctx context.Context, messageIDs []string, sourceQueueKey string, destQueueKey string) error
}

func NewRedisStore

func NewRedisStore(client *redis.Client, inhooksDBName string) (RedisStore, error)

type SchedulerService

type SchedulerService interface {
	MoveDueScheduled(ctx context.Context, f *models.Flow, sink *models.Sink) error
}

func NewSchedulerService

func NewSchedulerService(redisStore RedisStore, timeSvc TimeService) SchedulerService

type TimeService

type TimeService interface {
	Now() time.Time
}

func NewTimeService

func NewTimeService() TimeService