benthos

README

Conduit Connector Template

This is a template project for building Conduit connectors in Go.

How to use

Clone this repo and implement the various relevant methods.

Specification

The spec.go file provides a programmatic representation of the configuration options. This is used by the Conduit server to validate configuration and dynamically display configuration options to end users.

How to build?

Run make build to build the connector.

Testing

Run make test to run all the unit tests. Run make test-integration to run the integration tests.

The Docker compose file at test/docker-compose.yml can be used to quickly start a Kafka instance.

Configuration

There are 2 types of "configs", general configs (that apply to both Sources and Destinations) and Source/Destination specific configs.

General configs should be added to config/config.go whereas any source or destination specific configs should be added to source/config.go and destination/config.go respectively

Documentation

Index

• Constants
• Variables
• func NewDestination() sdk.Destination
• func NewSource() sdk.Source
• func Specification() sdk.Specification
• type Config
• type Destination
  • func (d *Destination) Close(ctx context.Context) error
  • func (d *Destination) Configure(ctx context.Context, cfg map[string]string) error
  • func (d *Destination) Connect(ctx context.Context) error
  • func (d *Destination) Open(ctx context.Context) error
  • func (d *Destination) Parameters() map[string]sdk.Parameter
  • func (d *Destination) Read(ctx context.Context) (*service.Message, service.AckFunc, error)
  • func (d *Destination) Teardown(ctx context.Context) error
  • func (d *Destination) Write(ctx context.Context, records []sdk.Record) (int, error)
• type DestinationConfig
  • func ParseDestinationConfig(cfg map[string]string) (DestinationConfig, error)
• type Source
  • func (s *Source) Ack(ctx context.Context, position sdk.Position) error
  • func (s *Source) Close(ctx context.Context) error
  • func (s *Source) Configure(ctx context.Context, cfg map[string]string) error
  • func (s *Source) Connect(ctx context.Context) error
  • func (s *Source) Open(ctx context.Context, pos sdk.Position) error
  • func (s *Source) Parameters() map[string]sdk.Parameter
  • func (s *Source) Read(ctx context.Context) (sdk.Record, error)
  • func (s *Source) Teardown(ctx context.Context) error
  • func (s *Source) Write(ctx context.Context, message *service.Message) error
• type SourceConfig
  • func ParseSourceConfig(cfg map[string]string) (SourceConfig, error)

Constants

const (
	BenthosYaml = "benthos.yaml"
)

Variables

var Connector = sdk.Connector{
	NewSpecification: Specification,
	NewSource:        NewSource,
	NewDestination:   NewDestination,
}

var (
	ErrEmptyConfig = errors.New("missing or empty config")
)

var Required = []string{BenthosYaml}

Functions

func NewDestination

func NewDestination() sdk.Destination

func NewSource

func NewSource() sdk.Source

func Specification

func Specification() sdk.Specification

Specification returns the connector's specification.

Types

type Config

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

type Destination

type Destination struct {
	sdk.UnimplementedDestination
	// contains filtered or unexported fields
}

func (*Destination) Close

func (d *Destination) Close(ctx context.Context) error

func (*Destination) Configure

func (d *Destination) Configure(ctx context.Context, cfg map[string]string) error

func (*Destination) Connect

func (d *Destination) Connect(ctx context.Context) error

func (*Destination) Open

func (d *Destination) Open(ctx context.Context) error

func (*Destination) Parameters

func (d *Destination) Parameters() map[string]sdk.Parameter

func (*Destination) Read

func (d *Destination) Read(ctx context.Context) (*service.Message, service.AckFunc, error)

func (*Destination) Teardown

func (d *Destination) Teardown(ctx context.Context) error

func (*Destination) Write

func (d *Destination) Write(ctx context.Context, records []sdk.Record) (int, error)

type DestinationConfig

type DestinationConfig struct {
	Config
	// contains filtered or unexported fields
}

func ParseDestinationConfig

func ParseDestinationConfig(cfg map[string]string) (DestinationConfig, error)

type Source

type Source struct {
	sdk.UnimplementedSource
	// contains filtered or unexported fields
}

func (*Source) Ack

func (s *Source) Ack(ctx context.Context, position sdk.Position) error

func (*Source) Close

func (s *Source) Close(ctx context.Context) error

func (*Source) Configure

func (s *Source) Configure(ctx context.Context, cfg map[string]string) error

func (*Source) Connect

func (s *Source) Connect(ctx context.Context) error

func (*Source) Open

func (s *Source) Open(ctx context.Context, pos sdk.Position) error

func (*Source) Parameters

func (s *Source) Parameters() map[string]sdk.Parameter

func (*Source) Read

func (s *Source) Read(ctx context.Context) (sdk.Record, error)

func (*Source) Teardown

func (s *Source) Teardown(ctx context.Context) error

func (*Source) Write

func (s *Source) Write(ctx context.Context, message *service.Message) error

type SourceConfig

type SourceConfig struct {
	Config
	// contains filtered or unexported fields
}

func ParseSourceConfig

func ParseSourceConfig(cfg map[string]string) (SourceConfig, error)