fusion

type Fusion 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 {
	// Payloads can contain one or more message payloads.
	Payloads []Payload `json:"payloads"`

	// Ack will be used to signal an ACK/nACK when message has passed
	// through the pipeline. A no-op value must be set when there is
	// no need for ack. Ack must be idempotent. If acknowledge fails,
	// source is free to re-send the message through normal means.
	// Cause can be set when success=false to send the information
	// about the reason for failure.
	Ack func(success bool, cause error)
}

type NoOpProcessor struct{}

type Options struct {
	Workers int

	// Logger to use for the fusion instance. If not set, no-op logger
	// will be set.
	Logger Logger

	// DrainWithin is the timeout to wait to properly drain the channel
	// and send NACKs for all messages. If not set, stream will not be
	// drained.
	DrainWithin time.Duration
}

type Payload struct {
	Key []byte `json:"key"`
	Val []byte `json:"val"`
}

type Processor interface {
	// Processor can apply some processing to the message and return the result.
	// If the returned message has no payload, fusion will assume end of the
	// pipeline (i.e., a sink) and call the Ack() on the original message.
	Process(ctx context.Context, msg Message) (*Message, error)
}

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

type Source interface {
	// ConsumeFrom should return a channel to which it independently writes
	// the data stream to. It is the responsibility of this Source to close
	// the returned channel once the data is exhausted. goroutines spawned
	// by the source must be tied to the given context and exit when context
	// is cancelled.
	ConsumeFrom(ctx context.Context) (<-chan Message, error)
}

type SourceFunc func(ctx context.Context) (*Message, error)