infra

Documentation

Index

• type InfraSetup
  • func NewInfraSetup(jetstream JetStream, ...) *InfraSetup
  • func (i *InfraSetup) Setup() error
• type JetStream
• type JetStreamInfo
  • func NewJetStream(seedFile string, natsUri string) (*JetStreamInfo, error)
  • func (jetstream *JetStreamInfo) Close()
  • func (jetstream *JetStreamInfo) CreateConsumer(streamName string, consumerOpts ...jsm.ConsumerOption) error
  • func (jetstream *JetStreamInfo) CreateStream(streamName string, streamOpts ...jsm.StreamOption) error
  • func (jetstream *JetStreamInfo) IsStreamAvailable(streamName string) (bool, error)
• type MockNats
  • func (m *MockNats) Close()
  • func (m *MockNats) Fetch(messageCount int, ctx context.Context) ([]*NatsMessage, error)
  • func (m *MockNats) Publish(msg *NatsMessage) error
  • func (m *MockNats) Subscribe(subject, consumerName string) error
• type Nats
• type NatsImpl
  • func NewNats(seedFile string, natsUris string) (*NatsImpl, error)
  • func (n *NatsImpl) Close()
  • func (n *NatsImpl) Fetch(messageCount int, ctx context.Context) ([]*NatsMessage, error)
  • func (n *NatsImpl) Publish(msg *NatsMessage) error
  • func (n *NatsImpl) Subscribe(subject, consumerName string) error
• type NatsMessage
  • func NewNatsMessage(subject string) *NatsMessage
  • func (m *NatsMessage) Ack()
  • func (m *NatsMessage) AddHeader(key string, value string)
  • func (m *NatsMessage) GetBody() []byte
  • func (m *NatsMessage) GetHeader(key string) string
  • func (m *NatsMessage) GetHeaders() map[string][]string
  • func (m *NatsMessage) GetUnderlyingNatsMessage() *nats.Msg
  • func (m *NatsMessage) Nack()
  • func (m *NatsMessage) SetBody(body []byte)

Types

type InfraSetup

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

func NewInfraSetup

func NewInfraSetup(jetstream JetStream, streamName, streamSubjects, consumerName, consumerSubject string, maxAge time.Duration, replicas int) *InfraSetup

func (*InfraSetup) Setup

func (i *InfraSetup) Setup() error

type JetStream

type JetStream interface {
	CreateStream(streamName string, streamOpts ...jsm.StreamOption) error
	CreateConsumer(streamName string, consumerOpts ...jsm.ConsumerOption) error
	Close()
	IsStreamAvailable(streamName string) (bool, error)
}

type JetStreamInfo

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

func NewJetStream

func NewJetStream(seedFile string, natsUri string) (*JetStreamInfo, error)

func (*JetStreamInfo) Close

func (jetstream *JetStreamInfo) Close()

func (*JetStreamInfo) CreateConsumer

func (jetstream *JetStreamInfo) CreateConsumer(streamName string, consumerOpts ...jsm.ConsumerOption) error

func (*JetStreamInfo) CreateStream

func (jetstream *JetStreamInfo) CreateStream(streamName string, streamOpts ...jsm.StreamOption) error

func (*JetStreamInfo) IsStreamAvailable

func (jetstream *JetStreamInfo) IsStreamAvailable(streamName string) (bool, error)

type MockNats

type MockNats struct {
	mock.Mock
}

func (*MockNats) Close

func (m *MockNats) Close()

func (*MockNats) Fetch

func (m *MockNats) Fetch(messageCount int, ctx context.Context) ([]*NatsMessage, error)

func (*MockNats) Publish

func (m *MockNats) Publish(msg *NatsMessage) error

func (*MockNats) Subscribe

func (m *MockNats) Subscribe(subject, consumerName string) error

type Nats

type Nats interface {
	Publish(msg *NatsMessage) error
	Subscribe(subject, consumerName string) error
	Fetch(messageCount int, ctx context.Context) ([]*NatsMessage, error)
	Close()
}

type NatsImpl

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

func NewNats

func NewNats(seedFile string, natsUris string) (*NatsImpl, error)

func (*NatsImpl) Close

func (n *NatsImpl) Close()

func (*NatsImpl) Fetch

func (n *NatsImpl) Fetch(messageCount int, ctx context.Context) ([]*NatsMessage, error)

func (*NatsImpl) Publish

func (n *NatsImpl) Publish(msg *NatsMessage) error

func (*NatsImpl) Subscribe

func (n *NatsImpl) Subscribe(subject, consumerName string) error

type NatsMessage

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

func NewNatsMessage

func NewNatsMessage(subject string) *NatsMessage

func (*NatsMessage) Ack

func (m *NatsMessage) Ack()

func (*NatsMessage) AddHeader

func (m *NatsMessage) AddHeader(key string, value string)

func (*NatsMessage) GetBody

func (m *NatsMessage) GetBody() []byte

func (*NatsMessage) GetHeader

func (m *NatsMessage) GetHeader(key string) string

func (*NatsMessage) GetHeaders

func (m *NatsMessage) GetHeaders() map[string][]string

func (*NatsMessage) GetUnderlyingNatsMessage

func (m *NatsMessage) GetUnderlyingNatsMessage() *nats.Msg

func (*NatsMessage) Nack

func (m *NatsMessage) Nack()

func (*NatsMessage) SetBody

func (m *NatsMessage) SetBody(body []byte)