nsq

Nov 14, 2018

Changes in this version

+ const MAX_PARTITION_NUM

+ const MIN_RETRY_SLEEP

+ const OLD_VERSION_PID

+ var E_TOPIC_NOT_EXIST = "E_TOPIC_NOT_EXIST"

+ var E_TOPIC_NOT_EXIST_BYTES = []byte("E_TOPIC_NOT_EXIST")

+ var ErrNoProducer = errors.New("topic producer not found")

+ var ErrRetryBackground = errors.New("retrying in background")

+ var ErrTopicNotSet = errors.New("topic is not set as producer")

+ var FailedOnNotLeader = "E_FAILED_ON_NOT_LEADER"

+ var FailedOnNotLeaderBytes = []byte("E_FAILED_ON_NOT_LEADER")

+ var FailedOnNotWritable = "E_FAILED_ON_NOT_WRITABLE"

+ var FailedOnNotWritableBytes = []byte("E_FAILED_ON_NOT_WRITABLE")

+ var JSONHeaderExtVer = uint8(4)

+ var NoExtVer = uint8(0)

+ var OffsetSpecialType = "special"

+ var OffsetTimestampType = "timestamp"

+ var OffsetVirtualQueueType = "virtual_queue"

+ func FindString(src []string, f string) int

+ func IsFailedOnNotLeader(err error) bool

+ func IsFailedOnNotLeaderBytes(err []byte) bool

+ func IsFailedOnNotWritable(err error) bool

+ func IsFailedOnNotWritableBytes(err []byte) bool

+ func IsTopicNotExist(err error) bool

+ func IsTopicNotExistBytes(err []byte) bool

+ type AddrPartInfo struct

+ type CmdFuncT func(pid int) (*Command, error)

type Command

+ func CreateTopic(topic string, partition int) *Command

+ func CreateTopicWithExt(topic string, partition int) *Command

+ func MultiPublishTrace(topic string, part string, traceIDList []uint64, bodies [][]byte) (*Command, error)

+ func MultiPublishV2(topic string, bodies []*bytes.Buffer) (*Command, error)

+ func MultiPublishWithJsonExt(topic string, part string, extList []*MsgExt, bodies [][]byte) (*Command, error)

+ func MultiPublishWithPart(topic string, part string, bodies [][]byte) (*Command, error)

+ func MultiPublishWithPartV2(topic string, part string, bodies []*bytes.Buffer) (*Command, error)

+ func PublishTrace(topic string, part string, traceID uint64, body []byte) (*Command, error)

+ func PublishWithJsonExt(topic string, part string, body []byte, jsonExt []byte) (*Command, error)

+ func PublishWithPart(topic string, part string, body []byte) *Command

+ func SubscribeAdvanced(topic string, channel string, part string, consumeStart ConsumeOffset) *Command

+ func SubscribeAndTrace(topic string, channel string) *Command

+ func SubscribeOrdered(topic string, channel string, part string) *Command

+ func SubscribeWithPart(topic string, channel string, part string) *Command

+ func SubscribeWithPartAndTrace(topic string, channel string, part string) *Command

type Config

+ DesiredTag string

+ EnableMultiplexing bool

+ EnableOrdered bool

+ EnableTrace bool

+ Hasher hash.Hash32

+ LookupdSeeds []string

+ ProducerPoolSize int

+ PubBackgroundBuffer int

+ PubMaxBackgroundRetry int

+ PubMaxRetry int

+ PubStrategy PubStrategyType

+ PubTimeout time.Duration

type Conn

+ func (c *Conn) GetConnUID() string

+ type ConsumeOffset struct

+ OffsetType string

+ OffsetValue int64

+ func (self *ConsumeOffset) SetTime(sec int64)

+ func (self *ConsumeOffset) SetToEnd()

+ func (self *ConsumeOffset) SetVirtualQueueOffset(offset int64)

+ func (self *ConsumeOffset) ToString() string

type Consumer

+ func NewPartitionConsumer(topic string, part int, channel string, config *Config) (*Consumer, error)

+ func (r *Consumer) AddConcurrentHandlerFuncs(handler HandlerFunc, failHandler FailHandlerFunc, concurrency int)

+ func (r *Consumer) AddEtcdServiceAddr(address []string, cluster string, key string)

+ func (r *Consumer) ConnectToSeeds() error

+ func (r *Consumer) IsConsumeExt() bool

+ func (r *Consumer) SetConsumeExt(topicExt bool) bool

+ func (r *Consumer) SetConsumeOffset(partition int, offset ConsumeOffset) error

+ type FailHandlerFunc func(message *Message)

+ type FullMessageID [MsgIDLength]byte

type Message

+ ExtBytes []byte

+ ExtVer uint8

+ Offset uint64

+ Partition string

+ RawSize uint32

+ func DecodeMessageWithExt(b []byte, ext bool) (*Message, error)

+ func (m *Message) GetFullMsgID() FullMessageID

+ func (m *Message) GetJsonExt() (*MsgExt, error)

+ func (m *Message) GetTraceID() uint64

type MessageID

+ func GetCompatibleMsgIDFromNew(id NewMessageID, traceID uint64) MessageID

+ type MsgExt struct

+ Custom map[string]string

+ DispatchTag string

+ TraceID uint64

+ func (ext *MsgExt) ToJson() []byte

+ type NewMessageID uint64

+ func GetNewMessageID(old []byte) NewMessageID

+ type NsqLookupdNodeInfo struct

+ Epoch int64

+ HttpPort string

+ ID string

+ NodeIp string

+ RpcPort string

+ TcpPort string

type Producer

+ func (w *Producer) PublishWithPartitionIdAsync(topic string, partition string, body []byte, ext *MsgExt, ...) error

type ProducerTransaction

+ ResponseData []byte

+ type PubStrategyType int

+ const PubDynamicLoad

+ const PubRR

+ type RemoveProducerInfo struct

+ type TopicPartProducerInfo struct

+ func NewTopicPartProducerInfo(meta metaInfo, isMetaValid bool) *TopicPartProducerInfo

+ type TopicProducerMgr struct

+ func NewTopicProducerMgr(topics []string, conf *Config) (*TopicProducerMgr, error)

+ func (self *TopicProducerMgr) AddLookupdNodes(addresses []string)

+ func (self *TopicProducerMgr) ConnectToNSQLookupd(addr string) error

+ func (self *TopicProducerMgr) ConnectToSeeds() error

+ func (self *TopicProducerMgr) MultiPublish(topic string, body [][]byte) error

+ func (self *TopicProducerMgr) MultiPublishAndTrace(topic string, traceIDList []uint64, body [][]byte) (NewMessageID, uint64, uint32, error)

+ func (self *TopicProducerMgr) MultiPublishAsync(topic string, body [][]byte, doneChan chan *ProducerTransaction, ...) error

+ func (self *TopicProducerMgr) MultiPublishV2(topic string, body []*bytes.Buffer) error

+ func (self *TopicProducerMgr) MultiPublishWithJsonExt(topic string, body [][]byte, extList []*MsgExt) error

+ func (self *TopicProducerMgr) Publish(topic string, body []byte) error

+ func (self *TopicProducerMgr) PublishAndRetryBackground(topic string, body []byte) error

+ func (self *TopicProducerMgr) PublishAndTrace(topic string, traceID uint64, body []byte) (NewMessageID, uint64, uint32, error)

+ func (self *TopicProducerMgr) PublishAndTraceWithPartitionId(topic string, partition int, traceID uint64, body []byte) (NewMessageID, uint64, uint32, error)

+ func (self *TopicProducerMgr) PublishAsync(topic string, body []byte, doneChan chan *ProducerTransaction, ...) error

+ func (self *TopicProducerMgr) PublishAsyncWithJsonExt(topic string, body []byte, ext *MsgExt, doneChan chan *ProducerTransaction, ...) error

+ func (self *TopicProducerMgr) PublishAsyncWithPart(topic string, part int, body []byte, ext *MsgExt, ...) error

+ func (self *TopicProducerMgr) PublishOrdered(topic string, partitionKey []byte, body []byte) (NewMessageID, uint64, uint32, error)

+ func (self *TopicProducerMgr) PublishOrderedWithJsonExt(topic string, partitionKey []byte, body []byte, ext *MsgExt) (NewMessageID, uint64, uint32, error)

+ func (self *TopicProducerMgr) PublishWithExtAndRetryBackground(topic string, body []byte, ext *MsgExt) error

+ func (self *TopicProducerMgr) PublishWithJsonExt(topic string, body []byte, ext *MsgExt) (NewMessageID, uint64, uint32, error)

+ func (self *TopicProducerMgr) PublishWithJsonExtAndPartitionId(topic string, partition int, body []byte, ext *MsgExt) (NewMessageID, uint64, uint32, error)

+ func (self *TopicProducerMgr) PublishWithPartitionId(topic string, partition int, body []byte) error

+ func (self *TopicProducerMgr) SetEtcdConf(servers []string, cluster string, lookupPath string)

+ func (self *TopicProducerMgr) SetLogger(l logger, lvl LogLevel)

+ func (self *TopicProducerMgr) SetLoggerLevel(lvl LogLevel)

+ func (self *TopicProducerMgr) Stop()

+ func (self *TopicProducerMgr) TriggerCheckForError(err error, delay time.Duration)

Sep 20, 2015

Changes in this version

type Command

+ func DeferredPublish(topic string, delay time.Duration, body []byte) *Command

+ type ConfigFlag struct

+ func (c *ConfigFlag) Set(opt string) (err error)

+ func (c *ConfigFlag) String() string

Apr 8, 2015

Changes in this version

+ var ErrClosing = errors.New("closing")

+ type BackoffStrategy interface

+ Calculate func(attempt int) time.Duration

type Config

+ BackoffStrategy BackoffStrategy

+ DialTimeout time.Duration

+ LocalAddr net.Addr

+ RDYRedistributeInterval time.Duration

type ConnDelegate

+ OnContinue func(*Conn)

+ type ExponentialStrategy struct

+ func (s *ExponentialStrategy) Calculate(attempt int) time.Duration

+ type FullJitterStrategy struct

+ func (s *FullJitterStrategy) Calculate(attempt int) time.Duration

type LogLevel

+ func (lvl LogLevel) String() string

type Producer

+ func (w *Producer) Ping() error

Jan 22, 2015

Changes in this version

type Consumer

+ func (r *Consumer) Stats() *ConsumerStats

+ type ConsumerStats struct

+ Connections int

+ MessagesFinished uint64

+ MessagesReceived uint64

+ MessagesRequeued uint64

type Message

+ NSQDAddress string

Nov 10, 2014

Changes in this version

+ const FrameTypeError

+ const FrameTypeMessage

+ const FrameTypeResponse

+ const LogLevelDebug

+ const LogLevelDebugPrefix

+ const LogLevelError

+ const LogLevelErrorPrefix

+ const LogLevelInfo

+ const LogLevelInfoPrefix

+ const LogLevelWarning

+ const LogLevelWarningPrefix

+ const MsgIDLength

+ const StateClosing

+ const StateConnected

+ const StateDisconnected

+ const StateInit

+ const StateSubscribed

+ const VERSION

+ var ErrAlreadyConnected = errors.New("already connected")

+ var ErrNotConnected = errors.New("not connected")

+ var ErrOverMaxInFlight = errors.New("over configure max-inflight")

+ var ErrStopped = errors.New("stopped")

+ var MagicV1 = []byte(" V1")

+ var MagicV2 = []byte(" V2")

+ func IsValidChannelName(name string) bool

+ func IsValidTopicName(name string) bool

+ func ReadResponse(r io.Reader) ([]byte, error)

+ func ReadUnpackedResponse(r io.Reader) (int32, []byte, error)

+ func UnpackResponse(response []byte) (int32, []byte, error)

+ type AuthResponse struct

+ Identity string

+ IdentityUrl string

+ PermissionCount int64

+ type Command struct

+ Body []byte

+ Name []byte

+ Params [][]byte

+ func Auth(secret string) (*Command, error)

+ func Finish(id MessageID) *Command

+ func Identify(js map[string]interface{}) (*Command, error)

+ func MultiPublish(topic string, bodies [][]byte) (*Command, error)

+ func Nop() *Command

+ func Ping() *Command

+ func Publish(topic string, body []byte) *Command

+ func Ready(count int) *Command

+ func Register(topic string, channel string) *Command

+ func Requeue(id MessageID, delay time.Duration) *Command

+ func StartClose() *Command

+ func Subscribe(topic string, channel string) *Command

+ func Touch(id MessageID) *Command

+ func UnRegister(topic string, channel string) *Command

+ func (c *Command) String() string

+ func (c *Command) WriteTo(w io.Writer) (int64, error)

+ type Config struct

+ AuthSecret string

+ BackoffMultiplier time.Duration

+ ClientID string

+ DefaultRequeueDelay time.Duration

+ Deflate bool

+ DeflateLevel int

+ HeartbeatInterval time.Duration

+ Hostname string

+ LookupdPollInterval time.Duration

+ LookupdPollJitter float64

+ LowRdyIdleTimeout time.Duration

+ MaxAttempts uint16

+ MaxBackoffDuration time.Duration

+ MaxInFlight int

+ MaxRequeueDelay time.Duration

+ MsgTimeout time.Duration

+ OutputBufferSize int64

+ OutputBufferTimeout time.Duration

+ ReadTimeout time.Duration

+ SampleRate int32

+ Snappy bool

+ TlsConfig *tls.Config

+ TlsV1 bool

+ UserAgent string

+ WriteTimeout time.Duration

+ func NewConfig() *Config

+ func (c *Config) Set(option string, value interface{}) error

+ func (c *Config) Validate() error

+ type Conn struct

+ func NewConn(addr string, config *Config, delegate ConnDelegate) *Conn

+ func (c *Conn) Close() error

+ func (c *Conn) Connect() (*IdentifyResponse, error)

+ func (c *Conn) Flush() error

+ func (c *Conn) IsClosing() bool

+ func (c *Conn) LastMessageTime() time.Time

+ func (c *Conn) LastRDY() int64

+ func (c *Conn) MaxRDY() int64

+ func (c *Conn) RDY() int64

+ func (c *Conn) Read(p []byte) (int, error)

+ func (c *Conn) RemoteAddr() net.Addr

+ func (c *Conn) SetLogger(l logger, lvl LogLevel, format string)

+ func (c *Conn) SetRDY(rdy int64)

+ func (c *Conn) String() string

+ func (c *Conn) Write(p []byte) (int, error)

+ func (c *Conn) WriteCommand(cmd *Command) error

+ type ConnDelegate interface

+ OnBackoff func(*Conn)

+ OnClose func(*Conn)

+ OnError func(*Conn, []byte)

+ OnHeartbeat func(*Conn)

+ OnIOError func(*Conn, error)

+ OnMessage func(*Conn, *Message)

+ OnMessageFinished func(*Conn, *Message)

+ OnMessageRequeued func(*Conn, *Message)

+ OnResponse func(*Conn, []byte)

+ OnResume func(*Conn)

+ type Consumer struct

+ StopChan chan int

+ func NewConsumer(topic string, channel string, config *Config) (*Consumer, error)

+ func (r *Consumer) AddConcurrentHandlers(handler Handler, concurrency int)

+ func (r *Consumer) AddHandler(handler Handler)

+ func (r *Consumer) ChangeMaxInFlight(maxInFlight int)

+ func (r *Consumer) ConnectToNSQD(addr string) error

+ func (r *Consumer) ConnectToNSQDs(addresses []string) error

+ func (r *Consumer) ConnectToNSQLookupd(addr string) error

+ func (r *Consumer) ConnectToNSQLookupds(addresses []string) error

+ func (r *Consumer) DisconnectFromNSQD(addr string) error

+ func (r *Consumer) DisconnectFromNSQLookupd(addr string) error

+ func (r *Consumer) IsStarved() bool

+ func (r *Consumer) SetBehaviorDelegate(cb interface{})

+ func (r *Consumer) SetLogger(l logger, lvl LogLevel)

+ func (r *Consumer) Stop()

+ type DiscoveryFilter interface

+ Filter func([]string) []string

+ type ErrIdentify struct

+ Reason string

+ func (e ErrIdentify) Error() string

+ type ErrProtocol struct

+ Reason string

+ func (e ErrProtocol) Error() string

+ type FailedMessageLogger interface

+ LogFailedMessage func(message *Message)

+ type Handler interface

+ HandleMessage func(message *Message) error

+ type HandlerFunc func(message *Message) error

+ func (h HandlerFunc) HandleMessage(m *Message) error

+ type IdentifyResponse struct

+ AuthRequired bool

+ Deflate bool

+ MaxRdyCount int64

+ Snappy bool

+ TLSv1 bool

+ type LogLevel int

+ type Message struct

+ Attempts uint16

+ Body []byte

+ Delegate MessageDelegate

+ ID MessageID

+ Timestamp int64

+ func DecodeMessage(b []byte) (*Message, error)

+ func NewMessage(id MessageID, body []byte) *Message

+ func (m *Message) DisableAutoResponse()

+ func (m *Message) Finish()

+ func (m *Message) HasResponded() bool

+ func (m *Message) IsAutoResponseDisabled() bool

+ func (m *Message) Requeue(delay time.Duration)

+ func (m *Message) RequeueWithoutBackoff(delay time.Duration)

+ func (m *Message) Touch()

+ func (m *Message) WriteTo(w io.Writer) (int64, error)

+ type MessageDelegate interface

+ OnFinish func(*Message)

+ OnRequeue func(m *Message, delay time.Duration, backoff bool)

+ OnTouch func(*Message)

+ type MessageID [MsgIDLength]byte

+ type Producer struct

+ func NewProducer(addr string, config *Config) (*Producer, error)

+ func (w *Producer) MultiPublish(topic string, body [][]byte) error

+ func (w *Producer) MultiPublishAsync(topic string, body [][]byte, doneChan chan *ProducerTransaction, ...) error

+ func (w *Producer) Publish(topic string, body []byte) error

+ func (w *Producer) PublishAsync(topic string, body []byte, doneChan chan *ProducerTransaction, ...) error

+ func (w *Producer) SetLogger(l logger, lvl LogLevel)

+ func (w *Producer) Stop()

+ func (w *Producer) String() string

+ type ProducerTransaction struct

+ Args []interface{}

+ Error error