streamline

README

Streamline

Streamline is a Golang library designed to efficiently manage subscriptions and optimize the usage of NATS JetStream. This library aims to reduce the number of consumers and connections in a NATS JetStream environment, streamlining the event subscription process.

Features

• Efficient Subscription Management: Easily manage and organize your event subscriptions.
• Reduced Consumer Count: Optimally utilize NATS JetStream resources by minimizing the number of consumers.
• Connection Optimization: Decrease the total number of connections, enhancing overall system performance.
• Easy Integration: Seamlessly integrates with existing NATS JetStream setups.

Installation

To install Streamline, use the go get command:

go get github.com/weedbox/streamline

Contributing

Contributions to the Streamline project are welcome. Please refer to the contributing guidelines for more information.

License

This project is licensed under the Apache License.

Acknowledgments

• This project uses NATS JetStream, an open-source, cloud-native messaging system.
• Thanks to the Golang community for their continuous support.

Documentation

Index

• Variables
• type Agent
• type AgentOption
  • func WithHandler(handler nats.MsgHandler) AgentOption
  • func WithSubjectFilter(subjects []string) AgentOption
• type Option
  • func WithJetStream(js nats.JetStreamContext) Option
• type SubOption
  • func WithConsumer(c string) SubOption
  • func WithDeliverTopic(topic string) SubOption
  • func WithPendingSize(size int) SubOption
  • func WithStream(s string) SubOption
  • func WithSubjects(subjects []string) SubOption
• type Subscription
  • func NewSubscription(name string, js nats.JetStreamContext, opts ...SubOption) *Subscription
  • func (sub *Subscription) Close() error
  • func (sub *Subscription) RemoveAgent(agentName string) error
  • func (sub *Subscription) Subscribe(agentName string, subjects []string, fn nats.MsgHandler) (Agent, error)
• type SubscriptionAgent
  • func NewSubscriptionAgent(agentName string, js nats.JetStreamContext, s *nats.StreamInfo, ...) *SubscriptionAgent
  • func (sa *SubscriptionAgent) Name() string
• type SubscriptionManager
  • func NewSubscriptionManager(opts ...Option) *SubscriptionManager
  • func (sm *SubscriptionManager) CreateSubscription(name string, opts ...SubOption) (*Subscription, error)
  • func (sm *SubscriptionManager) RemoveSubscription(name string) error

Variables

var (
	ErrAgentExists = fmt.Errorf("Agent already exists")
)

Types

type Agent

type Agent interface {
	Name() string
}

type AgentOption

type AgentOption func(*SubscriptionAgent)

func WithHandler

func WithHandler(handler nats.MsgHandler) AgentOption

func WithSubjectFilter

func WithSubjectFilter(subjects []string) AgentOption

type Option

type Option func(*SubscriptionManager)

func WithJetStream

func WithJetStream(js nats.JetStreamContext) Option

type SubOption

type SubOption func(*Subscription)

func WithConsumer

func WithConsumer(c string) SubOption

func WithDeliverTopic

func WithDeliverTopic(topic string) SubOption

func WithPendingSize

func WithPendingSize(size int) SubOption

func WithStream

func WithStream(s string) SubOption

func WithSubjects

func WithSubjects(subjects []string) SubOption

type Subscription

type Subscription struct {
	Name string
	// contains filtered or unexported fields
}

func NewSubscription

func NewSubscription(name string, js nats.JetStreamContext, opts ...SubOption) *Subscription

func (*Subscription) Close

func (sub *Subscription) Close() error

func (*Subscription) RemoveAgent

func (sub *Subscription) RemoveAgent(agentName string) error

func (*Subscription) Subscribe

func (sub *Subscription) Subscribe(agentName string, subjects []string, fn nats.MsgHandler) (Agent, error)

type SubscriptionAgent

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

func NewSubscriptionAgent

func NewSubscriptionAgent(agentName string, js nats.JetStreamContext, s *nats.StreamInfo, c *nats.ConsumerInfo, opts ...AgentOption) *SubscriptionAgent

func (*SubscriptionAgent) Name

func (sa *SubscriptionAgent) Name() string

type SubscriptionManager

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

func NewSubscriptionManager

func NewSubscriptionManager(opts ...Option) *SubscriptionManager

func (*SubscriptionManager) CreateSubscription

func (sm *SubscriptionManager) CreateSubscription(name string, opts ...SubOption) (*Subscription, error)

func (*SubscriptionManager) RemoveSubscription

func (sm *SubscriptionManager) RemoveSubscription(name string) error