Merge pull request #578 from nevalang/stream_rewrite

Stream rewrite

.vscode/launch.json

@@ -17,7 +17,7 @@
       "mode": "auto",
       "program": "${workspaceFolder}/cmd/neva",
       "cwd": "${workspaceFolder}",
-      "args": ["run", "examples/struct_selector"]
+      "args": ["run", "e2e/add_nums_verbose/main"]
     },
     {
       "name": "LSP",

e2e/const_refs_verbose/e2e_test.go

@@ -7,14 +7,18 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// Nested map has only one key because keys are unordered
+// so having order in test will make it flacky.
 func Test(t *testing.T) {
 	cmd := exec.Command("neva", "run", "main")
 
 	out, err := cmd.CombinedOutput()
 	require.NoError(t, err)
+
 	require.Equal(
 		t,
-		"map[l:[1 2 3] m:map[one:1 three:3 two:2]]\n",
+		`{"l": [1, 2, 3], "m": {"key": 1}}
+`,
 		string(out),
 	)
 

e2e/const_refs_verbose/main/main.neva

@@ -8,11 +8,7 @@ const {
     two int = 2
     three int = 3
     numsList list<int> = [one, two, three]
-    numsMap map<int> = {
-        one: one,
-        two: two,
-        three: three
-    }
+    numsMap map<int> = { key: one }
     numsStruct NumsStruct = {
         l: numsList,
         m: numsMap

examples/const_refs/e2e_test.go

@@ -8,6 +8,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// Nested map has only one key because keys are unordered
+// so having order in test will make it flacky.
 func Test(t *testing.T) {
 	err := os.Chdir("..")
 	require.NoError(t, err)
@@ -22,7 +24,8 @@ func Test(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(
 		t,
-		"map[l:[1 2 3] m:map[one:1 three:3 two:2]]\n",
+		`{"l": [1, 2, 3], "m": {"key": 1}}
+`,
 		string(out),
 	)
 

examples/const_refs/main.neva

@@ -8,11 +8,7 @@ const {
     two int = 2
     three int = 3
     numsList list<int> = [one, two, three]
-    numsMap map<int> = {
-        one: one,
-        two: two,
-        three: three
-    }
+    numsMap map<int> = { key: one }
     numsStruct NumsStruct = {
         l: numsList,
         m: numsMap

examples/fizzbuzz/main.neva

@@ -1,20 +1,26 @@
 component Main(start) (stop) {
-	nodes { Mod, Range, Match<int>, Unwrap<int>, Println, Lock<int> }
+	nodes { Range, PrintLine, Match }
 	net {
 		:start -> [
             (1 -> range:from),
             (101 -> range:to)
         ]
-        range -> unwrap
+        range.data -> printLine -> match:data
+        100 -> match:case[0] -> :stop
+    }
+}
+
+component PrintLine(data int) (data int) {
+    nodes { Mod, Println, Lock<int> }
+    net {
+        :data -> [mod:data, lock:data]
 
-        unwrap:some   -> [mod:data, lock:data]
         15 -> mod:case[0] -> ('FizzBuzz' -> println)
         3 -> mod:case[1] -> ('Fizz'     -> println)
         5 -> mod:case[2] -> ('Buzz'     -> println)
         mod:else -> println
 
         println -> lock:sig
-        lock -> match:data
-        100 -> match:case[0] -> :stop
+        lock:data -> :data
     }
-}
+}

examples/iterate_over_list/e2e_test.go

@@ -22,7 +22,11 @@ func Test(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(
 		t,
-		"50\n30\n20\n100\n<nil>\n",
+		`{"data": 50, "idx": 0, "last": false}
+{"data": 30, "idx": 1, "last": false}
+{"data": 20, "idx": 2, "last": false}
+{"data": 100, "idx": 3, "last": true}
+`,
 		string(out),
 	)
 

examples/iterate_over_list/main.neva

@@ -1,9 +1,10 @@
 const lst list<int> = [50, 30, 20, 100]
 
 component Main(start) (stop) {
-    nodes { Println<maybe<int>>, Iter<int>, Unwrap<int> }
+    nodes { Println<stream<int>>, Iter<int>, Match<bool> }
     net {
-        :start -> ($lst -> iter -> println -> unwrap)
-        unwrap:none -> :stop
+        :start -> ($lst -> iter -> println) 
+        println.last -> match:data
+        true -> match:case[0] -> :stop
     }
 }

internal/runtime/funcs/int_add.go

@@ -8,7 +8,10 @@ import (
 
 type intAdd struct{}
 
-func (intAdd) Create(io runtime.FuncIO, _ runtime.Msg) (func(ctx context.Context), error) {
+func (intAdd) Create(
+	io runtime.FuncIO,
+	_ runtime.Msg,
+) (func(ctx context.Context), error) {
 	seqIn, err := io.In.Port("seq")
 	if err != nil {
 		return nil, err
@@ -20,29 +23,29 @@ func (intAdd) Create(io runtime.FuncIO, _ runtime.Msg) (func(ctx context.Context
 	}
 
 	return func(ctx context.Context) {
-		var (
-			acc int64
-			cur runtime.Msg
-		)
+		var acc int64 = 0
 
 		for {
+			var item map[string]runtime.Msg
 			select {
 			case <-ctx.Done():
 				return
-			case cur = <-seqIn:
+			case msg := <-seqIn:
+				item = msg.Map()
 			}
 
-			if cur == nil {
+			acc += item["data"].Int()
+
+			if item["last"].Bool() {
 				select {
 				case <-ctx.Done():
 					return
 				case resOut <- runtime.NewIntMsg(acc):
-					acc = 0
+					acc = 0 // reset
 					continue
 				}
 			}
 
-			acc += cur.Int()
 		}
 	}, nil
 }

internal/runtime/funcs/int_mul.go

@@ -20,29 +20,28 @@ func (intMul) Create(io runtime.FuncIO, _ runtime.Msg) (func(ctx context.Context
 	}
 
 	return func(ctx context.Context) {
-		var (
-			acc int64 = 1
-			cur runtime.Msg
-		)
+		var acc int64 = 1
 
 		for {
+			var item map[string]runtime.Msg
 			select {
 			case <-ctx.Done():
 				return
-			case cur = <-seqIn:
+			case msg := <-seqIn:
+				item = msg.Map()
 			}
 
-			if cur == nil {
+			acc *= item["data"].Int()
+
+			if item["last"].Bool() {
 				select {
 				case <-ctx.Done():
 					return
 				case resOut <- runtime.NewIntMsg(acc):
-					acc = 1
+					acc = 1 // reset
 					continue
 				}
 			}
-
-			acc *= cur.Int()
 		}
 	}, nil
 }

internal/runtime/funcs/int_sub.go

@@ -21,35 +21,36 @@ func (intSub) Create(io runtime.FuncIO, _ runtime.Msg) (func(ctx context.Context
 
 	return func(ctx context.Context) {
 		var (
-			acc        int64 = 0
-			inProgress bool  = false
-			cur        runtime.Msg
+			acc     int64 = 0
+			started bool  = false
 		)
 
 		for {
+			var item map[string]runtime.Msg
 			select {
 			case <-ctx.Done():
 				return
-			case cur = <-seqIn:
+			case msg := <-seqIn:
+				item = msg.Map()
 			}
 
-			if cur == nil {
+			if !started {
+				acc = item["data"].Int()
+				started = true
+			} else {
+				acc -= item["data"].Int()
+			}
+
+			if item["last"].Bool() {
 				select {
 				case <-ctx.Done():
 					return
 				case resOut <- runtime.NewIntMsg(acc):
 					acc = 0
-					inProgress = false
+					started = false
 					continue
 				}
 			}
-
-			if !inProgress {
-				acc = cur.Int()
-				inProgress = true
-			} else {
-				acc -= cur.Int()
-			}
 		}
 	}, nil
 }

internal/runtime/funcs/list_iter.go

@@ -8,7 +8,10 @@ import (
 
 type list_iter struct{}
 
-func (c list_iter) Create(io runtime.FuncIO, _ runtime.Msg) (func(ctx context.Context), error) {
+func (c list_iter) Create(
+	io runtime.FuncIO,
+	_ runtime.Msg,
+) (func(ctx context.Context), error) {
 	dataIn, err := io.In.Port("data")
 	if err != nil {
 		return nil, err
@@ -21,25 +24,27 @@ func (c list_iter) Create(io runtime.FuncIO, _ runtime.Msg) (func(ctx context.Co
 
 	return func(ctx context.Context) {
 		for {
+			var list []runtime.Msg
+
 			select {
-			case data, ok := <-dataIn:
-				if !ok {
-					return // lstIn channel closed
-				}
-				for i := 0; i < len(data.List()); i++ {
-					select {
-					case <-ctx.Done():
-						return
-					case seqOut <- data.List()[i]:
-					}
-				}
+			case <-ctx.Done():
+				return
+			case dataMsg := <-dataIn:
+				list = dataMsg.List()
+			}
+
+			for idx := 0; idx < len(list); idx++ {
+				item := streamItem(
+					list[idx],
+					int64(idx),
+					idx == len(list)-1,
+				)
+
 				select {
 				case <-ctx.Done():
 					return
-				case seqOut <- nil:
+				case seqOut <- item:
 				}
-			case <-ctx.Done():
-				return
 			}
 		}
 	}, nil

internal/runtime/funcs/port_streamer.go

@@ -7,9 +7,9 @@ import (
 	"github.com/nevalang/neva/internal/runtime"
 )
 
-type portSequencer struct{}
+type portStreamer struct{}
 
-func (portSequencer) Create(
+func (portStreamer) Create(
 	io runtime.FuncIO,
 	_ runtime.Msg,
 ) (func(context.Context), error) {
@@ -24,28 +24,29 @@ func (portSequencer) Create(
 	}
 
 	return func(ctx context.Context) {
-		var msg runtime.Msg
-
 		for {
-			for _, slot := range portIn {
+			l := len(portIn)
+
+			for i, slot := range portIn {
+				var msg runtime.Msg
 				select {
 				case <-ctx.Done():
 					return
 				case msg = <-slot:
 				}
 
+				item := streamItem(
+					msg,
+					int64(i),
+					i == l-1,
+				)
+
 				select {
 				case <-ctx.Done():
 					return
-				case streamOut <- msg:
+				case streamOut <- item:
 				}
 			}
-
-			select {
-			case <-ctx.Done():
-				return
-			case streamOut <- nil:
-			}
 		}
 	}, nil
 }