nsqd

package
v0.3.7-HA.1.3 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 10, 2016 License: MIT Imports: 36 Imported by: 0

README

nsqd

nsqd is the daemon that receives, queues, and delivers messages to clients.

Read the docs

Documentation

Index

Constants

View Source
const (
	MsgIDLength      = 16
	MsgTraceIDLength = 8
)
View Source
const (
	TLSNotRequired = iota
	TLSRequiredExceptHTTP
	TLSRequired
)
View Source
const (
	MAX_NODE_ID = 1024 * 1024
)
View Source
const (
	MAX_POSSIBLE_MSG_SIZE = 1 << 28
)
View Source
const (
	MAX_TOPIC_PARTITION = 1023
)

Variables

View Source
var (
	ErrMsgNotInFlight                 = errors.New("Message ID not in flight")
	ErrMsgAlreadyInFlight             = errors.New("Message ID already in flight")
	ErrConsumeDisabled                = errors.New("Consume is disabled currently")
	ErrMsgDeferred                    = errors.New("Message is deferred")
	ErrSetConsumeOffsetNotFirstClient = errors.New("consume offset can only be changed by the first consume client")
	ErrNotDiskQueueReader             = errors.New("the consume channel is not disk queue reader")
)
View Source
var (
	ErrConfirmSizeInvalid    = errors.New("Confirm data size invalid.")
	ErrConfirmCntInvalid     = errors.New("Confirm message count invalid.")
	ErrMoveOffsetInvalid     = errors.New("move offset invalid")
	ErrOffsetTypeMismatch    = errors.New("offset type mismatch")
	ErrReadQueueCountMissing = errors.New("read queue count info missing")
	ErrExiting               = errors.New("exiting")
)
View Source
var (
	ErrTopicPartitionMismatch = errors.New("topic partition mismatch")
	ErrTopicNotExist          = errors.New("topic does not exist")
)
View Source
var (
	ErrInvalidMessageID    = errors.New("message id is invalid")
	ErrWriteOffsetMismatch = errors.New("write offset mismatch")
)
View Source
var (
	ErrInvalidOffset = errors.New("invalid offset")
)

Functions

func GetTopicFullName

func GetTopicFullName(topic string, part int) string

func GetTraceIDFromFullMsgID

func GetTraceIDFromFullMsgID(id FullMessageID) uint64

func MessageHeaderBytes

func MessageHeaderBytes() int

func NewBufioReader

func NewBufioReader(r io.Reader) *bufio.Reader

func NsqLogger

func NsqLogger() *levellogger.LevelLogger

func PutBufioReader

func PutBufioReader(br *bufio.Reader)

Types

type BackendOffset

type BackendOffset int64

type BackendQueue

type BackendQueue interface {
	Put([]byte) error
	ReadChan() chan []byte // this is expected to be an *unbuffered* channel
	Close() error
	Delete() error
	Depth() int64
	Empty() error
}

BackendQueue represents the behavior for the secondary message storage system

type BackendQueueEnd

type BackendQueueEnd interface {
	Offset() BackendOffset
	TotalMsgCnt() int64
	IsSame(BackendQueueEnd) bool
}

type BackendQueueReader

type BackendQueueReader interface {
	ConfirmRead(BackendOffset, int64) error
	ResetReadToConfirmed() (BackendQueueEnd, error)
	SkipReadToOffset(BackendOffset, int64) (BackendQueueEnd, error)
	SkipReadToEnd() (BackendQueueEnd, error)
	Close() error
	// left data to be read
	Depth() int64
	DepthSize() int64
	GetQueueReadEnd() BackendQueueEnd
	GetQueueConfirmed() BackendQueueEnd
	Delete() error
	UpdateQueueEnd(BackendQueueEnd, bool) (bool, error)
	TryReadOne() (ReadResult, bool)
}

for channel consumer

type BackendQueueWriter

type BackendQueueWriter interface {
	Put([]byte) (BackendOffset, int32, int64, error)
	Close() error
	Delete() error
	Empty() error
	Flush() error
	GetQueueWriteEnd() BackendQueueEnd
	GetQueueReadEnd() BackendQueueEnd
	RollbackWrite(BackendOffset, uint64) error
	ResetWriteEnd(BackendOffset, int64) error
}

for topic producer

func NewDiskQueueWriter

func NewDiskQueueWriter(name string, dataPath string, maxBytesPerFile int64,
	minMsgSize int32, maxMsgSize int32,
	syncEvery int64) BackendQueueWriter

type Channel

type Channel struct {
	sync.RWMutex

	// stat counters
	EnableTrace int32
	// contains filtered or unexported fields
}

Channel represents the concrete type for a NSQ channel (and also implements the Queue interface)

There can be multiple channels per topic, each with there own unique set of subscribers (clients).

Channels maintain all client and message metadata, orchestrating in-flight messages, timeouts, requeuing, etc.

func NewChannel

func NewChannel(topicName string, part int, channelName string, chEnd BackendQueueEnd, opt *Options,
	deleteCallback func(*Channel), consumeDisabled int32, notify func(v interface{})) *Channel

NewChannel creates a new instance of the Channel type and returns a pointer

func (*Channel) AddClient

func (c *Channel) AddClient(clientID int64, client Consumer) error

AddClient adds a client to the Channel's client list

func (*Channel) Close

func (c *Channel) Close() error

Close cleanly closes the Channel

func (*Channel) ConfirmBackendQueue

func (c *Channel) ConfirmBackendQueue(msg *Message) (BackendOffset, int64, bool, error)

in order not to make the confirm map too large, we need handle this case: a old message is not confirmed, and we keep all the newer confirmed messages so we can confirm later. indicated weather the confirmed offset is changed

func (*Channel) ConfirmBackendQueueOnSlave

func (c *Channel) ConfirmBackendQueueOnSlave(offset BackendOffset, cnt int64, allowBackward bool) error

func (*Channel) ContinueConsumeForOrder

func (c *Channel) ContinueConsumeForOrder()

func (*Channel) Delete

func (c *Channel) Delete() error

Delete empties the channel and closes

func (*Channel) Depth

func (c *Channel) Depth() int64

func (*Channel) DepthSize

func (c *Channel) DepthSize() int64

func (*Channel) DepthTimestamp

func (c *Channel) DepthTimestamp() int64

func (*Channel) DisableConsume

func (c *Channel) DisableConsume(disable bool)

func (*Channel) Exiting

func (c *Channel) Exiting() bool

Exiting returns a boolean indicating if this channel is closed/exiting

func (*Channel) FinishMessage

func (c *Channel) FinishMessage(clientID int64, id MessageID) (BackendOffset, int64, bool, error)

FinishMessage successfully discards an in-flight message

func (*Channel) GetChannelDebugStats

func (c *Channel) GetChannelDebugStats() string

func (*Channel) GetChannelEnd

func (c *Channel) GetChannelEnd() BackendQueueEnd

func (*Channel) GetClientMsgChan

func (c *Channel) GetClientMsgChan() chan *Message

func (*Channel) GetClients

func (c *Channel) GetClients() map[int64]Consumer

func (*Channel) GetClientsCount

func (c *Channel) GetClientsCount() int

func (*Channel) GetConfirmed

func (c *Channel) GetConfirmed() BackendQueueEnd

func (*Channel) GetInflightNum

func (c *Channel) GetInflightNum() int

func (*Channel) GetName

func (c *Channel) GetName() string

func (*Channel) GetTopicName

func (c *Channel) GetTopicName() string

func (*Channel) GetTopicPart

func (c *Channel) GetTopicPart() int

func (*Channel) IsConfirmed

func (c *Channel) IsConfirmed(msg *Message) bool

func (*Channel) IsConsumeDisabled

func (c *Channel) IsConsumeDisabled() bool

func (*Channel) IsOrdered

func (c *Channel) IsOrdered() bool

func (*Channel) IsPaused

func (c *Channel) IsPaused() bool

func (*Channel) IsTraced

func (c *Channel) IsTraced() bool

func (*Channel) IsWaitingMoreData

func (c *Channel) IsWaitingMoreData() bool

func (*Channel) Pause

func (c *Channel) Pause() error

func (*Channel) RemoveClient

func (c *Channel) RemoveClient(clientID int64)

RemoveClient removes a client from the Channel's client list

func (*Channel) RequeueClientMessages

func (c *Channel) RequeueClientMessages(clientID int64)

func (*Channel) RequeueMessage

func (c *Channel) RequeueMessage(clientID int64, id MessageID, timeout time.Duration) error

RequeueMessage requeues a message based on `time.Duration`, ie:

`timeoutMs` == 0 - requeue a message immediately `timeoutMs` > 0 - asynchronously wait for the specified timeout

and requeue a message

func (*Channel) SetConsumeOffset

func (c *Channel) SetConsumeOffset(offset BackendOffset, cnt int64, force bool) error

func (*Channel) SetOrdered

func (c *Channel) SetOrdered(enable bool)

func (*Channel) SetTrace

func (c *Channel) SetTrace(enable bool)

func (*Channel) StartInFlightTimeout

func (c *Channel) StartInFlightTimeout(msg *Message, clientID int64, timeout time.Duration) error

func (*Channel) TouchMessage

func (c *Channel) TouchMessage(clientID int64, id MessageID, clientMsgTimeout time.Duration) error

TouchMessage resets the timeout for an in-flight message

func (*Channel) TryWakeupRead

func (c *Channel) TryWakeupRead()

func (*Channel) UnPause

func (c *Channel) UnPause() error

func (*Channel) UpdateQueueEnd

func (c *Channel) UpdateQueueEnd(end BackendQueueEnd, forceReload bool) error

When topic message is put, update the new end of the queue

type ChannelStats

type ChannelStats struct {
	ChannelName string `json:"channel_name"`
	// message size need to consume
	Depth          int64  `json:"depth"`
	DepthSize      int64  `json:"depth_size"`
	DepthTimestamp string `json:"depth_ts"`
	BackendDepth   int64  `json:"backend_depth"`
	// total size sub past hour on this channel
	HourlySubSize int64         `json:"hourly_subsize"`
	InFlightCount int           `json:"in_flight_count"`
	DeferredCount int           `json:"deferred_count"`
	MessageCount  uint64        `json:"message_count"`
	RequeueCount  uint64        `json:"requeue_count"`
	TimeoutCount  uint64        `json:"timeout_count"`
	Clients       []ClientStats `json:"clients"`
	Paused        bool          `json:"paused"`

	E2eProcessingLatency *quantile.Result `json:"e2e_processing_latency"`
}

func NewChannelStats

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

type Channels

type Channels []*Channel

func (Channels) Len

func (c Channels) Len() int

func (Channels) Swap

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

type ChannelsByName

type ChannelsByName struct {
	Channels
}

func (ChannelsByName) Less

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

type ClientPubStats

type ClientPubStats struct {
	RemoteAddress string `json:"remote_address"`
	UserAgent     string `json:"user_agent"`
	Protocol      string `json:"protocol"`
	PubCount      int64  `json:"pub_count"`
	ErrCount      int64  `json:"err_count"`
	LastPubTs     int64  `json:"last_pub_ts"`
}

type ClientStats

type ClientStats struct {
	// TODO: deprecated, remove in 1.0
	Name string `json:"name"`

	ClientID        string `json:"client_id"`
	Hostname        string `json:"hostname"`
	Version         string `json:"version"`
	RemoteAddress   string `json:"remote_address"`
	State           int32  `json:"state"`
	ReadyCount      int64  `json:"ready_count"`
	InFlightCount   int64  `json:"in_flight_count"`
	MessageCount    uint64 `json:"message_count"`
	FinishCount     uint64 `json:"finish_count"`
	RequeueCount    uint64 `json:"requeue_count"`
	TimeoutCount    int64  `json:"timeout_count"`
	DeferredCount   int64  `json:"deferred_count"`
	ConnectTime     int64  `json:"connect_ts"`
	SampleRate      int32  `json:"sample_rate"`
	Deflate         bool   `json:"deflate"`
	Snappy          bool   `json:"snappy"`
	UserAgent       string `json:"user_agent"`
	Authed          bool   `json:"authed,omitempty"`
	AuthIdentity    string `json:"auth_identity,omitempty"`
	AuthIdentityURL string `json:"auth_identity_url,omitempty"`

	TLS                           bool   `json:"tls"`
	CipherSuite                   string `json:"tls_cipher_suite"`
	TLSVersion                    string `json:"tls_version"`
	TLSNegotiatedProtocol         string `json:"tls_negotiated_protocol"`
	TLSNegotiatedProtocolIsMutual bool   `json:"tls_negotiated_protocol_is_mutual"`
}

type ClientV2

type ClientV2 struct {
	// 64bit atomic vars need to be first for proper alignment on 32bit platforms
	ReadyCount    int64
	InFlightCount int64
	MessageCount  uint64
	FinishCount   uint64
	RequeueCount  uint64
	TimeoutCount  int64
	DeferredCount int64

	ID int64

	UserAgent string

	// original connection
	net.Conn

	// reading/writing interfaces
	Reader *bufio.Reader
	Writer *bufio.Writer

	OutputBufferSize    int
	OutputBufferTimeout time.Duration

	HeartbeatInterval time.Duration

	MsgTimeout time.Duration

	State          int32
	ConnectTime    time.Time
	Channel        *Channel
	ReadyStateChan chan int
	// this is only used by notify messagebump to quit
	// and should be closed by the read loop only
	ExitChan chan int

	ClientID string
	Hostname string

	SampleRate int32

	IdentifyEventChan chan identifyEvent
	SubEventChan      chan *Channel

	TLS     int32
	Snappy  int32
	Deflate int32

	LenSlice []byte

	AuthSecret string
	AuthState  *auth.State

	EnableTrace bool
	// contains filtered or unexported fields
}

func NewClientV2

func NewClientV2(id int64, conn net.Conn, opts *Options, tls *tls.Config) *ClientV2

func (*ClientV2) Auth

func (c *ClientV2) Auth(secret string) error

func (*ClientV2) Empty

func (c *ClientV2) Empty()

func (*ClientV2) Exit

func (c *ClientV2) Exit()

func (*ClientV2) FinalClose

func (c *ClientV2) FinalClose()

func (*ClientV2) FinishedMessage

func (c *ClientV2) FinishedMessage()

func (*ClientV2) Flush

func (c *ClientV2) Flush() error

func (*ClientV2) HasAuthorizations

func (c *ClientV2) HasAuthorizations() bool

func (*ClientV2) Identify

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

func (*ClientV2) IsAuthorized

func (c *ClientV2) IsAuthorized(topic, channel string) (bool, error)

func (*ClientV2) IsReadyForMessages

func (c *ClientV2) IsReadyForMessages() bool

func (*ClientV2) LockRead

func (c *ClientV2) LockRead()

func (*ClientV2) LockWrite

func (c *ClientV2) LockWrite()

func (*ClientV2) Pause

func (c *ClientV2) Pause()

func (*ClientV2) QueryAuthd

func (c *ClientV2) QueryAuthd() error

func (*ClientV2) RequeuedMessage

func (c *ClientV2) RequeuedMessage(delayed bool)

func (*ClientV2) SendingMessage

func (c *ClientV2) SendingMessage()

func (*ClientV2) SetHeartbeatInterval

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

func (*ClientV2) SetMsgTimeout

func (c *ClientV2) SetMsgTimeout(msgTimeout int) error

func (*ClientV2) SetOutputBufferSize

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

func (*ClientV2) SetOutputBufferTimeout

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

func (*ClientV2) SetReadyCount

func (c *ClientV2) SetReadyCount(count int64)

func (*ClientV2) SetSampleRate

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

func (*ClientV2) StartClose

func (c *ClientV2) StartClose()

func (*ClientV2) Stats

func (c *ClientV2) Stats() ClientStats

func (*ClientV2) String

func (c *ClientV2) String() string

func (*ClientV2) TimedOutMessage

func (c *ClientV2) TimedOutMessage(isDefer bool)

func (*ClientV2) UnPause

func (c *ClientV2) UnPause()

func (*ClientV2) UnlockRead

func (c *ClientV2) UnlockRead()

func (*ClientV2) UnlockWrite

func (c *ClientV2) UnlockWrite()

func (*ClientV2) UpgradeDeflate

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

func (*ClientV2) UpgradeSnappy

func (c *ClientV2) UpgradeSnappy() error

func (*ClientV2) UpgradeTLS

func (c *ClientV2) UpgradeTLS() error

type Consumer

type Consumer interface {
	UnPause()
	Pause()
	TimedOutMessage(delayed bool)
	Stats() ClientStats
	Exit()
	Empty()
}

type DiskQueueSnapshot

type DiskQueueSnapshot struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewDiskQueueSnapshot

func NewDiskQueueSnapshot(readFrom string, dataPath string, endInfo BackendQueueEnd) *DiskQueueSnapshot

newDiskQueue instantiates a new instance of DiskQueueSnapshot, retrieving metadata from the filesystem and starting the read ahead goroutine

func (*DiskQueueSnapshot) Close

func (d *DiskQueueSnapshot) Close() error

Close cleans up the queue and persists metadata

func (*DiskQueueSnapshot) ReadOne

func (d *DiskQueueSnapshot) ReadOne() ReadResult

readOne performs a low level filesystem read for a single []byte while advancing read positions and rolling files, if necessary

func (*DiskQueueSnapshot) ReadRaw

func (d *DiskQueueSnapshot) ReadRaw(size int32) ([]byte, error)

func (*DiskQueueSnapshot) SeekTo

func (d *DiskQueueSnapshot) SeekTo(voffset BackendOffset) error

func (*DiskQueueSnapshot) SeekToEnd

func (d *DiskQueueSnapshot) SeekToEnd() error

func (*DiskQueueSnapshot) UpdateQueueEnd

func (d *DiskQueueSnapshot) UpdateQueueEnd(e BackendQueueEnd)

Put writes a []byte to the queue

type FullMessageID

type FullMessageID [MsgIDLength]byte

the new message total id will be ID+TraceID, the length is same with old id slice, the traceid only used for trace for business, the ID is used for internal. In order to be compatible with old format, we keep the attempts field.

type IMsgTracer

type IMsgTracer interface {
	Start()
	TracePub(topic string, traceID uint64, msg *Message, diskOffset BackendOffset, currentCnt int64)
	// state will be READ_QUEUE, Start, Req, Fin, Timeout
	TraceSub(topic string, state string, traceID uint64, msg *Message, clientID string)
}

type IdentifyDataV2

type IdentifyDataV2 struct {
	ShortID string `json:"short_id"` // TODO: deprecated, remove in 1.0
	LongID  string `json:"long_id"`  // TODO: deprecated, remove in 1.0

	ClientID            string `json:"client_id"`
	Hostname            string `json:"hostname"`
	HeartbeatInterval   int    `json:"heartbeat_interval"`
	OutputBufferSize    int    `json:"output_buffer_size"`
	OutputBufferTimeout int    `json:"output_buffer_timeout"`
	FeatureNegotiation  bool   `json:"feature_negotiation"`
	TLSv1               bool   `json:"tls_v1"`
	Deflate             bool   `json:"deflate"`
	DeflateLevel        int    `json:"deflate_level"`
	Snappy              bool   `json:"snappy"`
	SampleRate          int32  `json:"sample_rate"`
	UserAgent           string `json:"user_agent"`
	MsgTimeout          int    `json:"msg_timeout"`
}

type LogMsgTracer

type LogMsgTracer struct {
	MID string
}

just print the trace log

func (*LogMsgTracer) Start

func (self *LogMsgTracer) Start()

func (*LogMsgTracer) TracePub

func (self *LogMsgTracer) TracePub(topic string, traceID uint64, msg *Message, diskOffset BackendOffset, currentCnt int64)

func (*LogMsgTracer) TraceSub

func (self *LogMsgTracer) TraceSub(topic string, state string, traceID uint64, msg *Message, clientID string)

type Message added in v0.2.29

type Message struct {
	ID        MessageID
	TraceID   uint64
	Body      []byte
	Timestamp int64
	Attempts  uint16
	// contains filtered or unexported fields
}

func DecodeMessage

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

func NewMessage added in v0.2.29

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

func NewMessageWithTs

func NewMessageWithTs(id MessageID, body []byte, ts int64) *Message

func (*Message) GetFullMsgID

func (m *Message) GetFullMsgID() FullMessageID

func (*Message) WriteTo added in v0.2.29

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

func (*Message) WriteToWithDetail

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

type MessageID added in v0.2.29

type MessageID uint64

func GetMessageIDFromFullMsgID

func GetMessageIDFromFullMsgID(id FullMessageID) MessageID

type MsgIDGenerator

type MsgIDGenerator interface {
	NextID() uint64
	Reset(uint64)
}

type NSQD

type NSQD struct {
	sync.RWMutex

	MetaNotifyChan       chan interface{}
	OptsNotificationChan chan struct{}
	// contains filtered or unexported fields
}

func New added in v0.3.6

func New(opts *Options) *NSQD

func (*NSQD) CheckMagicCode

func (n *NSQD) CheckMagicCode(name string, partition int, code int64, tryFix bool) (string, error)

func (*NSQD) CleanClientPubStats

func (n *NSQD) CleanClientPubStats(remote string, protocol string)

func (*NSQD) CloseExistingTopic

func (n *NSQD) CloseExistingTopic(topicName string, partition int) error

this just close the topic and remove from map, but keep the data for later.

func (*NSQD) DeleteExistingTopic

func (n *NSQD) DeleteExistingTopic(topicName string, part int) error

DeleteExistingTopic removes a topic only if it exists

func (*NSQD) Exit

func (n *NSQD) Exit()

func (*NSQD) GetError added in v0.2.31

func (n *NSQD) GetError() error

func (*NSQD) GetExistingTopic

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

GetExistingTopic gets a topic only if it exists

func (*NSQD) GetHealth added in v0.2.31

func (n *NSQD) GetHealth() string

func (*NSQD) GetOpts

func (n *NSQD) GetOpts() *Options

func (*NSQD) GetStartTime added in v0.3.3

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

func (*NSQD) GetStats

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

func (*NSQD) GetTopic

func (n *NSQD) GetTopic(topicName string, part int) *Topic

GetTopic performs a thread safe operation to return a pointer to a Topic object (potentially new)

func (*NSQD) GetTopicDefaultPart

func (n *NSQD) GetTopicDefaultPart(topicName string) int

func (*NSQD) GetTopicIgnPart

func (n *NSQD) GetTopicIgnPart(topicName string) *Topic

func (*NSQD) GetTopicMapCopy

func (n *NSQD) GetTopicMapCopy() map[string]map[int]*Topic

func (*NSQD) GetTopicMapRef

func (n *NSQD) GetTopicMapRef() map[string]map[int]*Topic

should be protected by read lock

func (*NSQD) GetTopicPartitions

func (n *NSQD) GetTopicPartitions(topicName string) map[int]*Topic

func (*NSQD) GetTopicStats

func (n *NSQD) GetTopicStats(topic string) []TopicStats

func (*NSQD) GetTopicWithDisabled

func (n *NSQD) GetTopicWithDisabled(topicName string, part int) *Topic

func (*NSQD) IsAuthEnabled added in v0.2.29

func (n *NSQD) IsAuthEnabled() bool

func (*NSQD) IsHealthy added in v0.2.31

func (n *NSQD) IsHealthy() bool

func (*NSQD) LoadMetadata

func (n *NSQD) LoadMetadata(disabled int32)

func (*NSQD) Notify

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

func (*NSQD) PersistMetadata

func (n *NSQD) PersistMetadata(currentTopicMap map[string]map[int]*Topic) error

func (*NSQD) SetHealth added in v0.2.31

func (n *NSQD) SetHealth(err error)

func (*NSQD) SetTopicMagicCode

func (n *NSQD) SetTopicMagicCode(t *Topic, code int64) error

func (*NSQD) Start

func (n *NSQD) Start()

func (*NSQD) SwapOpts

func (n *NSQD) SwapOpts(opts *Options)

func (*NSQD) TriggerOptsNotification

func (n *NSQD) TriggerOptsNotification()

type Options added in v0.3.6

type Options struct {
	// basic options
	ID                         int64         `flag:"worker-id" cfg:"id"`
	Verbose                    bool          `flag:"verbose"`
	ClusterID                  string        `flag:"cluster-id"`
	ClusterLeadershipAddresses string        `flag:"cluster-leadership-addresses" cfg:"cluster_leadership_addresses"`
	TCPAddress                 string        `flag:"tcp-address"`
	RPCPort                    string        `flag:"rpc-port"`
	ReverseProxyPort           string        `flag:"reverse-proxy-port"`
	HTTPAddress                string        `flag:"http-address"`
	HTTPSAddress               string        `flag:"https-address"`
	BroadcastAddress           string        `flag:"broadcast-address"`
	BroadcastInterface         string        `flag:"broadcast-interface"`
	NSQLookupdTCPAddresses     []string      `flag:"lookupd-tcp-address" cfg:"nsqlookupd_tcp_addresses"`
	AuthHTTPAddresses          []string      `flag:"auth-http-address" cfg:"auth_http_addresses"`
	LookupPingInterval         time.Duration `flag:"lookup-ping-interval" arg:"5s"`

	// diskqueue options
	DataPath        string        `flag:"data-path"`
	MemQueueSize    int64         `flag:"mem-queue-size"`
	MaxBytesPerFile int64         `flag:"max-bytes-per-file"`
	SyncEvery       int64         `flag:"sync-every"`
	SyncTimeout     time.Duration `flag:"sync-timeout"`

	QueueScanInterval        time.Duration
	QueueScanRefreshInterval time.Duration
	QueueScanSelectionCount  int
	QueueScanWorkerPoolMax   int
	QueueScanDirtyPercent    float64

	// msg and command options
	MsgTimeout    time.Duration `flag:"msg-timeout" arg:"60s"`
	MaxMsgTimeout time.Duration `flag:"max-msg-timeout"`
	MaxMsgSize    int64         `flag:"max-msg-size" deprecated:"max-message-size" cfg:"max_msg_size"`
	MaxBodySize   int64         `flag:"max-body-size"`
	MaxReqTimeout time.Duration `flag:"max-req-timeout"`
	MaxConfirmWin int64         `flag:"max-confirm-win"`
	ClientTimeout time.Duration

	// client overridable configuration options
	MaxHeartbeatInterval   time.Duration `flag:"max-heartbeat-interval"`
	MaxRdyCount            int64         `flag:"max-rdy-count"`
	MaxOutputBufferSize    int64         `flag:"max-output-buffer-size"`
	MaxOutputBufferTimeout time.Duration `flag:"max-output-buffer-timeout"`

	// statsd integration
	StatsdAddress  string        `flag:"statsd-address"`
	StatsdPrefix   string        `flag:"statsd-prefix"`
	StatsdProtocol string        `flag:"statsd-protocol"`
	StatsdInterval time.Duration `flag:"statsd-interval" arg:"60s"`
	StatsdMemStats bool          `flag:"statsd-mem-stats"`

	// e2e message latency
	E2EProcessingLatencyWindowTime  time.Duration `flag:"e2e-processing-latency-window-time"`
	E2EProcessingLatencyPercentiles []float64     `flag:"e2e-processing-latency-percentile" cfg:"e2e_processing_latency_percentiles"`

	// TLS config
	TLSCert             string `flag:"tls-cert"`
	TLSKey              string `flag:"tls-key"`
	TLSClientAuthPolicy string `flag:"tls-client-auth-policy"`
	TLSRootCAFile       string `flag:"tls-root-ca-file"`
	TLSRequired         int    `flag:"tls-required"`
	TLSMinVersion       uint16 `flag:"tls-min-version"`

	// compression
	DeflateEnabled  bool `flag:"deflate"`
	MaxDeflateLevel int  `flag:"max-deflate-level"`
	SnappyEnabled   bool `flag:"snappy"`

	LogLevel int32  `flag:"log-level" cfg:"log_level"`
	LogDir   string `flag:"log-dir" cfg:"log_dir"`
	Logger   levellogger.Logger
}

func NewOptions added in v0.3.6

func NewOptions() *Options

type ReadResult

type ReadResult struct {
	Offset    BackendOffset
	MovedSize BackendOffset
	CurCnt    int64
	Data      []byte
	Err       error
}

type RemoteMsgTracer

type RemoteMsgTracer struct {
	// local machine id
	MID string
	// contains filtered or unexported fields
}

this tracer will send the trace info to remote server for each seconds

func (*RemoteMsgTracer) Start

func (self *RemoteMsgTracer) Start()

func (*RemoteMsgTracer) TracePub

func (self *RemoteMsgTracer) TracePub(topic string, traceID uint64, msg *Message, diskOffset BackendOffset, currentCnt int64)

func (*RemoteMsgTracer) TraceSub

func (self *RemoteMsgTracer) TraceSub(topic string, state string, traceID uint64, msg *Message, clientID string)

type Topic

type Topic struct {
	sync.Mutex

	EnableTrace int32
	// contains filtered or unexported fields
}

func NewTopic

func NewTopic(topicName string, part int, opt *Options,
	deleteCallback func(*Topic), writeDisabled int32,
	notify func(v interface{})) *Topic

Topic constructor

func (*Topic) AggregateChannelE2eProcessingLatency

func (t *Topic) AggregateChannelE2eProcessingLatency() *quantile.Quantile

func (*Topic) BufferPoolGet

func (t *Topic) BufferPoolGet(capacity int) *bytes.Buffer

func (*Topic) BufferPoolPut

func (t *Topic) BufferPoolPut(b *bytes.Buffer)

func (*Topic) Close

func (t *Topic) Close() error

Close persists all outstanding topic data and closes all its channels

func (*Topic) Delete

func (t *Topic) Delete() error

Delete empties the topic and all its channels and closes

func (*Topic) DeleteExistingChannel

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

DeleteExistingChannel removes a channel from the topic only if it exists

func (*Topic) DisableForSlave

func (t *Topic) DisableForSlave()

func (*Topic) Empty

func (t *Topic) Empty() error

func (*Topic) EnableForMaster

func (t *Topic) EnableForMaster()

func (*Topic) Exiting

func (t *Topic) Exiting() bool

Exiting returns a boolean indicating if this topic is closed/exiting

func (*Topic) ForceFlush

func (t *Topic) ForceFlush()

func (*Topic) GetChannel

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

GetChannel performs a thread safe operation to return a pointer to a Channel object (potentially new) for the given Topic

func (*Topic) GetChannelMapCopy

func (t *Topic) GetChannelMapCopy() map[string]*Channel

should be protected by read lock outside

func (*Topic) GetCommitted

func (t *Topic) GetCommitted() BackendQueueEnd

func (*Topic) GetDiskQueueSnapshot

func (t *Topic) GetDiskQueueSnapshot() *DiskQueueSnapshot

func (*Topic) GetExistingChannel

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

func (*Topic) GetFullName

func (t *Topic) GetFullName() string

func (*Topic) GetMagicCode

func (t *Topic) GetMagicCode() int64

should be protected by the topic lock for all partitions

func (*Topic) GetMsgGenerator

func (t *Topic) GetMsgGenerator() MsgIDGenerator

func (*Topic) GetPubStats

func (t *Topic) GetPubStats() []ClientPubStats

func (*Topic) GetTopicChannelDebugStat

func (t *Topic) GetTopicChannelDebugStat(channelName string) string

func (*Topic) GetTopicName

func (t *Topic) GetTopicName() string

func (*Topic) GetTopicPart

func (t *Topic) GetTopicPart() int

func (*Topic) IsWriteDisabled

func (t *Topic) IsWriteDisabled() bool

func (*Topic) MarkAsRemoved

func (t *Topic) MarkAsRemoved() (string, error)

func (*Topic) NotifyReloadChannels

func (t *Topic) NotifyReloadChannels()

func (*Topic) PrintCurrentStats

func (t *Topic) PrintCurrentStats()

func (*Topic) PutMessage

PutMessage writes a Message to the queue

func (*Topic) PutMessageNoLock

func (t *Topic) PutMessageNoLock(m *Message) (MessageID, BackendOffset, int32, BackendQueueEnd, error)

func (*Topic) PutMessageOnReplica

func (t *Topic) PutMessageOnReplica(m *Message, offset BackendOffset) (BackendQueueEnd, error)

func (*Topic) PutMessages

func (t *Topic) PutMessages(msgs []*Message) (MessageID, BackendOffset, int32, int64, BackendQueueEnd, error)

PutMessages writes multiple Messages to the queue

func (*Topic) PutMessagesNoLock

func (t *Topic) PutMessagesNoLock(msgs []*Message) (MessageID, BackendOffset, int32, int64, BackendQueueEnd, error)

func (*Topic) PutMessagesOnReplica

func (t *Topic) PutMessagesOnReplica(msgs []*Message, offset BackendOffset) (BackendQueueEnd, error)

func (*Topic) RemovePubStats

func (t *Topic) RemovePubStats(remote string, protocol string)

func (*Topic) ResetBackendEndNoLock

func (t *Topic) ResetBackendEndNoLock(vend BackendOffset, totalCnt int64) error

func (*Topic) RollbackNoLock

func (t *Topic) RollbackNoLock(vend BackendOffset, diffCnt uint64) error

func (*Topic) SetAutoCommit

func (t *Topic) SetAutoCommit(enable bool)

func (*Topic) SetMagicCode

func (t *Topic) SetMagicCode(code int64) error

should be protected by the topic lock for all partitions

func (*Topic) SetMsgGenerator

func (t *Topic) SetMsgGenerator(idGen MsgIDGenerator)

func (*Topic) SetTrace

func (t *Topic) SetTrace(enable bool)

func (*Topic) TotalDataSize

func (t *Topic) TotalDataSize() int64

func (*Topic) TotalMessageCnt

func (t *Topic) TotalMessageCnt() uint64

func (*Topic) UpdateCommittedOffset

func (t *Topic) UpdateCommittedOffset(offset BackendQueueEnd)

note: multiple writer should be protected by lock

func (*Topic) UpdatePubStats

func (t *Topic) UpdatePubStats(remote string, agent string, protocol string, count int64, hasErr bool)

type TopicStats

type TopicStats struct {
	TopicName      string           `json:"topic_name"`
	TopicFullName  string           `json:"topic_full_name"`
	TopicPartition string           `json:"topic_partition"`
	Channels       []ChannelStats   `json:"channels"`
	Depth          int64            `json:"depth"`
	BackendDepth   int64            `json:"backend_depth"`
	MessageCount   uint64           `json:"message_count"`
	IsLeader       bool             `json:"is_leader"`
	HourlyPubSize  int64            `json:"hourly_pubsize"`
	Clients        []ClientPubStats `json:"client_pub_stats"`

	E2eProcessingLatency *quantile.Result `json:"e2e_processing_latency"`
}

func NewTopicStats

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

type Topics

type Topics []*Topic

func (Topics) Len

func (t Topics) Len() int

func (Topics) Swap

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

type TopicsByName

type TopicsByName struct {
	Topics
}

func (TopicsByName) Less

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

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL