fakensq

README

Fake NSQ

Fake NSQ Consumer and Producer for backend/internal/pkg/nsq

Built to test the correctness of the nsqio wrapper

Limitations:

• Consumer must registered first before publishing message.
• Do not expecting message to be stored, all message directly consumed.
• Message published before any active consumer will be lost.
• Message requeue not working
• Message is always finished

Documentation

Index

• type ConsumerConfig
  • func (cc *ConsumerConfig) Validate() error
• type ErrorConsumerFake
  • func (ecm *ErrorConsumerFake) As(err error, target interface{}) bool
  • func (ecm *ErrorConsumerFake) Channel() string
  • func (ecm *ErrorConsumerFake) Error() string
  • func (ecm *ErrorConsumerFake) Is(err, target error) bool
  • func (ecm *ErrorConsumerFake) Message() []byte
  • func (ecm *ErrorConsumerFake) Topic() string
  • func (ecm *ErrorConsumerFake) Unwrap(err error) error
• type FakeConsumer
  • func NewFakeConsumer(config ConsumerConfig) (*FakeConsumer, error)
  • func (cm *FakeConsumer) AddConcurrentHandlers(handler nsqio.Handler, concurrency int)
  • func (cm *FakeConsumer) AddHandler(handler nsqio.Handler)
  • func (cm *FakeConsumer) BufferMultiplier() int
  • func (cm *FakeConsumer) ChangeMaxInFlight(n int)
  • func (cm *FakeConsumer) Channel() string
  • func (cm *FakeConsumer) Concurrency() int
  • func (cm *FakeConsumer) ConnectToNSQLookupds(addresses []string) error
  • func (cm *FakeConsumer) Stop()
  • func (cm *FakeConsumer) Topic() string
• type FakeLookupd
• type FakeProducer
  • func NewFakeProducer(consumer *FakeConsumer) *FakeProducer
  • func (fp *FakeProducer) MultiPublish(topic string, messages [][]byte) error
  • func (fp *FakeProducer) Ping() error
  • func (fp *FakeProducer) Publish(topic string, message []byte) error
  • func (fp *FakeProducer) Stop()
• type Message
• type MessageDelegator
  • func (mdm *MessageDelegator) OnFinish(message *nsqio.Message)
  • func (mdm *MessageDelegator) OnRequeue(m *nsqio.Message, t time.Duration, backoff bool)
  • func (mdm *MessageDelegator) OnTouch(m *nsqio.Message)

Types

type ConsumerConfig

type ConsumerConfig struct {
	Topic            string
	Channel          string
	Concurrency      int
	BufferMultiplier int
}

ConsumerConfig of fake nsq

func (*ConsumerConfig) Validate

func (cc *ConsumerConfig) Validate() error

Validate consumer configuration

type ErrorConsumerFake

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

ErrorConsumerFake for throwing error from the mock consumer

func (*ErrorConsumerFake) As

func (ecm *ErrorConsumerFake) As(err error, target interface{}) bool

As implementation of error

func (*ErrorConsumerFake) Channel

func (ecm *ErrorConsumerFake) Channel() string

Channel return the channel of error

func (*ErrorConsumerFake) Error

func (ecm *ErrorConsumerFake) Error() string

Error return the error string from error consumer mock

func (*ErrorConsumerFake) Is

func (ecm *ErrorConsumerFake) Is(err, target error) bool

Is implementation of error

func (*ErrorConsumerFake) Message

func (ecm *ErrorConsumerFake) Message() []byte

Message return the message that return error

func (*ErrorConsumerFake) Topic

func (ecm *ErrorConsumerFake) Topic() string

Topic return the topic of error

func (*ErrorConsumerFake) Unwrap

func (ecm *ErrorConsumerFake) Unwrap(err error) error

Unwrap implementation of error

type FakeConsumer

type FakeConsumer struct {
	*FakeLookupd

	ErrChan chan error

	// nsq configuration
	MaxInFlight int
	// contains filtered or unexported fields
}

FakeConsumer struct

func NewFakeConsumer

func NewFakeConsumer(config ConsumerConfig) (*FakeConsumer, error)

NewFakeConsumer function

func (*FakeConsumer) AddConcurrentHandlers

func (cm *FakeConsumer) AddConcurrentHandlers(handler nsqio.Handler, concurrency int)

AddConcurrentHandlers for nsq

func (*FakeConsumer) AddHandler

func (cm *FakeConsumer) AddHandler(handler nsqio.Handler)

AddHandler for nsq

func (*FakeConsumer) BufferMultiplier

func (cm *FakeConsumer) BufferMultiplier() int

BufferMultiplier return the number of buffer multiplier

func (*FakeConsumer) ChangeMaxInFlight

func (cm *FakeConsumer) ChangeMaxInFlight(n int)

ChangeMaxInFlight message in nsq consumer

func (*FakeConsumer) Channel

func (cm *FakeConsumer) Channel() string

Channel return the consumer channel

func (*FakeConsumer) Concurrency

func (cm *FakeConsumer) Concurrency() int

Concurrency return the number of conccurent worker

func (*FakeConsumer) ConnectToNSQLookupds

func (cm *FakeConsumer) ConnectToNSQLookupds(addresses []string) error

ConnectToNSQLookupds for nsq

func (*FakeConsumer) Stop

func (cm *FakeConsumer) Stop()

Stop consumer backend mock

func (*FakeConsumer) Topic

func (cm *FakeConsumer) Topic() string

Topic return the consumer topic

type FakeLookupd

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

FakeLookupd for storing all information regarding topics and channel

type FakeProducer

type FakeProducer struct {
	*FakeLookupd
}

FakeProducer struct

func NewFakeProducer

func NewFakeProducer(consumer *FakeConsumer) *FakeProducer

NewFakeProducer for publishing message to NSQ

func (*FakeProducer) MultiPublish

func (fp *FakeProducer) MultiPublish(topic string, messages [][]byte) error

MultiPublish message

func (*FakeProducer) Ping

func (fp *FakeProducer) Ping() error

Ping will always return nil

func (*FakeProducer) Publish

func (fp *FakeProducer) Publish(topic string, message []byte) error

Publish a message this function might block if the channel is full

func (*FakeProducer) Stop

func (fp *FakeProducer) Stop()

Stop fake producer

type Message

type Message struct {
	Name string
	Body []byte
}

Message mock

type MessageDelegator

type MessageDelegator struct {
}

MessageDelegator implement Delegator of nsqio

func (*MessageDelegator) OnFinish

func (mdm *MessageDelegator) OnFinish(message *nsqio.Message)

func (*MessageDelegator) OnRequeue

func (mdm *MessageDelegator) OnRequeue(m *nsqio.Message, t time.Duration, backoff bool)

func (*MessageDelegator) OnTouch

func (mdm *MessageDelegator) OnTouch(m *nsqio.Message)