source

package
v0.3.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: May 11, 2023 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Index

Constants

View Source
const (
	// ConfigKeyBufferSize is a config name for a buffer size.
	ConfigKeyBufferSize = "bufferSize"
	// ConfigKeyDeliverSubject is a config name for a deliver subject.
	ConfigKeyDeliverSubject = "deliverSubject"
	// ConfigKeyDurable is a config name for a durable name.
	ConfigKeyDurable = "durable"
	// ConfigKeyDeliverPolicy is a config name for a message deliver policy.
	ConfigKeyDeliverPolicy = "deliverPolicy"
	// ConfigKeyAckPolicy is a config name for a message acknowledge policy.
	ConfigKeyAckPolicy = "ackPolicy"
)

Variables

This section is empty.

Functions

func NewSource

func NewSource() sdk.Source

NewSource creates new instance of the Source.

Types

type Config

type Config struct {
	config.Config

	BufferSize int `key:"bufferSize" validate:"omitempty,min=64"`
	// Durable is the name of the Consumer, if set will make a consumer durable,
	// allowing resuming consumption where left off.
	Durable string `key:"durable" validate:"required"`
	// DeliverSubject specifies the JetStream consumer deliver subject.
	DeliverSubject string `json:"deliverSubject" validate:"required"`
	// DeliverPolicy defines where in the stream the connector should start receiving messages.
	DeliverPolicy nats.DeliverPolicy `key:"deliverPolicy" validate:"oneof=0 2"`
	// AckPolicy defines how messages should be acknowledged.
	AckPolicy nats.AckPolicy `key:"ackPolicy" validate:"oneof=0 1 2"`
}

Config holds source specific configurable values.

func Parse

func Parse(cfg map[string]string) (Config, error)

Parse maps the incoming map to the Config and validates it.

type Source

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

Source operates source logic.

func (*Source) Ack

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

Ack acknowledges a message at the given position.

func (*Source) Configure

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

Configure parses and initializes the config.

func (*Source) Open

func (s *Source) Open(_ context.Context, position sdk.Position) error

Open opens a connection to NATS and initializes iterators.

func (*Source) Parameters

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

Parameters is a map of named Parameters that describe how to configure the Source.

func (*Source) Read

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

Read fetches a record from an iterator. If there's no record will return sdk.ErrBackoffRetry.

func (*Source) Teardown

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

Teardown closes connections, stops iterator.

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL