Documentation
¶
Overview ¶
Package source defines the Source interface and implementations for reading records into the StreamForge processing pipeline.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DefaultGenerator ¶
DefaultGenerator produces simple records with sequential keys and values.
Types ¶
type ChannelSource ¶
type ChannelSource struct {
// contains filtered or unexported fields
}
ChannelSource reads records from an in-memory channel. Useful for testing.
func NewChannelSource ¶
func NewChannelSource(ch chan *record.Record) *ChannelSource
NewChannelSource creates a source that reads from the given channel.
func (*ChannelSource) Close ¶
func (c *ChannelSource) Close() error
Close marks the source as closed.
func (*ChannelSource) Commit ¶
func (c *ChannelSource) Commit(_ context.Context) error
Commit is a no-op for channel sources.
type GeneratorFunc ¶
GeneratorFunc produces a record given a sequence number.
type GeneratorOption ¶
type GeneratorOption func(*GeneratorSource)
GeneratorOption configures the generator source.
func WithMaxCount ¶
func WithMaxCount(n int64) GeneratorOption
WithMaxCount sets the maximum number of records to generate. A value <= 0 means unlimited.
func WithRate ¶
func WithRate(d time.Duration) GeneratorOption
WithRate sets the time between generated records.
type GeneratorSource ¶
type GeneratorSource struct {
// contains filtered or unexported fields
}
GeneratorSource produces records using a generator function at a configurable rate. Useful for benchmarks and testing.
func NewGeneratorSource ¶
func NewGeneratorSource(fn GeneratorFunc, opts ...GeneratorOption) *GeneratorSource
NewGeneratorSource creates a source that generates records using the given function.
func (*GeneratorSource) Close ¶
func (g *GeneratorSource) Close() error
Close is a no-op for generator sources.
func (*GeneratorSource) Commit ¶
func (g *GeneratorSource) Commit(_ context.Context) error
Commit is a no-op for generator sources.
type KafkaSource ¶
type KafkaSource struct {
// contains filtered or unexported fields
}
KafkaSource reads records from a Kafka topic using segmentio/kafka-go.
func NewKafkaSource ¶
func NewKafkaSource(cfg KafkaSourceConfig) *KafkaSource
NewKafkaSource creates a new Kafka consumer source.
func (*KafkaSource) Commit ¶
func (k *KafkaSource) Commit(ctx context.Context) error
Commit commits the current offset to Kafka.
type KafkaSourceConfig ¶
type KafkaSourceConfig struct {
Brokers []string
Topic string
GroupID string
MaxWait time.Duration
MinBytes int
MaxBytes int
}
KafkaSourceConfig holds configuration for a Kafka source.
type Source ¶
type Source interface {
// Open initializes the source and prepares it for reading.
Open(ctx context.Context) error
// Read returns the next record from the source. It blocks until a record
// is available, the context is cancelled, or an error occurs.
Read(ctx context.Context) (*record.Record, error)
// Commit acknowledges that all records returned so far have been processed.
Commit(ctx context.Context) error
// Close releases resources held by the source.
Close() error
}
Source is the interface for reading records into a topology.
Source Files