Method Run

lang/lang.go:367–466

Run kicks off the function engine. Use the context to shut it down.

(ctx context.Context)

// Run kicks off the function engine. Use the context to shut it down.
func (obj *Lang) Run(ctx context.Context) (reterr error) {
	defer obj.wg.Wait()

	ctx, cancel := context.WithCancel(ctx) // wrap parent
	defer cancel()

	//<-obj.funcs.Started() // wait for startup (will not block forever)

	// Sanity checks for graph size.
	//if count := obj.funcs.NumVertices(); count != 0 {
	//	return fmt.Errorf("expected empty graph on start, got %d vertices", count)
	//}
	//defer func() {
	//	if count := obj.funcs.NumVertices(); count != 0 {
	//		err := fmt.Errorf("expected empty graph on exit, got %d vertices", count)
	//		reterr = errwrap.Append(reterr, err)
	//	}
	//}()

	txn := obj.funcs.Txn()
	defer txn.Free() // remember to call Free()
	txn.AddGraph(obj.graph)
	if err := txn.Commit(); err != nil {
		return errwrap.Wrapf(err, "error adding to function graph engine")
	}
	defer func() {
		if err := txn.Reverse(); err != nil { // should remove everything we added
			reterr = errwrap.Append(reterr, err)
		}
	}()

	//obj.Logf("function engine starting took: %s", time.Since(timing))
	// wait for some activity
	obj.Logf("stream...")

	tableChan := obj.funcs.Stream() // after obj.funcs.Setup runs

	obj.wg.Add(1)
	go func() {
		defer obj.wg.Done()
		defer close(obj.streamChan)
		defer cancel() // if this loop errors, it should cancel and err

		var table interfaces.Table
		var ok bool
		for {
			select {
			case table, ok = <-tableChan:
				if !ok {
					return
				}

			case <-ctx.Done():
				obj.errAppend(ctx.Err())
				return
			}

			// this call returns the graph