stream: Add Zip for combining 2 streams

examples/stream_zip/e2e_test.go

@@ -0,0 +1,33 @@
+package test
+
+import (
+	"os"
+	"os/exec"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test(t *testing.T) {
+	err := os.Chdir("..")
+	require.NoError(t, err)
+
+	wd, err := os.Getwd()
+	require.NoError(t, err)
+	defer os.Chdir(wd)
+
+	cmd := exec.Command("neva", "run", "stream_zip")
+
+	out, err := cmd.CombinedOutput()
+	require.NoError(t, err)
+	require.Equal(
+		t,
+		`{"first": 0, "second": "a"}
+{"first": 1, "second": "b"}
+{"first": 2, "second": "c"}
+`,
+		string(out),
+	)
+
+	require.Equal(t, 0, cmd.ProcessState.ExitCode())
+}

examples/stream_zip/main.neva

@@ -0,0 +1,17 @@
+const strs list<string> = ['a', 'b', 'c']
+
+component Main(start) (stop) {
+  nodes {
+    r1 Range,
+    Iter<string>
+    Zip<int, string>,
+    For<ZipRes<int, string>>{Println<ZipRes<int, string>>}
+  }
+  :start -> [
+    (0 -> r1:from),
+    (3 -> r1:to)
+  ]
+  r1 -> zip:first
+  $strs -> iter -> zip:second
+  zip -> for -> :stop
+}

internal/runtime/funcs/registry.go

@@ -34,6 +34,7 @@ func CreatorRegistry() map[string]runtime.FuncCreator {
 		"list_to_stream":       listToStream{},
 		"stream_int_range":     streamIntRange{},
 		"stream_product":       streamProduct{},
+		"stream_zip":           streamZip{},
 
 		// builders
 		"struct_builder": structBuilder{},

internal/runtime/funcs/stream_zip.go

@@ -0,0 +1,82 @@
+package funcs
+
+import (
+	"context"
+
+	"github.com/nevalang/neva/internal/runtime"
+)
+
+type streamZip struct{}
+
+func (streamZip) Create(
+	io runtime.FuncIO,
+	_ runtime.Msg,
+) (func(ctx context.Context), error) {
+	firstIn, err := io.In.Port("first")
+	if err != nil {
+		return nil, err
+	}
+
+	secondIn, err := io.In.Port("second")
+	if err != nil {
+		return nil, err
+	}
+
+	seqOut, err := io.Out.Port("seq")
+	if err != nil {
+		return nil, err
+	}
+
+	return func(ctx context.Context) {
+		for {
+			var firstData, secondData []runtime.Msg
+			for {
+				var item map[string]runtime.Msg
+				select {
+				case <-ctx.Done():
+					return
+				case seqMsg := <-firstIn:
+					item = seqMsg.Map()
+				}
+				if firstData = append(firstData, item["data"]); item["last"].Bool() {
+					break
+				}
+			}
+			for {
+				var item map[string]runtime.Msg
+				select {
+				case <-ctx.Done():
+					return
+				case seqMsg := <-secondIn:
+					item = seqMsg.Map()
+				}
+				if secondData = append(secondData, item["data"]); item["last"].Bool() {
+					break
+				}
+			}
+
+			n := len(firstData)
+			if m := len(secondData); m < n {
+				n = m
+			}
+
+			idx := int64(0)
+			for i, e1 := range firstData {
+				e2 := secondData[i]
+				select {
+				case <-ctx.Done():
+					return
+				case seqOut <- runtime.NewMapMsg(map[string]runtime.Msg{
+					"idx":  runtime.NewIntMsg(idx),
+					"last": runtime.NewBoolMsg(i == n-1),
+					"data": runtime.NewMapMsg(map[string]runtime.Msg{
+						"first":  e1,
+						"second": e2,
+					}),
+				}):
+					idx++
+				}
+			}
+		}
+	}, nil
+}

std/builtin/streams.neva

@@ -95,4 +95,12 @@ pub type ProductRes<T, R> struct {
 } 
 
 #extern(stream_product)
-pub component Product<T, R>(first stream<T>, second stream<R>) (seq stream<ProductRes<T, R>>)
+pub component Product<T, R>(first stream<T>, second stream<R>) (seq stream<ProductRes<T, R>>)
+
+pub type ZipRes<T, R> struct {
+    first T
+    second R
+}
+
+#extern(stream_zip)
+pub component Zip<T, R>(first stream<T>, second stream<R>) (seq stream<ZipRes<T, R>>)