Add Node interface

coding/node.go

@@ -0,0 +1,248 @@
+package coding
+
+import (
+	ipld "github.com/ipfs/go-ipld"
+	stream "github.com/ipfs/go-ipld/coding/stream"
+)
+
+type UnexpectedTokenError stream.ReaderToken
+
+func (e *UnexpectedTokenError) Error() string {
+	return "Unexpected token " + stream.TokenName(stream.ReaderToken(*e))
+}
+
+type aborter interface {
+	Abort() error
+}
+
+type args struct {
+	path      []interface{}
+	tokenType stream.ReaderToken
+	value     interface{}
+	result    chan error
+}
+
+type valueIterator struct {
+	c   chan *args
+	e   *error
+	val *args
+}
+
+func (v *valueIterator) Value() (interface{}, error) {
+
+	if v.c == nil {
+		return nil, nil
+	}
+
+	if v.val == nil {
+		v.val = <-v.c
+		if v.val == nil {
+			return nil, *v.e
+		}
+		close(v.val.result)
+	}
+
+	res := v.val.value
+	v.val.result = nil
+
+	switch v.val.tokenType {
+	case stream.TokenValue:
+		v.val = nil
+		v.c = nil
+		return res, nil
+	case stream.TokenValuePart:
+		v.val = nil
+		return res, nil
+	default:
+		err := UnexpectedTokenError(v.val.tokenType)
+		return nil, &err
+	}
+}
+
+func (v *valueIterator) Abort() error {
+	val, err := v.Value()
+	for val != nil {
+		val, err = v.Value()
+	}
+	return err
+}
+
+type iterator struct {
+	c     chan *args
+	e     *error
+	key   *args
+	val   *args
+	child aborter
+}
+
+func (i *iterator) Next() (bool, error) {
+	if i.child != nil {
+		err := i.child.Abort()
+		i.child = nil
+		if err != nil {
+			return false, err
+		}
+	}
+
+	if i.key != nil && i.key.result != nil {
+		i.key.result <- stream.NodeReadSkip
+		close(i.key.result)
+	}
+
+	if i.c == nil {
+		return false, nil
+	}
+
+	i.key = <-i.c
+	if i.key == nil {
+		return false, *i.e
+	}
+
+	i.val = nil
+	switch i.key.tokenType {
+	case stream.TokenEndNode, stream.TokenEndArray:
+		close(i.key.result)
+		i.c = nil
+		i.key = nil
+		i.val = nil
+		return false, nil
+	case stream.TokenKey, stream.TokenIndex:
+		break
+	default:
+		close(i.key.result)
+		err := UnexpectedTokenError(i.key.tokenType)
+		return false, &err
+	}
+
+	return true, nil
+}
+
+func (i *iterator) Iterate() (ipld.NodeIterator, error) {
+	if i.child != nil || i.key == nil {
+		// Already iterating, don't iterate twice
+		// Or key not fetched yet
+		// Or iteration end
+		return nil, nil
+	} else if i.key.result != nil {
+		// Close key return channel
+		close(i.key.result)
+		i.key.result = nil
+	}
+
+	// Fetch the value of the key
+	if i.val == nil {
+		i.val = <-i.c
+		if i.val == nil {
+			return nil, *i.e
+		}
+		close(i.val.result)
+		i.val.result = nil
+	}
+
+	switch i.val.tokenType {
+	case stream.TokenValuePart, stream.TokenValue:
+		return nil, nil
+	case stream.TokenArray, stream.TokenNode:
+		break
+	default:
+		err := UnexpectedTokenError(i.val.tokenType)
+		return nil, &err
+	}
+
+	// create the child iterator
+	res := new(iterator)
+	res.c = i.c
+	res.e = i.e
+	i.child = res
+	return res, nil
+}
+
+func (i *iterator) Abort() error {
+	cont, err := i.Next()
+	for cont {
+		cont, err = i.Next()
+	}
+	return err
+}
+
+func (i *iterator) Key() interface{} {
+	if i.key == nil {
+		return nil
+	} else if ii, ok := i.key.value.(int); ok && ii >= 0 {
+		return uint64(ii)
+	} else {
+		return i.key.value
+	}
+}
+
+func (i *iterator) Value() (interface{}, error) {
+	if i.child != nil || i.key == nil {
+		// Iterating over a child structure
+		// Key not fetched yet or end of iteration
+		return nil, nil
+	} else if i.key.result != nil {
+		// Close key return channel
+		close(i.key.result)
+		i.key.result = nil
+	}
+
+	// Fetch the value of the key
+	if i.val == nil {
+		i.val = <-i.c
+		if i.val == nil {
+			return nil, *i.e
+		}
+		close(i.val.result)
+		i.val.result = nil
+	}
+
+	switch i.val.tokenType {
+	case stream.TokenValuePart:
+		res := &valueIterator{i.c, i.e, i.val}
+		i.child = res
+		return res, nil
+	case stream.TokenValue:
+		return i.val.value, nil
+	case stream.TokenArray, stream.TokenNode:
+		return nil, nil
+	default:
+		err := UnexpectedTokenError(i.val.tokenType)
+		return nil, &err
+	}
+}
+
+func (i *iterator) read(path []interface{}, tokenType stream.ReaderToken, value interface{}) error {
+	res := make(chan error)
+	i.c <- &args{path, tokenType, value, res}
+	return <-res
+}
+
+func NewNodeFromReader(r stream.NodeReader) (interface{}, error) {
+	var i *iterator = new(iterator)
+	i.c = make(chan *args)
+	i.e = new(error)
+	go func() {
+		*i.e = r.Read(i.read)
+		close(i.c)
+	}()
+
+	a := <-i.c
+	if a == nil {
+		return nil, *i.e
+	}
+
+	close(a.result)
+	switch a.tokenType {
+	case stream.TokenNode:
+		return i, nil
+	case stream.TokenArray:
+		return i, nil
+	case stream.TokenValue:
+		return a.value, nil
+	case stream.TokenValuePart:
+		return &valueIterator{i.c, i.e, a}, nil
+	default:
+		err := UnexpectedTokenError(a.tokenType)
+		return nil, &err
+	}
+}