source

Documentation

Overview

Package source contains several implementations of the batch.Source interface for common data source scenarios, including:

- Channel: For using existing channels as batch sources - Error: For simulating error-only sources without data - Nil: For testing timing behavior without emitting data

Each source implementation handles context cancellation properly and ensures channels are closed appropriately.

Basic usage of the Channel source:

input := make(chan interface{}, 2)
input <- "a"
input <- "b"
close(input)

src := &Channel{Input: input}
out, errs := src.Read(context.Background())
for item := range out {
    fmt.Println(item)
}
for range errs {
}

Output:

a
b

Index

• type Channel
  • func (s *Channel) Read(ctx context.Context) (<-chan interface{}, <-chan error)
• type Error
  • func (s *Error) Read(ctx context.Context) (<-chan interface{}, <-chan error)
• type Nil
  • func (s *Nil) Read(ctx context.Context) (<-chan interface{}, <-chan error)

Examples

• Channel

Types

type Channel

type Channel struct {
	// Input is the channel from which this source will read data.
	// The Channel source will not close this channel.
	Input <-chan interface{}
	// BufferSize controls the size of the output buffer (default: 100)
	BufferSize int
}

Channel is a Source that reads from an input channel until it's closed. It simplifies using an existing channel as a data source for batch processing.

Example

package main

import (
	"context"
	"fmt"

	"github.com/MasterOfBinary/gobatch/source"
)

func main() {
	input := make(chan interface{}, 2)
	input <- "a"
	input <- "b"
	close(input)

	src := &source.Channel{Input: input}
	out, errs := src.Read(context.Background())
	for item := range out {
		fmt.Println(item)
	}
	for range errs {
	}
}

Output:

a
b

func (*Channel) Read

func (s *Channel) Read(ctx context.Context) (<-chan interface{}, <-chan error)

Read implements the Source interface by forwarding items from the Input channel to the output channel until Input is closed or context is canceled.

The returned channels are always created (never nil) and always closed properly when the source is done providing data or context is canceled.

type Error

type Error struct {
	// Errs is the channel from which this source will read errors.
	// The Error source will not close this channel.
	Errs <-chan error
	// BufferSize controls the size of the error buffer (default: 10)
	BufferSize int
}

Error is a Source that only emits errors from a channel and provides no data. It is useful for testing error handling in batch processing pipelines and for representing error-only streams.

func (*Error) Read

func (s *Error) Read(ctx context.Context) (<-chan interface{}, <-chan error)

Read implements the Source interface by forwarding errors from the Errs channel to the error channel until Errs is closed or context is canceled.

The returned channels are always created (never nil) and always closed properly when the source is done providing errors or context is canceled. The output channel is always empty, as this source produces only errors.

type Nil

type Nil struct {
	// Duration specifies how long the source will wait before closing its channels.
	// If zero, it will close immediately.
	Duration time.Duration
}

Nil is a Source that does nothing but sleeps for a given duration before closing. It is useful for testing shutdown sequences and empty pipeline behavior.

func (*Nil) Read

func (s *Nil) Read(ctx context.Context) (<-chan interface{}, <-chan error)

Read implements the Source interface by waiting for a specified duration (or until context cancellation) and then closing the channels. It never emits any data or errors.

The returned channels are always created (never nil) and always closed properly when the source completes waiting or context is canceled.