nsqd

Sep 24, 2015

Changes in this version

+ var ErrIDBackwards = errors.New("ID went backward")

type NSQD

+ func New(opts *Options) *NSQD

+ func (n *NSQD) RealHTTPSAddr() *net.TCPAddr

+ type Options struct

+ AuthHTTPAddresses []string

+ BroadcastAddress string

+ ClientTimeout time.Duration

+ DataPath string

+ DeflateEnabled bool

+ E2EProcessingLatencyPercentiles []float64

+ E2EProcessingLatencyWindowTime time.Duration

+ HTTPAddress string

+ HTTPSAddress string

+ ID int64

+ Logger logger

+ MaxBodySize int64

+ MaxBytesPerFile int64

+ MaxDeflateLevel int

+ MaxHeartbeatInterval time.Duration

+ MaxMsgSize int64

+ MaxMsgTimeout time.Duration

+ MaxOutputBufferSize int64

+ MaxOutputBufferTimeout time.Duration

+ MaxRdyCount int64

+ MaxReqTimeout time.Duration

+ MemQueueSize int64

+ MsgTimeout time.Duration

+ NSQLookupdTCPAddresses []string

+ QueueScanDirtyPercent float64

+ QueueScanInterval time.Duration

+ QueueScanRefreshInterval time.Duration

+ QueueScanSelectionCount int

+ QueueScanWorkerPoolMax int

+ SnappyEnabled bool

+ StatsdAddress string

+ StatsdInterval time.Duration

+ StatsdMemStats bool

+ StatsdPrefix string

+ SyncEvery int64

+ SyncTimeout time.Duration

+ TCPAddress string

+ TLSCert string

+ TLSClientAuthPolicy string

+ TLSKey string

+ TLSMinVersion uint16

+ TLSRequired int

+ TLSRootCAFile string

+ Verbose bool

+ func NewOptions() *Options

Apr 26, 2015

Changes in this version

type NSQD

+ func (n *NSQD) GetStartTime() time.Time

+ func (n *NSQD) RealHTTPAddr() *net.TCPAddr

+ func (n *NSQD) RealTCPAddr() *net.TCPAddr

Jan 22, 2015

Changes in this version

+ const MOVEFILE_REPLACE_EXISTING — windows/amd64

+ const TLSNotRequired

+ const TLSRequired

+ const TLSRequiredExceptHTTP

Aug 26, 2014

Changes in this version

type NSQD

+ func (n *NSQD) GetError() error

+ func (n *NSQD) GetHealth() string

+ func (n *NSQD) IsHealthy() bool

+ func (n *NSQD) SetHealth(err error)

Jul 28, 2014

Changes in this version

type ClientStats

+ CipherSuite string

+ TLSNegotiatedProtocol string

+ TLSNegotiatedProtocolIsMutual bool

+ TLSVersion string

Jul 25, 2014

Changes in this version

+ const ERROR_SUCCESS — windows/amd64

+ const ERROR_UNABLE_TO_MOVE_REPLACEMENT — windows/amd64

+ const ERROR_UNABLE_TO_MOVE_REPLACEMENT_2 — windows/amd64

+ const ERROR_UNABLE_TO_REMOVE_REPLACED — windows/amd64

+ const MsgIDLength

+ const REPLACEFILE_IGNORE_MERGE_ERRORS — windows/amd64

+ func GetLastError() uint32 — windows/amd64

type ClientStats

+ AuthIdentity string

+ AuthIdentityURL string

+ Authed bool

+ type Message struct

+ Attempts uint16

+ Body []byte

+ ID MessageID

+ Timestamp int64

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

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

+ type MessageID [MsgIDLength]byte

type NSQD

+ func (n *NSQD) IsAuthEnabled() bool

Apr 29, 2014

Changes in this version

type ClientStats

+ ClientID string

+ Hostname string

type NSQD

+ func (n *NSQD) GetStats() []TopicStats

Feb 16, 2014

Changes in this version

+ type GUIDFactory struct

+ func (f *GUIDFactory) NewGUID(workerId int64) (GUID, error)

Feb 6, 2014

Changes in this version

+ func NewNSQDOptions() *nsqdOptions

type ClientStats

+ Deflate bool

+ SampleRate int32

+ Snappy bool

+ TLS bool

+ UserAgent string

type ClientV2

+ Deflate int32

+ SampleRate int32

+ SampleRateUpdateChan chan int32

+ Snappy int32

+ TLS int32

+ UserAgent string

+ func (c *ClientV2) SetSampleRate(sampleRate int32) error

type IdentifyDataV2

+ SampleRate int32

+ UserAgent string

+ type NSQD struct

+ func NewNSQD(options *nsqdOptions) *NSQD

+ func (n *NSQD) DeleteExistingTopic(topicName string) error

+ func (n *NSQD) Exit()

+ func (n *NSQD) GetExistingTopic(topicName string) (*Topic, error)

+ func (n *NSQD) GetTopic(topicName string) *Topic

+ func (n *NSQD) LoadMetadata()

+ func (n *NSQD) Main()

+ func (n *NSQD) Notify(v interface{})

+ func (n *NSQD) PersistMetadata() error

type Topic

+ func (t *Topic) IsPaused() bool

+ func (t *Topic) Pause() error

+ func (t *Topic) UnPause() error

type TopicStats

+ Paused bool

Dec 7, 2013

Changes in this version

type ChannelStats

+ E2eProcessingLatency *util.PercentileResult

type Topic

+ func (t *Topic) AggregateChannelE2eProcessingLatency() *util.Quantile

type TopicStats

+ E2eProcessingLatency *util.PercentileResult

Oct 23, 2013

Changes in this version

+ const DefaultBufferSize

type ClientV2

+ func (c *ClientV2) UpgradeDeflate(level int) error

+ func (c *ClientV2) UpgradeSnappy() error

type IdentifyDataV2

+ Deflate bool

+ DeflateLevel int

+ Snappy bool

+ type LookupPeer struct

+ Info PeerInfo

+ func NewLookupPeer(addr string, connectCallback func(*LookupPeer)) *LookupPeer

+ func (lp *LookupPeer) Close() error

+ func (lp *LookupPeer) Command(cmd *nsq.Command) ([]byte, error)

+ func (lp *LookupPeer) Connect() error

+ func (lp *LookupPeer) Read(data []byte) (int, error)

+ func (lp *LookupPeer) String() string

+ func (lp *LookupPeer) Write(data []byte) (int, error)

+ type PeerInfo struct

+ Address string

+ BroadcastAddress string

+ HttpPort int

+ TcpPort int

+ Version string

Aug 26, 2013

Changes in this version

type ClientV2

+ ID int64

+ OutputBufferSize int

+ func (c *ClientV2) Flush() error

+ func (c *ClientV2) UpgradeTLS() error

+ type Context struct

type IdentifyDataV2

+ TLSv1 bool

+ type Uint64Slice []uint64

+ func (s Uint64Slice) Len() int

+ func (s Uint64Slice) Less(i, j int) bool

+ func (s Uint64Slice) Swap(i, j int)

Jun 9, 2013

Changes in this version

type BackendQueue

+ Delete func() error

type ClientV2

+ OutputBufferTimeout *time.Ticker

+ OutputBufferTimeoutUpdateChan chan time.Duration

+ func (c *ClientV2) Identify(data IdentifyDataV2) error

+ func (c *ClientV2) SetOutputBufferSize(desiredSize int) error

+ func (c *ClientV2) SetOutputBufferTimeout(desiredTimeout int) error

type DiskQueue

+ func (d *DiskQueue) Delete() error

type DummyBackendQueue

+ func (d *DummyBackendQueue) Delete() error

+ type IdentifyDataV2 struct

+ FeatureNegotiation bool

+ HeartbeatInterval int

+ LongId string

+ OutputBufferSize int

+ OutputBufferTimeout int

+ ShortId string

May 13, 2013

Changes in this version

type ClientV2

+ func (c *ClientV2) SetHeartbeatInterval(desiredInterval int) error

Apr 12, 2013

Changes in this version

type ClientV2

+ Heartbeat *time.Ticker

+ HeartbeatInterval time.Duration

+ HeartbeatUpdateChan chan time.Duration

+ func (c *ClientV2) Empty()

type Consumer

+ Empty func()

Feb 28, 2013

Changes in this version

type ClientV2

+ SubEventChan chan *Channel

Feb 8, 2013

Changes in this version

type Channel

+ func (c *Channel) Empty() error

+ func (c *Channel) TouchMessage(client Consumer, id nsq.MessageID) error

type NSQd

+ func (n *NSQd) Notify(v interface{})

+ type Notifier interface

+ Notify func(v interface{})

type ProtocolV2

+ func (p *ProtocolV2) TOUCH(client *ClientV2, params [][]byte) ([]byte, error)

type Topic

+ func (t *Topic) Empty() error

+ func (t *Topic) PutMessages(messages []*nsq.Message) error

Jan 7, 2013

Changes in this version

+ type ChannelStats struct

+ BackendDepth int64

+ ChannelName string

+ Clients []ClientStats

+ DeferredCount int

+ Depth int64

+ InFlightCount int

+ MessageCount uint64

+ Paused bool

+ RequeueCount uint64

+ TimeoutCount uint64

+ func NewChannelStats(c *Channel, clients []ClientStats) ChannelStats

type ClientStats

+ ConnectTime int64

+ FinishCount uint64

+ InFlightCount int64

+ MessageCount uint64

+ Name string

+ ReadyCount int64

+ RemoteAddress string

+ RequeueCount uint64

+ State int32

+ Version string

type ClientV2

+ Writer *bufio.Writer

type NSQd

+ func (n *NSQd) PersistMetadata()

type ProtocolV2

+ func (p *ProtocolV2) Flush(client *ClientV2) error

+ func (p *ProtocolV2) IDENTIFY(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) MPUB(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) SendMessage(client *ClientV2, msg *nsq.Message, buf *bytes.Buffer) error

+ type TopicStats struct

+ BackendDepth int64

+ Channels []ChannelStats

+ Depth int64

+ MessageCount uint64

+ TopicName string

+ func NewTopicStats(t *Topic, channels []ChannelStats) TopicStats

Oct 25, 2012

Changes in this version

+ var ErrSequenceExpired = errors.New("sequence expired")

+ var ErrTimeBackwards = errors.New("time has gone backwards")

+ func EmptyQueue(q Queue) error

+ func FlushQueue(q Queue) error

+ func NewNsqdOptions() *nsqdOptions

+ func WriteMessageToBackend(buf *bytes.Buffer, msg *nsq.Message, q Queue) error

+ type BackendQueue interface

+ Close func() error

+ Depth func() int64

+ Empty func() error

+ Put func([]byte) error

+ ReadChan func() chan []byte

+ func NewDiskQueue(name string, dataPath string, maxBytesPerFile int64, syncEvery int64) BackendQueue

+ func NewDummyBackendQueue() BackendQueue

+ type Channel struct

+ func NewChannel(topicName string, channelName string, options *nsqdOptions, ...) *Channel

+ func (c *Channel) AddClient(client Consumer)

+ func (c *Channel) BackendQueue() BackendQueue

+ func (c *Channel) Close() error

+ func (c *Channel) Deferred() map[string]*pqueue.Item

+ func (c *Channel) Delete() error

+ func (c *Channel) Depth() int64

+ func (c *Channel) Exiting() bool

+ func (c *Channel) FinishMessage(client Consumer, id []byte) error

+ func (c *Channel) InFlight() map[string]*pqueue.Item

+ func (c *Channel) IsPaused() bool

+ func (c *Channel) MemoryChan() chan *nsq.Message

+ func (c *Channel) Pause()

+ func (c *Channel) PutMessage(msg *nsq.Message) error

+ func (c *Channel) RemoveClient(client Consumer)

+ func (c *Channel) RequeueMessage(client Consumer, id []byte, timeout time.Duration) error

+ func (c *Channel) StartDeferredTimeout(msg *nsq.Message, timeout time.Duration) error

+ func (c *Channel) StartInFlightTimeout(msg *nsq.Message, client Consumer) error

+ func (c *Channel) UnPause()

+ type Channels []*Channel

+ func (c Channels) Len() int

+ func (c Channels) Swap(i, j int)

+ type ChannelsByName struct

+ func (c ChannelsByName) Less(i, j int) bool

+ type ClientStats struct

+ type ClientV2 struct

+ Channel *Channel

+ ConnectTime time.Time

+ ExitChan chan int

+ FinishCount uint64

+ InFlightCount int64

+ LastReadyCount int64

+ LongIdentifier string

+ MessageCount uint64

+ Reader *bufio.Reader

+ ReadyCount int64

+ ReadyStateChan chan int

+ RequeueCount uint64

+ ShortIdentifier string

+ State int32

+ func NewClientV2(conn net.Conn) *ClientV2

+ func (c *ClientV2) FinishedMessage()

+ func (c *ClientV2) IsReadyForMessages() bool

+ func (c *ClientV2) Pause()

+ func (c *ClientV2) RequeuedMessage()

+ func (c *ClientV2) SendingMessage()

+ func (c *ClientV2) SetReadyCount(count int64)

+ func (c *ClientV2) StartClose()

+ func (c *ClientV2) Stats() ClientStats

+ func (c *ClientV2) String() string

+ func (c *ClientV2) TimedOutMessage()

+ func (c *ClientV2) UnPause()

+ type Consumer interface

+ Close func() error

+ Pause func()

+ Stats func() ClientStats

+ TimedOutMessage func()

+ UnPause func()

+ type DiskQueue struct

+ func (d *DiskQueue) Close() error

+ func (d *DiskQueue) Depth() int64

+ func (d *DiskQueue) Empty() error

+ func (d *DiskQueue) Put(data []byte) error

+ func (d *DiskQueue) ReadChan() chan []byte

+ type DummyBackendQueue struct

+ func (d *DummyBackendQueue) Close() error

+ func (d *DummyBackendQueue) Depth() int64

+ func (d *DummyBackendQueue) Empty() error

+ func (d *DummyBackendQueue) Put([]byte) error

+ func (d *DummyBackendQueue) ReadChan() chan []byte

+ type GUID int64

+ func NewGUID(workerId int64) (GUID, error)

+ func (g GUID) Hex() []byte

+ type NSQd struct

+ func NewNSQd(workerId int64, options *nsqdOptions) *NSQd

+ func (n *NSQd) DeleteExistingTopic(topicName string) error

+ func (n *NSQd) Exit()

+ func (n *NSQd) GetExistingTopic(topicName string) (*Topic, error)

+ func (n *NSQd) GetTopic(topicName string) *Topic

+ func (n *NSQd) LoadMetadata()

+ func (n *NSQd) Main()

+ type ProtocolV2 struct

+ func (p *ProtocolV2) CLS(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) Exec(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) FIN(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) IOLoop(conn net.Conn) error

+ func (p *ProtocolV2) NOP(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) PUB(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) RDY(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) REQ(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) SUB(client *ClientV2, params [][]byte) ([]byte, error)

+ func (p *ProtocolV2) Send(client *ClientV2, frameType int32, data []byte) error

+ type Queue interface

+ BackendQueue func() BackendQueue

+ Deferred func() map[string]*pqueue.Item

+ InFlight func() map[string]*pqueue.Item

+ MemoryChan func() chan *nsq.Message

+ type TcpProtocol struct

+ func (p *TcpProtocol) Handle(clientConn net.Conn)

+ type Topic struct

+ func NewTopic(topicName string, options *nsqdOptions) *Topic

+ func (t *Topic) BackendQueue() BackendQueue

+ func (t *Topic) Close() error

+ func (t *Topic) Deferred() map[string]*pqueue.Item

+ func (t *Topic) Delete() error

+ func (t *Topic) DeleteExistingChannel(channelName string) error

+ func (t *Topic) Depth() int64

+ func (t *Topic) Exiting() bool

+ func (t *Topic) GetChannel(channelName string) *Channel

+ func (t *Topic) GetExistingChannel(channelName string) (*Channel, error)

+ func (t *Topic) InFlight() map[string]*pqueue.Item

+ func (t *Topic) MemoryChan() chan *nsq.Message

+ func (t *Topic) PutMessage(msg *nsq.Message) error

+ type Topics []*Topic

+ func (t Topics) Len() int

+ func (t Topics) Swap(i, j int)

+ type TopicsByName struct

+ func (t TopicsByName) Less(i, j int) bool