services

Jun 28, 2024

Changes in this version

type InhooksConfigService

+ GetTransformDefinition func(transformID string) *models.TransformDefinition

+ type MessageTransformer interface

+ Transform func(ctx context.Context, transformDefinition *models.TransformDefinition, ...) error

+ func NewMessageTransformer(config *lib.TransformConfig) MessageTransformer

Aug 9, 2023

Changes in this version

+ type MessageVerifier interface

+ Verify func(flow *models.Flow, m *models.Message) error

+ func NewMessageVerifier() MessageVerifier

May 15, 2023

Changes in this version

+ type CleanupService interface

+ CleanupDoneQueue func(ctx context.Context, f *models.Flow, sink *models.Sink, ...) (int, error)

+ func NewCleanupService(redisStore RedisStore, timeSvc TimeService) CleanupService

type RedisStore

+ ZRemDel func(ctx context.Context, queueKey string, messageIDs []string, ...) error

+ ZRemRangeBelowScore func(ctx context.Context, queueKey string, maxScore int) (int, error)

May 15, 2023

Changes in this version

+ type ProcessingRecoveryService interface

+ AddToCache func(mID string, ttl time.Duration)

+ MoveProcessingToReady func(ctx context.Context, flow *models.Flow, sink *models.Sink, ...) ([]string, error)

+ func NewProcessingRecoveryService(redisStore RedisStore) (ProcessingRecoveryService, error)

type RedisStore

+ LRangeAll func(ctx context.Context, queueKey string) ([]string, error)

+ LRemRPush func(ctx context.Context, sourceQueueKey, destQueueKey string, messageIDs []string) error

+ type RetryCalculator interface

+ NextAttemptInterval func(attemptsCount int, retryInterval *time.Duration, retryExpMultiplier *float64) time.Duration

+ func NewRetryCalculator() RetryCalculator

May 14, 2023

Changes in this version

+ type InhooksConfigService interface

+ FindFlowForSource func(sourceSlug string) *models.Flow

+ GetFlow func(flowID string) *models.Flow

+ GetFlows func() map[string]*models.Flow

+ Load func(path string) error

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

+ type MessageBuilder interface

+ FromHttp func(flow *models.Flow, r *http.Request, reqID string) ([]*models.Message, error)

+ func NewMessageBuilder(timeSvc TimeService) MessageBuilder

+ type MessageEnqueuer interface

+ Enqueue func(ctx context.Context, messages []*models.Message) error

+ func NewMessageEnqueuer(redisStore RedisStore, timeSvc TimeService) MessageEnqueuer

+ type MessageFetcher interface

+ GetMessageForProcessing func(ctx context.Context, timeout time.Duration, flowID string, sinkID string) (*models.Message, error)

+ func NewMessageFetcher(redisStore RedisStore, timeSvc TimeService) MessageFetcher

+ type MessageProcessor interface

+ Process func(ctx context.Context, sink *models.Sink, m *models.Message) error

+ func NewMessageProcessor(httpClient *http.Client) MessageProcessor

+ type ProcessingResultsService interface

+ HandleFailed func(ctx context.Context, sink *models.Sink, m *models.Message, processingErr error) (models.QueueStatus, error)

+ HandleOK func(ctx context.Context, m *models.Message) error

+ func NewProcessingResultsService(timeSvc TimeService, redisStore RedisStore) ProcessingResultsService

+ type RedisStore interface

+ BLMove func(ctx context.Context, timeout time.Duration, sourceQueueKey string, ...) ([]byte, error)

+ Dequeue func(ctx context.Context, timeout time.Duration, key string) ([]byte, error)

+ Enqueue func(ctx context.Context, key string, value []byte) error

+ Get func(ctx context.Context, messageKey string) ([]byte, error)

+ SetAndEnqueue func(ctx context.Context, messageKey string, value []byte, queueKey string, ...) error

+ SetAndMove func(ctx context.Context, messageKey string, value []byte, ...) error

+ SetAndZAdd func(ctx context.Context, messageKey string, value []byte, queueKey string, ...) error

+ SetLRemZAdd func(ctx context.Context, messageKey string, value []byte, ...) error

+ ZRangeBelowScore func(ctx context.Context, queueKey string, score float64) ([]string, error)

+ ZRemRpush func(ctx context.Context, messageIDs []string, sourceQueueKey string, ...) error

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

+ type SchedulerService interface

+ MoveDueScheduled func(ctx context.Context, f *models.Flow, sink *models.Sink) error

+ func NewSchedulerService(redisStore RedisStore, timeSvc TimeService) SchedulerService

+ type TimeService interface

+ Now func() time.Time

+ func NewTimeService() TimeService