nats

Documentation

Overview

Package nats provides NATS streaming connectors.

Index

• type JetStreamSink
  • func NewJetStreamSink(config *JetStreamSinkConfig) (*JetStreamSink, error)
  • func (js *JetStreamSink) In() chan<- any
• type JetStreamSinkConfig
• type JetStreamSource
  • func NewJetStreamSource(ctx context.Context, config *JetStreamSourceConfig) (*JetStreamSource, error)
  • func (js *JetStreamSource) Out() <-chan any
  • func (js *JetStreamSource) Via(operator streams.Flow) streams.Flow
• type JetStreamSourceConfig
  • func NewJetStreamSourceConfig(conn *nats.Conn, jetStreamContext nats.JetStreamContext, subject string) *JetStreamSourceConfig
• type StreamingSink
  • func NewStreamingSink(conn stan.Conn, topic string) *StreamingSink
  • func (ns *StreamingSink) In() chan<- any
• type StreamingSource
  • func NewStreamingSource(ctx context.Context, conn stan.Conn, subscriptionType stan.SubscriptionOption, ...) *StreamingSource
  • func (ns *StreamingSource) Out() <-chan any
  • func (ns *StreamingSource) Via(operator streams.Flow) streams.Flow

Types

type JetStreamSink

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

JetStreamSink represents a NATS JetStream sink connector.

func NewJetStreamSink

func NewJetStreamSink(config *JetStreamSinkConfig) (*JetStreamSink, error)

NewJetStreamSink returns a new JetStreamSink connector. The stream for the configured subject is expected to exist.

func (*JetStreamSink) In

func (js *JetStreamSink) In() chan<- any

In returns the input channel of the JetStreamSink connector.

type JetStreamSinkConfig

type JetStreamSinkConfig struct {
	Conn         *nats.Conn
	JetStreamCtx nats.JetStreamContext
	Subject      string
	DrainConn    bool          // Determines whether to drain the connection when the upstream is closed.
	PubOpts      []nats.PubOpt // PubOpt configures options for publishing JetStream messages.
}

JetStreamSinkConfig specifies parameters for the JetStream sink connector.

type JetStreamSource

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

JetStreamSource represents a NATS JetStream source connector.

func NewJetStreamSource

func NewJetStreamSource(ctx context.Context, config *JetStreamSourceConfig) (*JetStreamSource, error)

NewJetStreamSource returns a new JetStreamSource connector. A pull-based subscription is used to consume data from the subject.

func (*JetStreamSource) Out

func (js *JetStreamSource) Out() <-chan any

Out returns the output channel of the JetStreamSource connector.

func (*JetStreamSource) Via

func (js *JetStreamSource) Via(operator streams.Flow) streams.Flow

Via streams data to a specified operator and returns it.

type JetStreamSourceConfig

type JetStreamSourceConfig struct {
	Conn           *nats.Conn
	JetStreamCtx   nats.JetStreamContext
	Subject        string
	ConsumerName   string         // For an ephemeral pull consumer use an empty string.
	FetchBatchSize int            // FetchBatchSize is used by the pull consumer.
	Ack            bool           // Ack determines whether to acknowledge delivered messages by the consumer.
	SubOpts        []nats.SubOpt  // SubOpt configures options for subscribing to JetStream consumers.
	PullOpts       []nats.PullOpt // PullOpt are the options that can be passed when pulling a batch of messages.
	AckOpts        []nats.AckOpt  // AckOpt are the options that can be passed when acknowledge a message.
}

JetStreamSourceConfig specifies parameters for the JetStream source connector. Use NewJetStreamSourceConfig to create a new JetStreamSourceConfig with default values.

func NewJetStreamSourceConfig

func NewJetStreamSourceConfig(conn *nats.Conn, jetStreamContext nats.JetStreamContext,
	subject string) *JetStreamSourceConfig

NewJetStreamSourceConfig returns a new JetStreamSourceConfig with default values.

type StreamingSink

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

StreamingSink represents a NATS Streaming sink connector. Deprecated: Use JetStreamSink instead.

func NewStreamingSink

func NewStreamingSink(conn stan.Conn, topic string) *StreamingSink

NewStreamingSink returns a new StreamingSink connector.

func (*StreamingSink) In

func (ns *StreamingSink) In() chan<- any

In returns the input channel of the StreamingSink connector.

type StreamingSource

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

StreamingSource represents a NATS Streaming source connector. Deprecated: Use JetStreamSource instead.

func NewStreamingSource

func NewStreamingSource(ctx context.Context, conn stan.Conn, subscriptionType stan.SubscriptionOption,
	topics ...string) *StreamingSource

NewStreamingSource returns a new StreamingSource connector.

func (*StreamingSource) Out

func (ns *StreamingSource) Out() <-chan any

Out returns the output channel of the StreamingSource connector.

func (*StreamingSource) Via

func (ns *StreamingSource) Via(operator streams.Flow) streams.Flow

Via streams data to a specified operator and returns it.