session

type MarshalFunc ¶

type MarshalFunc func(v interface{}) ([]byte, error)

type MetaManager ¶

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

MetaManager manages the list of metas, but only the leader will it requests to. If the one is not the actual leader, it will retry with another.

func NewMetaManager(addrs []string, creator NodeSessionCreator) *MetaManager

func (m *MetaManager) Close() error

func (m *MetaManager) QueryConfig(ctx context.Context, tableName string) (*replication.QueryCfgResponse, error)

type MockCodec ¶

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

MockCodec is only used for testing. By default it does nothing on marshalling and unmarshalling, thus it returns no error even if the input was ill-formed.

func (p *MockCodec) Marshal(v interface{}) ([]byte, error)

func (p *MockCodec) MockMarshal(marshal MarshalFunc)

func (p *MockCodec) MockUnMarshal(unmarshal UnmarshalFunc)

func (p *MockCodec) String() string

func (p *MockCodec) Unmarshal(data []byte, v interface{}) error

type NodeSession ¶

type NodeSession interface {
	String() string

	// Invoke an rpc call.
	CallWithGpid(ctx context.Context, gpid *base.Gpid, args RpcRequestArgs, name string) (result RpcResponseResult, err error)

	// Get connection state.
	ConnState() rpc.ConnState

	Close() error
}

NodeSession represents the network session to a node (either a meta server or a replica server). It encapsulates the internal rpc process, including network communication and message (de)serialization.

func NewNodeSession(addr string, ntype NodeType) NodeSession

type NodeSessionCreator ¶

type NodeSessionCreator func(string, NodeType) NodeSession

NodeSessionCreator creates an instance of NodeSession, receiving argument `string` as host address, `NodeType` as type of the node.

type NodeType ¶

type NodeType string

const (
	NodeTypeMeta    NodeType = "meta"
	NodeTypeReplica          = "replica"
)

type PegasusCodec ¶

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

func NewPegasusCodec() *PegasusCodec

func (p *PegasusCodec) Marshal(v interface{}) ([]byte, error)

func (p *PegasusCodec) String() string

func (p *PegasusCodec) Unmarshal(data []byte, v interface{}) error

type PegasusRpcCall ¶

type PegasusRpcCall struct {
	Args   RpcRequestArgs
	Result RpcResponseResult
	Name   string // the rpc's name
	SeqId  int32
	Gpid   *base.Gpid
	RawReq []byte // the marshalled request in bytes
	Err    error
}

func MarshallPegasusRpc(codec rpc.Codec, seqId int32, gpid *base.Gpid, args RpcRequestArgs, name string) (*PegasusRpcCall, error)

func ReadRpcResponse(conn *rpc.RpcConn, codec rpc.Codec) (*PegasusRpcCall, error)

type ReplicaManager ¶

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

ReplicaManager manages the pool of sessions to replica servers, so that different tables that locate on the same replica server can share one ReplicaSession, without the effort of creating a new connection.

func NewReplicaManager(creator NodeSessionCreator) *ReplicaManager

func (rm *ReplicaManager) Close() error

func (rm *ReplicaManager) GetReplica(addr string) *ReplicaSession

func (rm *ReplicaManager) ReplicaCount() int

type ReplicaSession ¶

type ReplicaSession struct {
	NodeSession
}

ReplicaSession represents the network session between client and replica server.

func (rs *ReplicaSession) CheckAndSet(ctx context.Context, gpid *base.Gpid, request *rrdb.CheckAndSetRequest) (*rrdb.CheckAndSetResponse, error)

func (rs *ReplicaSession) ClearScanner(ctx context.Context, gpid *base.Gpid, contextId int64) error

func (rs *ReplicaSession) Del(ctx context.Context, gpid *base.Gpid, key *base.Blob) (*rrdb.UpdateResponse, error)

func (rs *ReplicaSession) Get(ctx context.Context, gpid *base.Gpid, key *base.Blob) (*rrdb.ReadResponse, error)

func (rs *ReplicaSession) GetScanner(ctx context.Context, gpid *base.Gpid, request *rrdb.GetScannerRequest) (*rrdb.ScanResponse, error)

func (rs *ReplicaSession) MultiDelete(ctx context.Context, gpid *base.Gpid, request *rrdb.MultiRemoveRequest) (*rrdb.MultiRemoveResponse, error)

func (rs *ReplicaSession) MultiGet(ctx context.Context, gpid *base.Gpid, request *rrdb.MultiGetRequest) (*rrdb.MultiGetResponse, error)

func (rs *ReplicaSession) MultiSet(ctx context.Context, gpid *base.Gpid, request *rrdb.MultiPutRequest) (*rrdb.UpdateResponse, error)

func (rs *ReplicaSession) Put(ctx context.Context, gpid *base.Gpid, key *base.Blob, value *base.Blob, expireTsSeconds int32) (*rrdb.UpdateResponse, error)

func (rs *ReplicaSession) Scan(ctx context.Context, gpid *base.Gpid, request *rrdb.ScanRequest) (*rrdb.ScanResponse, error)

func (rs *ReplicaSession) SortKeyCount(ctx context.Context, gpid *base.Gpid, hashKey *base.Blob) (*rrdb.CountResponse, error)

func (rs *ReplicaSession) TTL(ctx context.Context, gpid *base.Gpid, key *base.Blob) (*rrdb.TTLResponse, error)

type RpcRequestArgs ¶

type RpcRequestArgs interface {
	String() string
	Write(oprot thrift.TProtocol) error
}

a trait of the thrift-generated argument type (MetaQueryCfgArgs, RrdbPutArgs e.g.)

type RpcResponseResult ¶

type RpcResponseResult interface {
	String() string
	Read(iprot thrift.TProtocol) error
}

a trait of the thrift-generated result type (MetaQueryCfgResult e.g.)

type UnmarshalFunc ¶

type UnmarshalFunc func(data []byte, v interface{}) error