memcached

package
Version: v0.1.4 Latest Latest
Warning

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

Go to latest
Published: Sep 30, 2021 License: MIT Imports: 18 Imported by: 159

Documentation

Overview

Package memcached provides a memcached binary protocol client.

go implementation of upr client. See https://github.com/couchbaselabs/cbupr/blob/master/transport-spec.md TODO 1. Use a pool allocator to avoid garbage

Index

Constants

View Source
const (
	FeatureTcpNoDelay        = Feature(0x03)
	FeatureMutationToken     = Feature(0x04) // XATTR bit in data type field with dcp mutations
	FeatureXattr             = Feature(0x06)
	FeatureXerror            = Feature(0x07)
	FeatureSnappyCompression = Feature(0x0a)
	FeatureDataType          = Feature(0x0b)
	FeatureSyncReplication   = Feature(0x11)
	FeatureCollections       = Feature(0x12)
	FeatureOpenTracing       = Feature(0x13)
	FeaturePreserveExpiry    = Feature(0x14)
)
View Source
const (
	ObservedNotPersisted     = ObservedStatus(0x00) // found, not persisted
	ObservedPersisted        = ObservedStatus(0x01) // found, persisted
	ObservedNotFound         = ObservedStatus(0x80) // not found (or a persisted delete)
	ObservedLogicallyDeleted = ObservedStatus(0x81) // pending deletion (not persisted yet)
)

Observation status values.

View Source
const (
	// CASStore instructs the server to store the new value normally
	CASStore = CasOp(iota)
	// CASQuit instructs the client to stop attempting to CAS, leaving value untouched
	CASQuit
	// CASDelete instructs the server to delete the current value
	CASDelete
)
View Source
const (
	TapBeginBackfill = TapOpcode(iota)
	TapEndBackfill
	TapMutation
	TapDeletion
	TapCheckpointStart
	TapCheckpointEnd
)

Tap opcode values.

View Source
const (
	FeedStateInitial = iota
	FeedStateOpened  = iota
	FeedStateClosed  = iota
)
View Source
const (
	CompressionTypeStartMarker = iota // also means invalid
	CompressionTypeNone        = iota
	CompressionTypeSnappy      = iota
	CompressionTypeEndMarker   = iota // also means invalid
)
View Source
const (
	JSONDataType   uint8 = 1
	SnappyDataType uint8 = 2
	XattrDataType  uint8 = 4
)

kv_engine/include/mcbp/protocol/datatype.h

View Source
const TapNoBackfill = math.MaxUint64

Value for TapArguments.Backfill denoting that no past events at all should be sent.

Variables

View Source
var (
	ErrUnSuccessfulHello          = errors.New("Unsuccessful HELLO exchange")
	ErrInvalidHello               = errors.New("Invalid HELLO response")
	ErrPreserveExpiryNotSupported = errors.New("PreserveExpiry is not supported")
	ErrDurabilityNotSupported     = errors.New("Durability is not supported")
)
View Source
var (
	DefaultDialTimeout = time.Duration(0) // No timeout

	DefaultWriteTimeout = time.Duration(0) // No timeout

)
View Source
var ErrorInvalidLog = errors.New("couchbase.errorInvalidLog")

error codes

View Source
var ErrorInvalidOp error = fmt.Errorf("Invalid Operation")
View Source
var ErrorInvalidVersion error = fmt.Errorf("Invalid version for parsing")
View Source
var ErrorNoMaxTTL error = fmt.Errorf("This event has no max TTL")
View Source
var ErrorValueTooShort error = fmt.Errorf("Value length is too short")
View Source
var Healthy uint32 = 1
View Source
var ReceiveHook func(*gomemcached.MCResponse, int, error)

ReceiveHook is called after every packet is received (or attempted to be)

View Source
var TapRecvHook func(*gomemcached.MCRequest, int, error)

TapRecvHook is called after every incoming tap packet is received.

View Source
var TransmitHook func(*gomemcached.MCRequest, int, error)

TransmitHook is called after each packet is transmitted.

View Source
var UnHealthy uint32 = 0

Functions

func GetSubDocVal

func GetSubDocVal(subPaths []string) (extraBuf, valueBuf []byte)

func IfResStatusError

func IfResStatusError(response *gomemcached.MCResponse) bool

func SetDefaultDialTimeout

func SetDefaultDialTimeout(dial time.Duration)

func SetDefaultTimeouts

func SetDefaultTimeouts(dial, read, write time.Duration)

func UnwrapMemcachedError

func UnwrapMemcachedError(rv *gomemcached.MCResponse,
	err error) (*gomemcached.MCResponse, error)

UnwrapMemcachedError converts memcached errors to normal responses.

If the error is a memcached response, declare the error to be nil so a client can handle the status without worrying about whether it indicates success or failure.

Types

type CASState

type CASState struct {
	Value  []byte // Current value of key; update in place to new value
	Cas    uint64 // Current CAS value of key
	Exists bool   // Does a value exist for the key? (If not, Value will be nil)
	Err    error  // Error, if any, after CASNext returns false
	// contains filtered or unexported fields
}

CASState tracks the state of CAS over several operations.

This is used directly by CASNext and indirectly by CAS

type CasFunc

type CasFunc func(current []byte) ([]byte, CasOp)

CasFunc is type type of function to perform a CAS transform.

Input is the current value, or nil if no value exists. The function should return the new value (if any) to set, and the store/quit/delete operation.

type CasOp

type CasOp uint8

CasOp is the type of operation to perform on this CAS loop.

func (CasOp) Error

func (c CasOp) Error() string

User specified termination is returned as an error.

type Client

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

The Client itself.

func Connect

func Connect(prot, dest string) (rv *Client, err error)

Connect to a memcached server.

func ConnectTLS

func ConnectTLS(prot, dest string, config *tls.Config) (rv *Client, err error)

Connect to a memcached server using TLS.

func Wrap

func Wrap(conn memcachedConnection) (rv *Client, err error)

Wrap an existing transport.

func (*Client) Add

func (c *Client) Add(vb uint16, key string, flags int, exp int,
	body []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)

Add a value for a key (store if not exists).

func (*Client) Append

func (c *Client) Append(vb uint16, key string, data []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)

Append data to the value of a key.

func (*Client) Auth

func (c *Client) Auth(user, pass string) (*gomemcached.MCResponse, error)

Auth performs SASL PLAIN authentication against the server.

func (*Client) AuthList

func (c *Client) AuthList() (*gomemcached.MCResponse, error)

AuthList lists SASL auth mechanisms.

func (*Client) AuthPlain

func (c *Client) AuthPlain(user, pass string) (*gomemcached.MCResponse, error)

func (*Client) AuthScramSha

func (c *Client) AuthScramSha(user, pass string) (*gomemcached.MCResponse, error)

AuthScramSha performs SCRAM-SHA authentication against the server.

func (*Client) CAS

func (c *Client) CAS(vb uint16, k string, f CasFunc,
	initexp int) (*gomemcached.MCResponse, error)

CAS performs a CAS transform with the given function.

If the value does not exist, a nil current value will be sent to f.

func (*Client) CASNext

func (c *Client) CASNext(vb uint16, k string, exp int, state *CASState) bool

CASNext is a non-callback, loop-based version of CAS method.

Usage is like this:

var state memcached.CASState for client.CASNext(vb, key, exp, &state) {

state.Value = some_mutation(state.Value)

} if state.Err != nil { ... }

func (*Client) Close

func (c *Client) Close() error

Close the connection when you're done.

func (*Client) CollectionEnabled

func (c *Client) CollectionEnabled() bool

func (*Client) CollectionsGetCID

func (c *Client) CollectionsGetCID(scope string, collection string) (*gomemcached.MCResponse, error)

Retrieve the collections manifest.

func (*Client) Conn

func (c *Client) Conn() io.ReadWriteCloser

func (*Client) Decr

func (c *Client) Decr(vb uint16, key string,
	amt, def uint64, exp int, context ...*ClientContext) (uint64, error)

Decr decrements the value at the given key.

func (*Client) Del

func (c *Client) Del(vb uint16, key string, context ...*ClientContext) (*gomemcached.MCResponse, error)

Del deletes a key.

func (*Client) EnableFeatures

func (c *Client) EnableFeatures(features Features) (*gomemcached.MCResponse, error)

Send a hello command to enable specific features

func (*Client) EnableMutationToken

func (c *Client) EnableMutationToken() (*gomemcached.MCResponse, error)

Send a hello command to enable MutationTokens

func (*Client) Get

func (c *Client) Get(vb uint16, key string, context ...*ClientContext) (*gomemcached.MCResponse, error)

Get the value for a key.

func (*Client) GetAllVbSeqnos

func (c *Client) GetAllVbSeqnos(vbSeqnoMap map[uint16]uint64, context ...*ClientContext) (map[uint16]uint64, error)

Since the binary request supports only a single collection at a time, it is possible that this may be called multiple times in succession by callers to get vbSeqnos for multiple collections. Thus, caller could pass in a non-nil map so the gomemcached client won't need to allocate new map for each call to prevent too much GC NOTE: If collection is enabled and context is not given, KV will still return stats for default collection

func (*Client) GetAndTouch

func (c *Client) GetAndTouch(vb uint16, key string, exp int, context ...*ClientContext) (*gomemcached.MCResponse, error)

Get the value for a key, and update expiry

func (*Client) GetBulk

func (c *Client) GetBulk(vb uint16, keys []string, rv map[string]*gomemcached.MCResponse, subPaths []string, context ...*ClientContext) error

GetBulk gets keys in bulk

func (*Client) GetCollectionsManifest

func (c *Client) GetCollectionsManifest() (*gomemcached.MCResponse, error)

Retrieve the collections manifest.

func (*Client) GetMeta

func (c *Client) GetMeta(vb uint16, key string, context ...*ClientContext) (*gomemcached.MCResponse, error)

Get metadata for a key

func (*Client) GetRandomDoc

func (c *Client) GetRandomDoc(context ...*ClientContext) (*gomemcached.MCResponse, error)

Get a random document

func (*Client) GetSubdoc

func (c *Client) GetSubdoc(vb uint16, key string, subPaths []string, context ...*ClientContext) (*gomemcached.MCResponse, error)

Get the xattrs, doc value for the input key

func (*Client) Hijack

func (c *Client) Hijack() io.ReadWriteCloser

Hijack exposes the underlying connection from this client.

It also marks the connection as unhealthy since the client will have lost control over the connection and can't otherwise verify things are in good shape for connection pools.

func (*Client) Incr

func (c *Client) Incr(vb uint16, key string,
	amt, def uint64, exp int, context ...*ClientContext) (uint64, error)

Incr increments the value at the given key.

func (*Client) IsFeatureEnabled added in v0.1.4

func (c *Client) IsFeatureEnabled(feature Feature) bool

func (Client) IsHealthy

func (c Client) IsHealthy() bool

IsHealthy returns true unless the client is belived to have difficulty communicating to its server.

This is useful for connection pools where we want to non-destructively determine that a connection may be reused.

func (*Client) LastBucket added in v0.1.2

func (c *Client) LastBucket() string

func (*Client) NewUprFeed

func (mc *Client) NewUprFeed() (*UprFeed, error)

NewUprFeed creates a new UPR Feed. TODO: Describe side-effects on bucket instance and its connection pool.

func (*Client) NewUprFeedIface

func (mc *Client) NewUprFeedIface() (UprFeedIface, error)

func (*Client) NewUprFeedWithConfig

func (mc *Client) NewUprFeedWithConfig(ackByClient bool) (*UprFeed, error)

func (*Client) NewUprFeedWithConfigIface

func (mc *Client) NewUprFeedWithConfigIface(ackByClient bool) (UprFeedIface, error)

func (*Client) Observe

func (c *Client) Observe(vb uint16, key string) (result ObserveResult, err error)

Observe gets the persistence/replication/CAS state of a key

func (*Client) ObserveSeq

func (c *Client) ObserveSeq(vb uint16, vbuuid uint64) (result *ObserveSeqResult, err error)

func (*Client) Receive

func (c *Client) Receive() (*gomemcached.MCResponse, error)

Receive a response

func (*Client) ReceiveWithDeadline

func (c *Client) ReceiveWithDeadline(deadline time.Time) (*gomemcached.MCResponse, error)

func (*Client) SelectBucket

func (c *Client) SelectBucket(bucket string) (*gomemcached.MCResponse, error)

select bucket

func (*Client) Send

func (c *Client) Send(req *gomemcached.MCRequest) (rv *gomemcached.MCResponse, err error)

Send a custom request and get the response.

func (*Client) Set

func (c *Client) Set(vb uint16, key string, flags int, exp int,
	body []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)

Set the value for a key.

func (*Client) SetCas

func (c *Client) SetCas(vb uint16, key string, flags int, exp int, cas uint64,
	body []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)

SetCas set the value for a key with cas

func (*Client) SetDeadline

func (c *Client) SetDeadline(t time.Time)

func (*Client) SetKeepAliveOptions

func (c *Client) SetKeepAliveOptions(interval time.Duration)

func (*Client) SetReadDeadline

func (c *Client) SetReadDeadline(t time.Time)

func (*Client) SetReplica added in v0.1.4

func (c *Client) SetReplica(r bool)

Read from replica setting

func (*Client) StartTapFeed

func (mc *Client) StartTapFeed(args TapArguments) (*TapFeed, error)

StartTapFeed starts a TAP feed on a client connection.

The events can be read from the returned channel. The connection can no longer be used for other purposes; it's now reserved for receiving the TAP messages. To stop receiving events, close the client connection.

func (*Client) Stats

func (c *Client) Stats(key string) ([]StatValue, error)

Stats requests server-side stats.

Use "" as the stat key for toplevel stats.

func (*Client) StatsFunc added in v0.1.2

func (c *Client) StatsFunc(key string, fn func(key, val []byte)) error

Stats requests server-side stats.

Use "" as the stat key for toplevel stats.

func (*Client) StatsMap

func (c *Client) StatsMap(key string) (map[string]string, error)

StatsMap requests server-side stats similarly to Stats, but returns them as a map.

Use "" as the stat key for toplevel stats.

func (*Client) StatsMapForSpecifiedStats

func (c *Client) StatsMapForSpecifiedStats(key string, statsMap map[string]string) error

instead of returning a new statsMap, simply populate passed in statsMap, which contains all the keys for which stats needs to be retrieved

func (*Client) Transmit

func (c *Client) Transmit(req *gomemcached.MCRequest) error

Transmit send a request, but does not wait for a response.

func (*Client) TransmitResponse

func (c *Client) TransmitResponse(res *gomemcached.MCResponse) error

TransmitResponse send a response, does not wait.

func (*Client) TransmitWithDeadline

func (c *Client) TransmitWithDeadline(req *gomemcached.MCRequest, deadline time.Time) error

func (*Client) UprGetFailoverLog

func (mc *Client) UprGetFailoverLog(vb []uint16) (map[uint16]*FailoverLog, error)

UprGetFailoverLog for given list of vbuckets.

type ClientContext

type ClientContext struct {
	// Collection-based context
	CollId uint32

	// Impersonate context
	User string

	// VB-state related context
	// nil means not used in this context
	VbState *VbStateType

	// Preserve Expiry
	PreserveExpiry bool

	// Durability Level
	DurabilityLevel gomemcached.DurabilityLvl

	// Durability Timeout
	DurabilityTimeout time.Duration
}

func (*ClientContext) InitExtras

func (context *ClientContext) InitExtras(req *gomemcached.MCRequest, client *Client)

type ClientIface

type ClientIface interface {
	Add(vb uint16, key string, flags int, exp int, body []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)
	Append(vb uint16, key string, data []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)
	Auth(user, pass string) (*gomemcached.MCResponse, error)
	AuthList() (*gomemcached.MCResponse, error)
	AuthPlain(user, pass string) (*gomemcached.MCResponse, error)
	AuthScramSha(user, pass string) (*gomemcached.MCResponse, error)
	CASNext(vb uint16, k string, exp int, state *CASState) bool
	CAS(vb uint16, k string, f CasFunc, initexp int) (*gomemcached.MCResponse, error)
	CollectionsGetCID(scope string, collection string) (*gomemcached.MCResponse, error)
	CollectionEnabled() bool
	Close() error
	Decr(vb uint16, key string, amt, def uint64, exp int, context ...*ClientContext) (uint64, error)
	Del(vb uint16, key string, context ...*ClientContext) (*gomemcached.MCResponse, error)
	EnableMutationToken() (*gomemcached.MCResponse, error)
	EnableFeatures(features Features) (*gomemcached.MCResponse, error)
	Get(vb uint16, key string, context ...*ClientContext) (*gomemcached.MCResponse, error)
	GetAllVbSeqnos(vbSeqnoMap map[uint16]uint64, context ...*ClientContext) (map[uint16]uint64, error)
	GetAndTouch(vb uint16, key string, exp int, context ...*ClientContext) (*gomemcached.MCResponse, error)
	GetBulk(vb uint16, keys []string, rv map[string]*gomemcached.MCResponse, subPaths []string, context ...*ClientContext) error
	GetCollectionsManifest() (*gomemcached.MCResponse, error)
	GetMeta(vb uint16, key string, context ...*ClientContext) (*gomemcached.MCResponse, error)
	GetRandomDoc(context ...*ClientContext) (*gomemcached.MCResponse, error)
	GetSubdoc(vb uint16, key string, subPaths []string, context ...*ClientContext) (*gomemcached.MCResponse, error)
	Hijack() io.ReadWriteCloser
	Incr(vb uint16, key string, amt, def uint64, exp int, context ...*ClientContext) (uint64, error)
	LastBucket() string
	Observe(vb uint16, key string) (result ObserveResult, err error)
	ObserveSeq(vb uint16, vbuuid uint64) (result *ObserveSeqResult, err error)
	Receive() (*gomemcached.MCResponse, error)
	ReceiveWithDeadline(deadline time.Time) (*gomemcached.MCResponse, error)
	Send(req *gomemcached.MCRequest) (rv *gomemcached.MCResponse, err error)
	Set(vb uint16, key string, flags int, exp int, body []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)
	SetKeepAliveOptions(interval time.Duration)
	SetReadDeadline(t time.Time)
	SetDeadline(t time.Time)
	SetReplica(r bool)
	SelectBucket(bucket string) (*gomemcached.MCResponse, error)
	SetCas(vb uint16, key string, flags int, exp int, cas uint64, body []byte, context ...*ClientContext) (*gomemcached.MCResponse, error)
	Stats(key string) ([]StatValue, error)
	StatsFunc(key string, fn func(key, val []byte)) error
	StatsMap(key string) (map[string]string, error)
	StatsMapForSpecifiedStats(key string, statsMap map[string]string) error
	Transmit(req *gomemcached.MCRequest) error
	TransmitWithDeadline(req *gomemcached.MCRequest, deadline time.Time) error
	TransmitResponse(res *gomemcached.MCResponse) error
	UprGetFailoverLog(vb []uint16) (map[uint16]*FailoverLog, error)

	// UprFeed Related
	NewUprFeed() (*UprFeed, error)
	NewUprFeedIface() (UprFeedIface, error)
	NewUprFeedWithConfig(ackByClient bool) (*UprFeed, error)
	NewUprFeedWithConfigIface(ackByClient bool) (UprFeedIface, error)
}

type CollectionChangedEvent

type CollectionChangedEvent interface {
	GetCollectionId() (uint32, error)
	GetManifestId() (uint64, error)
	GetMaxTTL() (uint32, error)
}

type CollectionCreateEvent

type CollectionCreateEvent interface {
	GetSystemEventName() (string, error)
	GetScopeId() (uint32, error)
	GetCollectionId() (uint32, error)
	GetManifestId() (uint64, error)
	GetMaxTTL() (uint32, error)
}

type CollectionDropEvent

type CollectionDropEvent interface {
	GetScopeId() (uint32, error)
	GetCollectionId() (uint32, error)
	GetManifestId() (uint64, error)
}

type CollectionsFilter

type CollectionsFilter struct {
	ManifestUid    uint64
	UseManifestUid bool
	StreamId       uint16
	UseStreamId    bool

	// Use either ScopeId OR CollectionsList, not both
	CollectionsList []uint32
	ScopeId         uint32
}

Collection based filter

func (*CollectionsFilter) IsValid

func (c *CollectionsFilter) IsValid() error

func (*CollectionsFilter) ToStreamReqBody

func (c *CollectionsFilter) ToStreamReqBody() ([]byte, error)

type DcpStreamType

type DcpStreamType int32
const (
	NonCollectionStream    DcpStreamType = 0
	CollectionsNonStreamId DcpStreamType = iota
	CollectionsStreamId    DcpStreamType = iota
)
var UninitializedStream DcpStreamType = -1

func (DcpStreamType) String

func (t DcpStreamType) String() string

type FailoverLog

type FailoverLog [][2]uint64

FailoverLog containing vvuid and sequnce number

func (*FailoverLog) Latest

func (flogp *FailoverLog) Latest() (vbuuid, seqno uint64, err error)

type Feature

type Feature uint16

type Features

type Features []Feature

type FeedState

type FeedState int

func (FeedState) String

func (fs FeedState) String() string

type ObserveResult

type ObserveResult struct {
	Status          ObservedStatus // Whether the value has been persisted/deleted
	Cas             uint64         // Current value's CAS
	PersistenceTime time.Duration  // Node's average time to persist a value
	ReplicationTime time.Duration  // Node's average time to replicate a value
}

ObserveResult represents the data obtained by an Observe call

func (ObserveResult) CheckPersistence

func (result ObserveResult) CheckPersistence(cas uint64, deletion bool) (persisted bool, overwritten bool)

CheckPersistence checks whether a stored value has been persisted to disk yet.

type ObserveSeqResult

type ObserveSeqResult struct {
	Failover           uint8  // Set to 1 if a failover took place
	VbId               uint16 // vbucket id
	Vbuuid             uint64 // vucket uuid
	LastPersistedSeqNo uint64 // last persisted sequence number
	CurrentSeqNo       uint64 // current sequence number
	OldVbuuid          uint64 // Old bucket vbuuid
	LastSeqNo          uint64 // last sequence number received before failover
}

Sequence number based Observe Implementation

type ObservedStatus

type ObservedStatus uint8

ObservedStatus is the type reported by the Observe method

type PriorityType

type PriorityType string
const (
	PriorityDisabled PriorityType = ""
	PriorityLow      PriorityType = "low"
	PriorityMed      PriorityType = "medium"
	PriorityHigh     PriorityType = "high"
)

high > medium > disabled > low

type ScopeCreateEvent

type ScopeCreateEvent interface {
	GetSystemEventName() (string, error)
	GetScopeId() (uint32, error)
	GetManifestId() (uint64, error)
}

type ScopeDropEvent

type ScopeDropEvent interface {
	GetScopeId() (uint32, error)
	GetManifestId() (uint64, error)
}

type StatValue

type StatValue struct {
	// The stat key
	Key string
	// The stat value
	Val string
}

StatValue is one of the stats returned from the Stats method.

type SystemEventType

type SystemEventType int
const (
	CollectionCreate  SystemEventType = 0
	CollectionDrop    SystemEventType = iota
	CollectionFlush   SystemEventType = iota // KV did not implement
	ScopeCreate       SystemEventType = iota
	ScopeDrop         SystemEventType = iota
	CollectionChanged SystemEventType = iota
)
const InvalidSysEvent SystemEventType = -1

type TapArguments

type TapArguments struct {
	// Timestamp of oldest item to send.
	//
	// Use TapNoBackfill to suppress all past items.
	Backfill uint64
	// If set, server will disconnect after sending existing items.
	Dump bool
	// The indices of the vbuckets to watch; empty/nil to watch all.
	VBuckets []uint16
	// Transfers ownership of vbuckets during cluster rebalance.
	Takeover bool
	// If true, server will wait for client ACK after every notification.
	SupportAck bool
	// If true, client doesn't want values so server shouldn't send them.
	KeysOnly bool
	// If true, client wants the server to send checkpoint events.
	Checkpoint bool
	// Optional identifier to use for this client, to allow reconnects
	ClientName string
	// Registers this client (by name) till explicitly deregistered.
	RegisteredClient bool
}

TapArguments are parameters for requesting a TAP feed.

Call DefaultTapArguments to get a default one.

func DefaultTapArguments

func DefaultTapArguments() TapArguments

DefaultTapArguments returns a default set of parameter values to pass to StartTapFeed.

type TapEvent

type TapEvent struct {
	Opcode     TapOpcode // Type of event
	VBucket    uint16    // VBucket this event applies to
	Flags      uint32    // Item flags
	Expiry     uint32    // Item expiration time
	Key, Value []byte    // Item key/value
	Cas        uint64
}

TapEvent is a TAP notification of an operation on the server.

func (TapEvent) String

func (event TapEvent) String() string

type TapFeed

type TapFeed struct {
	C     <-chan TapEvent
	Error error
	// contains filtered or unexported fields
}

TapFeed represents a stream of events from a server.

func (*TapFeed) Close

func (feed *TapFeed) Close()

Close terminates a TapFeed.

Call this if you stop using a TapFeed before its channel ends.

type TapOpcode

type TapOpcode uint8

TapOpcode is the tap operation type (found in TapEvent)

func (TapOpcode) String

func (opcode TapOpcode) String() string

type Uleb128

type Uleb128 []byte

func (Uleb128) ToUint32 added in v0.1.1

func (u Uleb128) ToUint32(cachedLen int) (result uint32, bytesShifted int)

type UprEvent

type UprEvent struct {
	Opcode          gomemcached.CommandCode // Type of event
	Status          gomemcached.Status      // Response status
	VBucket         uint16                  // VBucket this event applies to
	DataType        uint8                   // data type
	Opaque          uint16                  // 16 MSB of opaque
	VBuuid          uint64                  // This field is set by downstream
	Flags           uint32                  // Item flags
	Expiry          uint32                  // Item expiration time
	Key, Value      []byte                  // Item key/value
	OldValue        []byte                  // TODO: TBD: old document value
	Cas             uint64                  // CAS value of the item
	Seqno           uint64                  // sequence number of the mutation
	RevSeqno        uint64                  // rev sequence number : deletions
	LockTime        uint32                  // Lock time
	MetadataSize    uint16                  // Metadata size
	SnapstartSeq    uint64                  // start sequence number of this snapshot
	SnapendSeq      uint64                  // End sequence number of the snapshot
	SnapshotType    uint32                  // 0: disk 1: memory
	FailoverLog     *FailoverLog            // Failover log containing vvuid and sequnce number
	Error           error                   // Error value in case of a failure
	ExtMeta         []byte                  // Extended Metadata
	AckSize         uint32                  // The number of bytes that can be Acked to DCP
	SystemEvent     SystemEventType         // Only valid if IsSystemEvent() is true
	SysEventVersion uint8                   // Based on the version, the way Extra bytes is parsed is different
	ValueLen        int                     // Cache it to avoid len() calls for performance
	CollectionId    uint32                  // Valid if Collection is in use
	StreamId        *uint16                 // Nil if not in use
}

UprEvent memcached events for UPR streams.

func (*UprEvent) GetCollectionId

func (event *UprEvent) GetCollectionId() (uint32, error)

func (*UprEvent) GetManifestId

func (event *UprEvent) GetManifestId() (uint64, error)

func (*UprEvent) GetMaxTTL

func (event *UprEvent) GetMaxTTL() (uint32, error)

func (*UprEvent) GetOsoBegin added in v0.1.1

func (event *UprEvent) GetOsoBegin() (bool, error)

Only if error is nil: Returns true if event states oso begins Return false if event states oso ends

func (*UprEvent) GetScopeId

func (event *UprEvent) GetScopeId() (uint32, error)

func (*UprEvent) GetSystemEventName

func (event *UprEvent) GetSystemEventName() (string, error)

func (*UprEvent) IsCollectionType

func (event *UprEvent) IsCollectionType() bool

func (*UprEvent) IsOsoSnapshot added in v0.1.1

func (event *UprEvent) IsOsoSnapshot() bool

func (*UprEvent) IsSeqnoAdv

func (event *UprEvent) IsSeqnoAdv() bool

func (*UprEvent) IsSnappyDataType

func (event *UprEvent) IsSnappyDataType() bool

func (*UprEvent) IsSystemEvent

func (event *UprEvent) IsSystemEvent() bool

func (*UprEvent) PopulateEvent

func (event *UprEvent) PopulateEvent(extras []byte)

func (*UprEvent) PopulateFieldsBasedOnStreamType

func (event *UprEvent) PopulateFieldsBasedOnStreamType(rq gomemcached.MCRequest, streamType DcpStreamType)

func (*UprEvent) PopulateOso added in v0.1.1

func (event *UprEvent) PopulateOso(extras []byte)

func (*UprEvent) PopulateSeqnoAdv

func (event *UprEvent) PopulateSeqnoAdv(extras []byte)

func (*UprEvent) String

func (event *UprEvent) String() string

type UprFeatures

type UprFeatures struct {
	Xattribute          bool
	CompressionType     int
	IncludeDeletionTime bool
	DcpPriority         PriorityType
	EnableExpiry        bool
	EnableStreamId      bool
	EnableOso           bool
}

type UprFeed

type UprFeed struct {
	C <-chan *UprEvent // Exported channel for receiving UPR events

	Error error // error
	// contains filtered or unexported fields
}

UprFeed represents an UPR feed. A feed contains a connection to a single host and multiple vBuckets

func (*UprFeed) ClientAck

func (feed *UprFeed) ClientAck(event *UprEvent) error

Client, after completing processing of an UprEvent, need to call this API to notify UprFeed, so that UprFeed can update its ack bytes stats and send ack to DCP if needed Client needs to set ackByClient flag to true in NewUprFeedWithConfig() call as a prerequisite for this call to work This API is not thread safe. Caller should NOT have more than one go rountine calling this API

func (*UprFeed) Close

func (feed *UprFeed) Close()

Close this UprFeed.

func (*UprFeed) CloseStream

func (feed *UprFeed) CloseStream(vbno, opaqueMSB uint16) error

CloseStream for specified vbucket.

func (*UprFeed) Closed

func (feed *UprFeed) Closed() bool

check if the UprFeed has been closed

func (*UprFeed) GetError

func (feed *UprFeed) GetError() error

func (*UprFeed) GetUprEventCh

func (feed *UprFeed) GetUprEventCh() <-chan *UprEvent

func (*UprFeed) GetUprStats

func (feed *UprFeed) GetUprStats() *UprStats

func (*UprFeed) SetPriorityAsync

func (feed *UprFeed) SetPriorityAsync(p PriorityType) error

func (*UprFeed) StartFeed

func (feed *UprFeed) StartFeed() error

StartFeed to start the upper feed.

func (*UprFeed) StartFeedWithConfig

func (feed *UprFeed) StartFeedWithConfig(datachan_len int) error

func (*UprFeed) UprOpen

func (feed *UprFeed) UprOpen(name string, sequence uint32, bufSize uint32) error

UprOpen to connect with a UPR producer. Name: name of te UPR connection sequence: sequence number for the connection bufsize: max size of the application

func (*UprFeed) UprOpenWithFeatures

func (feed *UprFeed) UprOpenWithFeatures(name string, sequence uint32, bufSize uint32, features UprFeatures) (error, UprFeatures)

func (*UprFeed) UprOpenWithXATTR

func (feed *UprFeed) UprOpenWithXATTR(name string, sequence uint32, bufSize uint32) error

UprOpen with XATTR enabled.

func (*UprFeed) UprRequestCollectionsStream

func (feed *UprFeed) UprRequestCollectionsStream(vbno, opaqueMSB uint16, flags uint32,
	vbuuid, startSequence, endSequence, snapStart, snapEnd uint64, filter *CollectionsFilter) error

func (*UprFeed) UprRequestStream

func (feed *UprFeed) UprRequestStream(vbno, opaqueMSB uint16, flags uint32,
	vuuid, startSequence, endSequence, snapStart, snapEnd uint64) error

UprRequestStream for a single vbucket.

type UprFeedIface

type UprFeedIface interface {
	Close()
	Closed() bool
	CloseStream(vbno, opaqueMSB uint16) error
	GetError() error
	GetUprStats() *UprStats
	ClientAck(event *UprEvent) error
	GetUprEventCh() <-chan *UprEvent
	StartFeed() error
	StartFeedWithConfig(datachan_len int) error
	UprOpen(name string, sequence uint32, bufSize uint32) error
	UprOpenWithXATTR(name string, sequence uint32, bufSize uint32) error
	UprOpenWithFeatures(name string, sequence uint32, bufSize uint32, features UprFeatures) (error, UprFeatures)
	UprRequestStream(vbno, opaqueMSB uint16, flags uint32, vuuid, startSequence, endSequence, snapStart, snapEnd uint64) error
	// Set DCP priority on an existing DCP connection. The command is sent asynchronously without waiting for a response
	SetPriorityAsync(p PriorityType) error

	// Various Collection-Type RequestStreams
	UprRequestCollectionsStream(vbno, opaqueMSB uint16, flags uint32, vbuuid, startSeq, endSeq, snapStart, snapEnd uint64, filter *CollectionsFilter) error
}

Exported interface - to allow for mocking

type UprStats

type UprStats struct {
	TotalBytes         uint64
	TotalMutation      uint64
	TotalBufferAckSent uint64
	TotalSnapShot      uint64
}

type UprStream

type UprStream struct {
	Vbucket  uint16 // Vbucket id
	Vbuuid   uint64 // vbucket uuid
	StartSeq uint64 // start sequence number
	EndSeq   uint64 // end sequence number

	StreamType DcpStreamType
	// contains filtered or unexported fields
}

UprStream is per stream data structure over an UPR Connection.

type VBSeqnos

type VBSeqnos [][2]uint64

Containing a pair of vbno and the high seqno

type VbStateType

type VbStateType uint8
const (
	VbAlive   VbStateType = 0x00
	VbActive  VbStateType = 0x01
	VbReplica VbStateType = 0x02
	VbPending VbStateType = 0x03
	VbDead    VbStateType = 0x04
)

Directories

Path Synopsis

Jump to

Keyboard shortcuts

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