nsq

Documentation

Index

• type GobMarshaler
  • func (GobMarshaler) Marshal(topic string, msg *message.Message) ([]byte, error)
  • func (GobMarshaler) Unmarshal(nsqMsg *nsq.Message) (*message.Message, error)
• type Marshaler
• type MarshalerUnmarshaler
• type NsqPublisher
  • func NewNsqPublisher(config NsqPublisherConfig, logger watermill.LoggerAdapter) (*NsqPublisher, error)
  • func (p NsqPublisher) Close() error
  • func (p NsqPublisher) Publish(topic string, messages ...*message.Message) error
• type NsqPublisherConfig
  • func (c NsqPublisherConfig) Validate() error
• type NsqSubscriber
  • func NewNsqSubscriber(config NsqSubscriberConfig, logger watermill.LoggerAdapter) (*NsqSubscriber, error)
  • func (s *NsqSubscriber) Close() error
  • func (s *NsqSubscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
• type NsqSubscriberConfig
  • func (c *NsqSubscriberConfig) Validate() error
• type Unmarshaler

Types

type GobMarshaler

type GobMarshaler struct{}

GobMarshaler is marshaller which is using Gob to marshal Watermill messages.

func (GobMarshaler) Marshal

func (GobMarshaler) Marshal(topic string, msg *message.Message) ([]byte, error)

func (GobMarshaler) Unmarshal

func (GobMarshaler) Unmarshal(nsqMsg *nsq.Message) (*message.Message, error)

type Marshaler

type Marshaler interface {
	Marshal(topic string, msg *message.Message) ([]byte, error)
}

type MarshalerUnmarshaler

type MarshalerUnmarshaler interface {
	Marshaler
	Unmarshaler
}

type NsqPublisher

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

func NewNsqPublisher

func NewNsqPublisher(config NsqPublisherConfig, logger watermill.LoggerAdapter) (*NsqPublisher, error)

NewNsqPublisher creates a new NsqPublisher.

func (NsqPublisher) Close

func (p NsqPublisher) Close() error

func (NsqPublisher) Publish

func (p NsqPublisher) Publish(topic string, messages ...*message.Message) error

Publish publishes message to NATS.

Publish will not return until an ack has been received from NATS Streaming. When one of messages delivery fails - function is interrupted.

type NsqPublisherConfig

type NsqPublisherConfig struct {
	NsqD string

	// nsq config
	*stdnsq.Config

	// Marshaler is marshaler used to marshal messages to stan format.
	Marshaler Marshaler
}

func (NsqPublisherConfig) Validate

func (c NsqPublisherConfig) Validate() error

type NsqSubscriber

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

func NewNsqSubscriber

func NewNsqSubscriber(config NsqSubscriberConfig, logger watermill.LoggerAdapter) (*NsqSubscriber, error)

func (*NsqSubscriber) Close

func (s *NsqSubscriber) Close() error

func (*NsqSubscriber) Subscribe

func (s *NsqSubscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

Subscribe subscribes messages from Nsq topic.

context.WithValue(ctx, "channel", "")

type NsqSubscriberConfig

type NsqSubscriberConfig struct {
	// nsq config
	// nsq increse MaxAttempts to avoid TestResendOnError, see github.com/ThreeDotsLabs/watermill/pubsub/tests/test_pubsub.go L54
	*nsq.Config
	LookupdAddrs []string
	NsqdAddrs    []string
	GroupName    string

	// CloseTimeout determines how long subscriber will wait for Ack/Nack on close.
	// When no Ack/Nack is received after CloseTimeout, subscriber will be closed.
	CloseTimeout time.Duration

	// How long subscriber should wait for Ack/Nack. When no Ack/Nack was received, message will be redelivered.
	// It is mapped to stan.AckWait option.
	AckWaitTimeout time.Duration

	// Requeue delay, default 1s
	//
	RequeueTimeout time.Duration

	// Unmarshaler is an unmarshaler used to unmarshaling messages from NATS format to Watermill format.
	Unmarshaler Unmarshaler
}

func (*NsqSubscriberConfig) Validate

func (c *NsqSubscriberConfig) Validate() error

type Unmarshaler

type Unmarshaler interface {
	Unmarshal(*nsq.Message) (*message.Message, error)
}