zookeeper

Documentation

Index

• Constants
• func HandleClientRestart(r ZkClientFacade)
• func StateToString(state zk.State) string
• func ValidateZookeeperClient(container ZkClientFacade, opts ...Option) error
• type Option
  • func WithTestCluster(ts *zk.TestCluster) Option
  • func WithZkName(name string) Option
• type Options
• type ZkClientFacade
• type ZkEventListener
  • func NewZkEventListener(client *ZookeeperClient) *ZkEventListener
  • func (l *ZkEventListener) Close()
  • func (l *ZkEventListener) ListenServiceEvent(conf *common.URL, zkPath string, listener remoting.DataListener)
  • func (l *ZkEventListener) ListenServiceNodeEvent(zkPath string, listener remoting.DataListener)
  • func (l *ZkEventListener) SetClient(client *ZookeeperClient)
• type ZookeeperClient
  • func NewMockZookeeperClient(name string, timeout time.Duration, opts ...Option) (*zk.TestCluster, *ZookeeperClient, <-chan zk.Event, error)
  • func NewZookeeperClient(name string, zkAddrs []string, timeout time.Duration) (*ZookeeperClient, error)
  • func (z *ZookeeperClient) Close()
  • func (z *ZookeeperClient) Create(basePath string) error
  • func (z *ZookeeperClient) CreateTempWithValue(basePath string, value []byte) error
  • func (z *ZookeeperClient) CreateWithValue(basePath string, value []byte) error
  • func (z *ZookeeperClient) Delete(basePath string) error
  • func (z *ZookeeperClient) Done() <-chan struct{}
  • func (z *ZookeeperClient) ExistW(zkPath string) (<-chan zk.Event, error)
  • func (z *ZookeeperClient) GetChildren(path string) ([]string, error)
  • func (z *ZookeeperClient) GetChildrenW(path string) ([]string, <-chan zk.Event, error)
  • func (z *ZookeeperClient) GetContent(zkPath string) ([]byte, *zk.Stat, error)
  • func (z *ZookeeperClient) HandleZkEvent(session <-chan zk.Event)
  • func (z *ZookeeperClient) RegisterEvent(zkPath string, event *chan struct{})
  • func (z *ZookeeperClient) RegisterTemp(basePath string, node string) (string, error)
  • func (z *ZookeeperClient) RegisterTempSeq(basePath string, data []byte) (string, error)
  • func (z *ZookeeperClient) SetContent(zkPath string, content []byte, version int32) (*zk.Stat, error)
  • func (z *ZookeeperClient) UnregisterEvent(zkPath string, event *chan struct{})
  • func (z *ZookeeperClient) ZkConnValid() bool

Constants

const (
	// ConnDelay connection delay interval
	ConnDelay = 3
	// MaxFailTimes max fail times
	MaxFailTimes = 15
)

Functions

func HandleClientRestart

func HandleClientRestart(r ZkClientFacade)

HandleClientRestart keeps the connection between client and server

func StateToString

func StateToString(state zk.State) string

nolint

func ValidateZookeeperClient

func ValidateZookeeperClient(container ZkClientFacade, opts ...Option) error

ValidateZookeeperClient validates client and sets options

Types

type Option

type Option func(*Options)

Option will define a function of handling Options

func WithTestCluster

func WithTestCluster(ts *zk.TestCluster) Option

WithTestCluster sets test cluster for zk client

func WithZkName

func WithZkName(name string) Option

WithZkName sets zk client name

type Options

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

nolint

type ZkClientFacade

type ZkClientFacade interface {
	ZkClient() *ZookeeperClient
	SetZkClient(*ZookeeperClient)
	ZkClientLock() *sync.Mutex
	WaitGroup() *sync.WaitGroup // for wait group control, zk client listener & zk client container
	Done() chan struct{}        // for zk client control
	RestartCallBack() bool
	GetUrl() common.URL
}

type ZkEventListener

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

nolint

func NewZkEventListener

func NewZkEventListener(client *ZookeeperClient) *ZkEventListener

NewZkEventListener returns a EventListener instance

func (*ZkEventListener) Close

func (l *ZkEventListener) Close()

Close will let client listen exit

func (*ZkEventListener) ListenServiceEvent

func (l *ZkEventListener) ListenServiceEvent(conf *common.URL, zkPath string, listener remoting.DataListener)

ListenServiceEvent is invoked by ZkConsumerRegistry::Register/ZkConsumerRegistry::get/ZkConsumerRegistry::getListener registry.go:Listen -> listenServiceEvent -> listenDirEvent -> listenServiceNodeEvent

|
--------> listenServiceNodeEvent

func (*ZkEventListener) ListenServiceNodeEvent

func (l *ZkEventListener) ListenServiceNodeEvent(zkPath string, listener remoting.DataListener)

ListenServiceNodeEvent listen a path node event

func (*ZkEventListener) SetClient

func (l *ZkEventListener) SetClient(client *ZookeeperClient)

nolint

type ZookeeperClient

type ZookeeperClient struct {
	ZkAddrs      []string
	sync.RWMutex // for conn
	Conn         *zk.Conn
	Timeout      time.Duration

	Wait sync.WaitGroup
	// contains filtered or unexported fields
}

ZookeeperClient represents zookeeper client Configuration

func NewMockZookeeperClient

func NewMockZookeeperClient(name string, timeout time.Duration, opts ...Option) (*zk.TestCluster, *ZookeeperClient, <-chan zk.Event, error)

NewMockZookeeperClient returns a mock client instance

func NewZookeeperClient

func NewZookeeperClient(name string, zkAddrs []string, timeout time.Duration) (*ZookeeperClient, error)

nolint

func (*ZookeeperClient) Close

func (z *ZookeeperClient) Close()

nolint

func (*ZookeeperClient) Create

func (z *ZookeeperClient) Create(basePath string) error

Create will create the node recursively, which means that if the parent node is absent, it will create parent node first. And the value for the basePath is ""

func (*ZookeeperClient) CreateTempWithValue

func (z *ZookeeperClient) CreateTempWithValue(basePath string, value []byte) error

CreateTempWithValue will create the node recursively, which means that if the parent node is absent, it will create parent node first，and set value in last child path If the path exist, it will update data

func (*ZookeeperClient) CreateWithValue

func (z *ZookeeperClient) CreateWithValue(basePath string, value []byte) error

CreateWithValue will create the node recursively, which means that if the parent node is absent, it will create parent node first.

func (*ZookeeperClient) Delete

func (z *ZookeeperClient) Delete(basePath string) error

nolint

func (*ZookeeperClient) Done

func (z *ZookeeperClient) Done() <-chan struct{}

nolint

func (*ZookeeperClient) ExistW

func (z *ZookeeperClient) ExistW(zkPath string) (<-chan zk.Event, error)

ExistW to judge watch whether it exists or not by @zkPath

func (*ZookeeperClient) GetChildren

func (z *ZookeeperClient) GetChildren(path string) ([]string, error)

GetChildren gets children by @path

func (*ZookeeperClient) GetChildrenW

func (z *ZookeeperClient) GetChildrenW(path string) ([]string, <-chan zk.Event, error)

GetChildrenW gets children watch by @path

func (*ZookeeperClient) GetContent

func (z *ZookeeperClient) GetContent(zkPath string) ([]byte, *zk.Stat, error)

GetContent gets content by @zkPath

func (*ZookeeperClient) HandleZkEvent

func (z *ZookeeperClient) HandleZkEvent(session <-chan zk.Event)

HandleZkEvent handles zookeeper events

func (*ZookeeperClient) RegisterEvent

func (z *ZookeeperClient) RegisterEvent(zkPath string, event *chan struct{})

RegisterEvent registers zookeeper events

func (*ZookeeperClient) RegisterTemp

func (z *ZookeeperClient) RegisterTemp(basePath string, node string) (string, error)

RegisterTemp registers temporary node by @basePath and @node

func (*ZookeeperClient) RegisterTempSeq

func (z *ZookeeperClient) RegisterTempSeq(basePath string, data []byte) (string, error)

RegisterTempSeq register temporary sequence node by @basePath and @data

func (*ZookeeperClient) SetContent

func (z *ZookeeperClient) SetContent(zkPath string, content []byte, version int32) (*zk.Stat, error)

nolint

func (*ZookeeperClient) UnregisterEvent

func (z *ZookeeperClient) UnregisterEvent(zkPath string, event *chan struct{})

UnregisterEvent unregisters zookeeper events

func (*ZookeeperClient) ZkConnValid

func (z *ZookeeperClient) ZkConnValid() bool

ZkConnValid validates zookeeper connection