mirror of
https://github.com/go-gitea/gitea
synced 2024-07-07 10:35:48 +00:00
104 lines
2.3 KiB
Go
104 lines
2.3 KiB
Go
|
package etcd
|
||
|
|
||
|
import (
|
||
|
"errors"
|
||
|
)
|
||
|
|
||
|
// Errors introduced by the Watch command.
|
||
|
var (
|
||
|
ErrWatchStoppedByUser = errors.New("Watch stopped by the user via stop channel")
|
||
|
)
|
||
|
|
||
|
// If recursive is set to true the watch returns the first change under the given
|
||
|
// prefix since the given index.
|
||
|
//
|
||
|
// If recursive is set to false the watch returns the first change to the given key
|
||
|
// since the given index.
|
||
|
//
|
||
|
// To watch for the latest change, set waitIndex = 0.
|
||
|
//
|
||
|
// If a receiver channel is given, it will be a long-term watch. Watch will block at the
|
||
|
//channel. After someone receives the channel, it will go on to watch that
|
||
|
// prefix. If a stop channel is given, the client can close long-term watch using
|
||
|
// the stop channel.
|
||
|
func (c *Client) Watch(prefix string, waitIndex uint64, recursive bool,
|
||
|
receiver chan *Response, stop chan bool) (*Response, error) {
|
||
|
logger.Debugf("watch %s [%s]", prefix, c.cluster.Leader)
|
||
|
if receiver == nil {
|
||
|
raw, err := c.watchOnce(prefix, waitIndex, recursive, stop)
|
||
|
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
return raw.Unmarshal()
|
||
|
}
|
||
|
defer close(receiver)
|
||
|
|
||
|
for {
|
||
|
raw, err := c.watchOnce(prefix, waitIndex, recursive, stop)
|
||
|
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
resp, err := raw.Unmarshal()
|
||
|
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
waitIndex = resp.Node.ModifiedIndex + 1
|
||
|
receiver <- resp
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (c *Client) RawWatch(prefix string, waitIndex uint64, recursive bool,
|
||
|
receiver chan *RawResponse, stop chan bool) (*RawResponse, error) {
|
||
|
|
||
|
logger.Debugf("rawWatch %s [%s]", prefix, c.cluster.Leader)
|
||
|
if receiver == nil {
|
||
|
return c.watchOnce(prefix, waitIndex, recursive, stop)
|
||
|
}
|
||
|
|
||
|
for {
|
||
|
raw, err := c.watchOnce(prefix, waitIndex, recursive, stop)
|
||
|
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
resp, err := raw.Unmarshal()
|
||
|
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
waitIndex = resp.Node.ModifiedIndex + 1
|
||
|
receiver <- raw
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// helper func
|
||
|
// return when there is change under the given prefix
|
||
|
func (c *Client) watchOnce(key string, waitIndex uint64, recursive bool, stop chan bool) (*RawResponse, error) {
|
||
|
|
||
|
options := Options{
|
||
|
"wait": true,
|
||
|
}
|
||
|
if waitIndex > 0 {
|
||
|
options["waitIndex"] = waitIndex
|
||
|
}
|
||
|
if recursive {
|
||
|
options["recursive"] = true
|
||
|
}
|
||
|
|
||
|
resp, err := c.getCancelable(key, options, stop)
|
||
|
|
||
|
if err == ErrRequestCancelled {
|
||
|
return nil, ErrWatchStoppedByUser
|
||
|
}
|
||
|
|
||
|
return resp, err
|
||
|
}
|