fusion

type Actor struct {
	Logger
	// contains filtered or unexported fields
}

type Backoff interface {
	// RetryAfter should return the time duration which should be
	// elapsed before the next queueForRetry.
	RetryAfter(msg Message) time.Duration
}

type DelayQueue interface {
	// Enqueue must save the message with priority based on the timestamp set.
	// If no timestamp is set, current timestamp should be assumed.
	Enqueue(msg Message) error

	// Dequeue should read one message that has an expired timestamp and call
	// readFn with it. Success/failure from readFn must be considered as ACK
	// or nACK respectively. When message is not available, Dequeue should not
	// block but return ErrNoMessage. Queue can return EOF to indicate that the
	// queue is fully drained. Other errors from the queue will be logged and
	// ignored.
	Dequeue(ctx context.Context, readFn ReadFn) error
}

type InMemQ struct {
	// contains filtered or unexported fields
}

type LineStream struct {
	From   io.Reader // From is the reader to use.
	Offset int       // Offset to start at.
	Size   int       // Number of lines to stream.
	// contains filtered or unexported fields
}

type Logger interface {
	Debugf(msg string, args ...interface{})
	Infof(msg string, args ...interface{})
	Warnf(msg string, args ...interface{})
	Errorf(msg string, args ...interface{})
}

type Message struct {
	Key []byte `json:"key" xml:"key"`
	Val []byte `json:"val" xml:"val"`

	// Time at which message arrived or should be processed when scheduled by
	// retrying logic. (Managed by the actor)
	Time time.Time `json:"time" xml:"time"`

	// Attempts is incremented by the actor every time an attempt is done to
	// process the message.
	Attempts int `json:"attempts" xml:"attempts"`
}

type Options struct {
	// Stream is the primary source of messages for the actor. Worker
	// threads read from stream continuously and process the messages.
	// If stream is not set, actor will rely entirely on the queue and
	// manually enqueued messages using actor.Enqueue().
	Stream Stream

	// Queue is used for retries and for manually enqueuing messages
	// for the actor. If queue is not set, an in-memory queue will be
	// used.
	Queue DelayQueue

	// Processor function to use for processing the messages. If not
	// set, a no-op processor will be used that skips everything.
	Processor Processor

	// Workers is the number of worker goroutines to spawn when actor
	// starts. Defaults to 1.
	Workers int

	// MaxRetries is the number of retry attempts allowed. If not set,
	// retries are disabled.
	MaxRetries int

	// Backoff strategy to be used for retries. Messages that need to
	// be retried are re-queued to a future time based on the delay
	// returned by backoff. If not set, retries are disabled.
	Backoff Backoff

	// OnFailure if set, will be called when all retries are exhausted
	// and the processing has not succeeded. If not set, such messages
	// will simply be logged and ignored. err argument will have error
	// that occurred in processor in the last retry attempt. If this
	// returns error, message will not be committed to the stream/queue
	// where it was read from.
	OnFailure func(msg Message, err error) error

	// Logger can be overridden to use custom logger.
	Logger Logger

	// PollInterval to use when non-blocking queues/streams return ErrNoMessage.
	// Defaults to 300ms.
	PollInterval time.Duration
}

type Processor func(ctx context.Context, msg Message) error

type ReadFn func(ctx context.Context, msg Message) error

type Stream interface {
	// Read should read the next message available in  the stream and
	// call readFn with it. Success/Failure of the readFn invocation
	// should be used as ACK/NACK respectively.
	Read(ctx context.Context, readFn ReadFn) error
}