vitalif
/
etcd
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add comments; event.go: add Create and Update const

release-0.4
Xiang Li 2013-09-07 08:54:58 -04:00
parent bd8ec6d67b
commit f50cf0497d
5 changed files with 68 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
file_system/event.go
View File

@ -8,7 +8,8 @@ import (
const (
Get = "get"
Set = "set"
Create = "create"
Update = "update"
Delete = "delete"
TestAndSet = "testAndSet"
)

12
file_system/event_test.go
View File

@ -13,7 +13,7 @@ func TestEventQueue(t *testing.T) {
// Add
for i := 0; i < 200; i++ {
e := newEvent(Set, "/foo", uint64(i), 0)
e := newEvent(Create, "/foo", uint64(i), 0)
eh.addEvent(e)
}
@ -33,11 +33,11 @@ func TestScanHistory(t *testing.T) {
eh := newEventHistory(100)
// Add
eh.addEvent(newEvent(Set, "/foo", 1, 0))
eh.addEvent(newEvent(Set, "/foo/bar", 2, 0))
eh.addEvent(newEvent(Set, "/foo/foo", 3, 0))
eh.addEvent(newEvent(Set, "/foo/bar/bar", 4, 0))
eh.addEvent(newEvent(Set, "/foo/foo/foo", 5, 0))
eh.addEvent(newEvent(Create, "/foo", 1, 0))
eh.addEvent(newEvent(Create, "/foo/bar", 2, 0))
eh.addEvent(newEvent(Create, "/foo/foo", 3, 0))
eh.addEvent(newEvent(Create, "/foo/bar/bar", 4, 0))
eh.addEvent(newEvent(Create, "/foo/foo/foo", 5, 0))
e, err := eh.scan("/foo", 1)
if err != nil || e.Index != 1 {

96
file_system/file_system.go
View File

@ -25,15 +25,15 @@ func New() *FileSystem {
}
func (fs *FileSystem) Get(keyPath string, recusive bool, index uint64, term uint64) (*Event, error) {
func (fs *FileSystem) Get(nodePath string, recusive bool, index uint64, term uint64) (*Event, error) {
// TODO: add recursive get
n, err := fs.InternalGet(keyPath, index, term)
n, err := fs.InternalGet(nodePath, index, term)
if err != nil {
return nil, err
}
e := newEvent(Get, keyPath, index, term)
e := newEvent(Get, nodePath, index, term)
if n.IsDir() { // node is dir
e.Dir = true
@ -65,14 +65,17 @@ func (fs *FileSystem) Get(keyPath string, recusive bool, index uint64, term uint
return e, nil
}
func (fs *FileSystem) Create(keyPath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
keyPath = path.Clean("/" + keyPath)
// Create function creates the Node at nodePath. Create will help to create intermediate directories with no ttl.
// If the node has already existed, create will fail.
// If any node on the path is a file, create will fail.
func (fs *FileSystem) Create(nodePath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
nodePath = path.Clean("/" + nodePath)
// make sure we can create the node
_, err := fs.InternalGet(keyPath, index, term)
_, err := fs.InternalGet(nodePath, index, term)
if err == nil { // key already exists
return nil, etcdErr.NewError(105, keyPath)
return nil, etcdErr.NewError(105, nodePath)
}
etcdError, _ := err.(etcdErr.Error)
@ -81,28 +84,28 @@ func (fs *FileSystem) Create(keyPath string, value string, expireTime time.Time,
return nil, err
}
dir, _ := path.Split(keyPath)
dir, _ := path.Split(nodePath)
// walk through the keyPath, create dirs and get the last directory node
// walk through the nodePath, create dirs and get the last directory node
d, err := fs.walk(dir, fs.checkDir)
if err != nil {
return nil, err
}
e := newEvent(Set, keyPath, fs.Index, fs.Term)
e := newEvent(Create, nodePath, fs.Index, fs.Term)
var n *Node
if len(value) != 0 { // create file
e.Value = value
n = newFile(keyPath, value, fs.Index, fs.Term, d, "", expireTime)
n = newFile(nodePath, value, fs.Index, fs.Term, d, "", expireTime)
} else { // create directory
e.Dir = true
n = newDir(keyPath, fs.Index, fs.Term, d, "", expireTime)
n = newDir(nodePath, fs.Index, fs.Term, d, "", expireTime)
}
@ -119,26 +122,26 @@ func (fs *FileSystem) Create(keyPath string, value string, expireTime time.Time,
e.TTL = int64(expireTime.Sub(time.Now()) / time.Second)
}
fs.WatcherHub.notify(e)
return e, nil
}
func (fs *FileSystem) Update(keyPath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
n, err := fs.InternalGet(keyPath, index, term)
// Update function updates the value/ttl of the node.
// If the node is a file, the value and the ttl can be updated.
// If the node is a directory, only the ttl can be updated.
func (fs *FileSystem) Update(nodePath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
n, err := fs.InternalGet(nodePath, index, term)
if err != nil { // if node does not exist, return error
if err != nil { // if the node does not exist, return error
return nil, err
}
e := newEvent(Set, keyPath, fs.Index, fs.Term)
e := newEvent(Update, nodePath, fs.Index, fs.Term)
if n.IsDir() { // if the node is a directory, we can only update ttl
if len(value) != 0 {
return nil, etcdErr.NewError(102, keyPath)
}
if n.ExpireTime != Permanent && expireTime != Permanent {
n.stopExpire <- true
return nil, etcdErr.NewError(102, nodePath)
}
} else { // if the node is a file, we can update value and ttl
@ -149,25 +152,27 @@ func (fs *FileSystem) Update(keyPath string, value string, expireTime time.Time,
}
n.Write(value, index, term)
if n.ExpireTime != Permanent && expireTime != Permanent {
n.stopExpire <- true
}
}
// update ttl
if n.ExpireTime != Permanent && expireTime != Permanent {
n.stopExpire <- true
}
if expireTime != Permanent {
go n.Expire()
e.Expiration = &n.ExpireTime
e.TTL = int64(expireTime.Sub(time.Now()) / time.Second)
}
fs.WatcherHub.notify(e)
return e, nil
}
func (fs *FileSystem) TestAndSet(keyPath string, prevValue string, prevIndex uint64, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
f, err := fs.InternalGet(keyPath, index, term)
func (fs *FileSystem) TestAndSet(nodePath string, prevValue string, prevIndex uint64,
value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
f, err := fs.InternalGet(nodePath, index, term)
if err != nil {
@ -175,15 +180,18 @@ func (fs *FileSystem) TestAndSet(keyPath string, prevValue string, prevIndex uin
}
if f.IsDir() { // can only test and set file
return nil, etcdErr.NewError(102, keyPath)
return nil, etcdErr.NewError(102, nodePath)
}
if f.Value == prevValue || f.ModifiedIndex == prevIndex {
// if test succeed, write the value
e := newEvent(TestAndSet, keyPath, index, term)
e := newEvent(TestAndSet, nodePath, index, term)
e.PrevValue = f.Value
e.Value = value
f.Write(value, index, term)
fs.WatcherHub.notify(e)
return e, nil
}
@ -191,14 +199,16 @@ func (fs *FileSystem) TestAndSet(keyPath string, prevValue string, prevIndex uin
return nil, etcdErr.NewError(101, cause)
}
func (fs *FileSystem) Delete(keyPath string, recurisive bool, index uint64, term uint64) (*Event, error) {
n, err := fs.InternalGet(keyPath, index, term)
// Delete function deletes the node at the given path.
// If the node is a directory, recursive must be true to delete it.
func (fs *FileSystem) Delete(nodePath string, recursive bool, index uint64, term uint64) (*Event, error) {
n, err := fs.InternalGet(nodePath, index, term)
if err != nil {
if err != nil { // if the node does not exist, return error
return nil, err
}
e := newEvent(Delete, keyPath, index, term)
e := newEvent(Delete, nodePath, index, term)
if n.IsDir() {
e.Dir = true
@ -206,11 +216,11 @@ func (fs *FileSystem) Delete(keyPath string, recurisive bool, index uint64, term
e.PrevValue = n.Value
}
callback := func(path string) {
callback := func(path string) { // notify function
fs.WatcherHub.notifyWithPath(e, path, true)
}
err = n.Remove(recurisive, callback)
err = n.Remove(recursive, callback)
if err != nil {
return nil, err
@ -221,9 +231,9 @@ func (fs *FileSystem) Delete(keyPath string, recurisive bool, index uint64, term
return e, nil
}
// walk function walks all the keyPath and apply the walkFunc on each directory
func (fs *FileSystem) walk(keyPath string, walkFunc func(prev *Node, component string) (*Node, error)) (*Node, error) {
components := strings.Split(keyPath, "/")
// walk function walks all the nodePath and apply the walkFunc on each directory
func (fs *FileSystem) walk(nodePath string, walkFunc func(prev *Node, component string) (*Node, error)) (*Node, error) {
components := strings.Split(nodePath, "/")
curr := fs.Root
@ -243,9 +253,9 @@ func (fs *FileSystem) walk(keyPath string, walkFunc func(prev *Node, component s
return curr, nil
}
// InternalGet function get the node of the given keyPath.
func (fs *FileSystem) InternalGet(keyPath string, index uint64, term uint64) (*Node, error) {
keyPath = path.Clean("/" + keyPath)
// InternalGet function get the node of the given nodePath.
func (fs *FileSystem) InternalGet(nodePath string, index uint64, term uint64) (*Node, error) {
nodePath = path.Clean("/" + nodePath)
// update file system known index and term
fs.Index, fs.Term = index, term
@ -264,7 +274,7 @@ func (fs *FileSystem) InternalGet(keyPath string, index uint64, term uint64) (*N
return nil, etcdErr.NewError(100, path.Join(parent.Path, name))
}
f, err := fs.walk(keyPath, walkFunc)
f, err := fs.walk(nodePath, walkFunc)
if err != nil {
return nil, err

8
file_system/node.go
View File

@ -34,9 +34,9 @@ type Node struct {
stopExpire chan bool // stop expire routine channel
}
func newFile(keyPath string, value string, createIndex uint64, createTerm uint64, parent *Node, ACL string, expireTime time.Time) *Node {
func newFile(nodePath string, value string, createIndex uint64, createTerm uint64, parent *Node, ACL string, expireTime time.Time) *Node {
return &Node{
Path: keyPath,
Path: nodePath,
CreateIndex: createIndex,
CreateTerm: createTerm,
ModifiedIndex: createIndex,
@ -49,9 +49,9 @@ func newFile(keyPath string, value string, createIndex uint64, createTerm uint64
}
}
func newDir(keyPath string, createIndex uint64, createTerm uint64, parent *Node, ACL string, expireTime time.Time) *Node {
func newDir(nodePath string, createIndex uint64, createTerm uint64, parent *Node, ACL string, expireTime time.Time) *Node {
return &Node{
Path: keyPath,
Path: nodePath,
CreateIndex: createIndex,
CreateTerm: createTerm,
Parent: parent,

6
file_system/watcher_test.go
View File

@ -19,7 +19,7 @@ func TestWatch(t *testing.T) {
// do nothing
}
e := newEvent(Set, "/foo/bar", 1, 0)
e := newEvent(Create, "/foo/bar", 1, 0)
wh.notify(e)
@ -31,7 +31,7 @@ func TestWatch(t *testing.T) {
c, _ = wh.watch("/foo", false, 0)
e = newEvent(Set, "/foo/bar", 1, 0)
e = newEvent(Create, "/foo/bar", 1, 0)
wh.notify(e)
@ -42,7 +42,7 @@ func TestWatch(t *testing.T) {
// do nothing
}
e = newEvent(Set, "/foo", 1, 0)
e = newEvent(Create, "/foo", 1, 0)
wh.notify(e)