dataport

Documentation

Index

• Variables
• func Application(addr string, stats int, timeout int, maxvbs int, config c.Config, ...)
• func ProtobufVersion() byte
• type Average
  • func (av *Average) Add(sample int64)
  • func (av *Average) Count() int64
  • func (av *Average) Max() int64
  • func (av *Average) Mean() int64
  • func (av *Average) Min() int64
  • func (av *Average) Sd() int64
  • func (av *Average) Sum() int64
  • func (av *Average) Variance() int64
• type Client
  • func NewClient(cluster, topic, raddr string, flags transport.TransportFlag, maxvbs int, ...) (c *Client, err error)
  • func (c *Client) Close() error
  • func (c *Client) Getcontext() ([]interface{}, error)
  • func (c *Client) SendKeyVersions(vbs []*common.VbKeyVersions, block bool) error
  • func (c *Client) SendVbmap(vbmap *common.VbConnectionMap) error
• type ConnectionError
  • func NewConnectionError() ConnectionError
  • func (ce ConnectionError) Append(hostUuids keeper, raddr string) keeper
• type RouterEndpoint
  • func NewRouterEndpoint(cluster, topic, raddr string, maxvbs int, config c.Config) (*RouterEndpoint, error)
  • func (endpoint *RouterEndpoint) Close() error
  • func (endpoint *RouterEndpoint) GetStatistics() map[string]interface{}
  • func (endpoint *RouterEndpoint) Ping() bool
  • func (endpoint *RouterEndpoint) ResetConfig(config c.Config) error
  • func (endpoint *RouterEndpoint) Send(data interface{}) error
  • func (endpoint *RouterEndpoint) WaitForExit() error
• type Server
  • func NewServer(laddr string, maxvbs int, config c.Config, appch chan<- interface{}) (s *Server, err error)
  • func (s *Server) Close() (err error)

Variables

var ErrorClientEmptyKeys = errors.New("dataport.clientEmptyKeys")

ErrorClientEmptyKeys

var ErrorDaemonExit = errors.New("dataport.daemonExit")

ErrorDaemonExit

var ErrorDuplicateClient = errors.New("dataport.duplicateClient")

ErrorDuplicateClient

var ErrorDuplicateStreamBegin = errors.New("dataport.duplicateStreamBegin")

ErrorDuplicateStreamBegin

var ErrorMissingPayload = errors.New("dataport.missingPlayload")

ErrorMissingPayload

var ErrorMissingStreamBegin = errors.New("dataport.missingStreamBegin")

ErrorMissingStreamBegin

var ErrorPayload = errors.New("dataport.daemonPayload")

ErrorPayload

var ErrorTransportVersion = errors.New("dataport.transportVersion")

ErrorTransportVersion

var ErrorWorkerKilled = errors.New("dataport.workerKilled")

ErrorWorkerKilled

Functions

func Application

func Application(
	addr string,
	stats int,
	timeout int,
	maxvbs int,
	config c.Config,
	callb func(string, interface{}) bool,
)

Application starts a new dataport application to receive mutations from the other end. Optionally it can print statistics and do callback for every mutation received.

func ProtobufVersion

func ProtobufVersion() byte

ProtobufVersion return version of protobuf schema used in packet transport.

Types

type Average

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

Average maintains the average and variance of a stream of numbers in a space-efficient manner.

func (*Average) Add

func (av *Average) Add(sample int64)

Add a sample to counting average.

func (*Average) Count

func (av *Average) Count() int64

Count return the number of samples counted so far.

func (*Average) Max

func (av *Average) Max() int64

Max return the maximum value of sample.

func (*Average) Mean

func (av *Average) Mean() int64

Mean return the sum of all samples by number of samples so far.

func (*Average) Min

func (av *Average) Min() int64

Min return the minimum value of sample.

func (*Average) Sd

func (av *Average) Sd() int64

GetStdDev return the standard-deviation of all samples so far.

func (*Average) Sum

func (av *Average) Sum() int64

GetTotal return the sum of all samples so far.

func (*Average) Variance

func (av *Average) Variance() int64

Variance return the variance of all samples so far.

type Client

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

Client is an active client for each remote host, and there can be multiple connections opened with remote host for the same endpoint.

func NewClient

func NewClient(
	cluster, topic, raddr string,
	flags transport.TransportFlag,
	maxvbs int,
	config common.Config) (c *Client, err error)

NewClient returns a pool of connection. Multiple connections, based on parameter `n`, can be used to speed up mutation transport across network. A vbucket is always binded to a connection and ensure that mutations within a vbucket are serialized.

func (*Client) Close

func (c *Client) Close() error

Close client and all its active connection with downstream. asynchronous call.

func (*Client) Getcontext

func (c *Client) Getcontext() ([]interface{}, error)

Getcontext from dataport client, synchronous call.

func (*Client) SendKeyVersions

func (c *Client) SendKeyVersions(vbs []*common.VbKeyVersions, block bool) error

SendKeyVersions for one or more vbuckets to the other end, asynchronous call.

func (*Client) SendVbmap

func (c *Client) SendVbmap(vbmap *common.VbConnectionMap) error

SendVbmap vbmap for this connection to the other end, synchronous call.

type ConnectionError

type ConnectionError map[string][]uint16 // bucket -> []vbuckets

ConnectionError to application

func NewConnectionError

func NewConnectionError() ConnectionError

NewConnectionError makes a new connection-error map.

func (ConnectionError) Append

func (ce ConnectionError) Append(hostUuids keeper, raddr string) keeper

Append {buckets,vbuckets} for connection error.

type RouterEndpoint

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

RouterEndpoint structure, per topic, to gather key-versions / mutations from one or more vbuckets and push them downstream to a specific node.

func NewRouterEndpoint

func NewRouterEndpoint(
	cluster, topic, raddr string, maxvbs int,
	config c.Config) (*RouterEndpoint, error)

NewRouterEndpoint instantiate a new RouterEndpoint routine and return its reference.

func (*RouterEndpoint) Close

func (endpoint *RouterEndpoint) Close() error

Close this endpoint.

func (*RouterEndpoint) GetStatistics

func (endpoint *RouterEndpoint) GetStatistics() map[string]interface{}

GetStatistics for this endpoint, synchronous call.

func (*RouterEndpoint) Ping

func (endpoint *RouterEndpoint) Ping() bool

Ping whether endpoint is active, synchronous call.

func (*RouterEndpoint) ResetConfig

func (endpoint *RouterEndpoint) ResetConfig(config c.Config) error

ResetConfig synchronous call.

func (*RouterEndpoint) Send

func (endpoint *RouterEndpoint) Send(data interface{}) error

Send KeyVersions to other end, asynchronous call. Asynchronous call. Return ErrorChannelFull that can be used by caller.

func (*RouterEndpoint) WaitForExit

func (endpoint *RouterEndpoint) WaitForExit() error

WaitForExit will block until endpoint exits.

type Server

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

Server handles an active dataport server of mutation for all vbuckets.

func NewServer

func NewServer(
	laddr string,
	maxvbs int,
	config c.Config,
	appch chan<- interface{}) (s *Server, err error)

NewServer creates a new dataport daemon.

func (*Server) Close

func (s *Server) Close() (err error)

Close the daemon listening for new connections and shuts down all read routines for this dataport server. synchronous call.