lang: interfaces, funcs: Port Func API to new Stream signature

This removes the `Close() error` and replaces it with a more modern
Stream api that takes a context. This removes boilerplate and makes
integration with concurrent code easier. The only downside is that there
isn't an explicit cleanup step, but only one function was even using
that and it was possible to switch it to a defer in Stream.

lang/funcs/dage/dage.go

@@ -289,7 +289,6 @@ func (obj *Engine) deleteVertex(f interfaces.Func) error {
 	// lookup when we're in an unlocked state.
 	delete(obj.state, f)
 	obj.graph.DeleteVertex(f)
-	//return f.Close() // XXX: change api to close with ctx instead.
 	return nil
 }
 
@@ -364,7 +363,6 @@ Loop:
 
 	rollback := func() {
 		for f := range status {
-			//f.Close() // shutdown if necessary
 			obj.graph.DeleteVertex(f)
 			delete(obj.state, f)
 		}
@@ -671,8 +669,7 @@ func (obj *Engine) Run(ctx context.Context) error {
 				if obj.Debug {
 					obj.SafeLogf("Running func `%s`", node)
 				}
-				//runErr := f.Stream(node.ctx) // XXX: new API!
-				runErr := f.Stream()
+				runErr := f.Stream(node.ctx)
 				if obj.Debug {
 					obj.SafeLogf("Exiting func `%s`", node)
 				}
@@ -688,15 +685,6 @@ func (obj *Engine) Run(ctx context.Context) error {
 				// if node never loaded, then we error in the node.output loop!
 			}(f, node)
 
-			node.wg.Add(1)
-			go func(f interfaces.Func, node *state) {
-				defer node.wg.Done()
-				select {
-				case <-ctx.Done():
-				}
-				_ = f.Close() // XXX: change api to close with ctx instead.
-			}(f, node)
-
 			// process events
 			obj.wgAg.Add(1)
 			node.wg.Add(1)

lang/funcs/dage/dage_test.go

@@ -35,9 +35,8 @@ type testFunc struct {
 	Func func(types.Value) (types.Value, error)
 	Meta *meta
 
-	value     types.Value
-	init      *interfaces.Init
-	closeChan chan struct{}
+	value types.Value
+	init  *interfaces.Init
 }
 
 func (obj *testFunc) String() string { return obj.Name }
@@ -60,12 +59,10 @@ func (obj *testFunc) Validate() error {
 
 func (obj *testFunc) Init(init *interfaces.Init) error {
 	obj.init = init
-	obj.closeChan = make(chan struct{})
 	return nil
 }
 
-// XXX: replace with ctx closer
-func (obj *testFunc) Stream() error {
+func (obj *testFunc) Stream(ctx context.Context) error {
 	defer close(obj.init.Output) // the sender closes
 	defer obj.init.Logf("stream closed")
 	obj.init.Logf("stream startup")
@@ -104,7 +101,7 @@ func (obj *testFunc) Stream() error {
 				}
 			}
 
-		case <-obj.closeChan:
+		case <-ctx.Done():
 			return nil
 		}
 
@@ -125,18 +122,12 @@ func (obj *testFunc) Stream() error {
 				}
 			}()
 
-		case <-obj.closeChan:
+		case <-ctx.Done():
 			return nil
 		}
 	}
 }
 
-// XXX: replace with ctx closer
-func (obj *testFunc) Close() error {
-	close(obj.closeChan)
-	return nil
-}
-
 type meta struct {
 	EventCount int
 	Event      chan struct{}

lang/funcs/dage/txn_test.go

@@ -20,6 +20,7 @@
 package dage
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"testing"
@@ -105,12 +106,11 @@ type testNullFunc struct {
 	name string
 }
 
-func (obj *testNullFunc) String() string              { return obj.name }
-func (obj *testNullFunc) Info() *interfaces.Info      { return nil }
-func (obj *testNullFunc) Validate() error             { return nil }
-func (obj *testNullFunc) Init(*interfaces.Init) error { return nil }
-func (obj *testNullFunc) Stream() error               { return nil }
-func (obj *testNullFunc) Close() error                { return nil }
+func (obj *testNullFunc) String() string               { return obj.name }
+func (obj *testNullFunc) Info() *interfaces.Info       { return nil }
+func (obj *testNullFunc) Validate() error              { return nil }
+func (obj *testNullFunc) Init(*interfaces.Init) error  { return nil }
+func (obj *testNullFunc) Stream(context.Context) error { return nil }
 
 func TestTxn1(t *testing.T) {
 	graph, err := pgraph.NewGraph("test")