Add stream support to length node

.github/workflows/graphs/build-quick.act

@@ -119,6 +119,12 @@ executions:
     dst:
       node: branch-v1-penguin-monkey-kiwi
       port: exec
+  - src:
+      node: gh-start
+      port: exec-on_pull_request
+    dst:
+      node: gh-actions-checkout-gray-butterfly-apple
+      port: exec
   - src:
       node: gh-start
       port: exec-on_workflow_dispatch

.github/workflows/graphs/build-test-publish.act

@@ -6272,6 +6272,12 @@ executions:
     dst:
       node: group-v1-orange-tiger-grapefruit
       port: exec-lion-beige-dragonfruit
+  - src:
+      node: gh-start
+      port: exec-on_pull_request
+    dst:
+      node: group-v1-orange-tiger-grapefruit
+      port: exec-lion-beige-dragonfruit
   - src:
       node: gh-start
       port: exec-on_workflow_dispatch

.github/workflows/graphs/test_env.act

@@ -266,6 +266,12 @@ executions:
     dst:
       node: run-v1-apple-silver-lion
       port: exec
+  - src:
+      node: gh-start
+      port: exec-on_pull_request
+    dst:
+      node: run-v1-apple-silver-lion
+      port: exec
   - src:
       node: gh-start
       port: exec-on_workflow_dispatch

.github/workflows/workflow.yml

@@ -5,6 +5,10 @@ on:
     branches:
       - main
 
+  pull_request:
+    branches:
+      - main
+
   workflow_dispatch:
     inputs:
       session_token:

core/base.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"io"
@@ -83,6 +84,7 @@ type DataStreamFactory struct {
 	SourcePath     string
 	SourceProvider any
 	Reader         io.Reader
+	Length         int64
 }
 
 func (dsf *DataStreamFactory) CloseStream() error {
@@ -93,6 +95,25 @@ func (dsf *DataStreamFactory) CloseStreamAndIgnoreError() {
 	_ = dsf.CloseStream()
 }
 
+func GetReaderLength(r io.Reader) int64 {
+	switch v := r.(type) {
+	case *bytes.Buffer:
+		return int64(v.Len())
+	case *bytes.Reader:
+		return int64(v.Len())
+	case *strings.Reader:
+		return int64(v.Len())
+	case *os.File:
+		stat, err := v.Stat()
+		if err != nil {
+			return 0
+		}
+		return stat.Size()
+	default:
+		return -1
+	}
+}
+
 var (
 	onceReIsExec sync.Once
 	reIsExec     *regexp.Regexp
@@ -130,9 +151,11 @@ type NodeBaseInterface interface {
 	SetId(id string)
 	GetNodeTypeId() string
 	GetName() string
+	GetLabel() string
 	GetId() string
 	GetCacheId() string
 	SetName(name string)
+	SetLabel(label string)
 	GetGraph() *ActionGraph
 
 	// Instead of checking for 'HasExecutionInterface',
@@ -151,6 +174,7 @@ type NodeBaseInterface interface {
 // The node that implements this component has outgoing connections.
 type NodeBaseComponent struct {
 	Name            string // Human readable name of the node
+	Label           string // Label of the node shown in the graph editor
 	Id              string // Unique identifier for the node
 	FullPath        string // Full path of the node within the graph hierarchy
 	CacheId         string // Unique identifier for the cache
@@ -221,10 +245,18 @@ func (n *NodeBaseComponent) GetName() string {
 	return n.Name
 }
 
+func (n *NodeBaseComponent) GetLabel() string {
+	return n.Label
+}
+
 func (n *NodeBaseComponent) SetName(name string) {
 	n.Name = name
 }
 
+func (n *NodeBaseComponent) SetLabel(label string) {
+	n.Label = label
+}
+
 func IsValidIndexPortId(id string) (string, int, bool) {
 	indexPortMatch := getIndexPortRegex().FindStringSubmatch(id)
 	if len(indexPortMatch) < 3 {

core/inputs.go

@@ -603,7 +603,7 @@ func ConvertValue(c *ExecutionState, v reflect.Value, requestedType reflect.Type
 			if err != nil {
 				return nil, err
 			}
-			return DataStreamFactory{Reader: reader}, nil
+			return DataStreamFactory{Reader: reader, Length: GetReaderLength(reader)}, nil
 		case reflectValueType:
 			// It is common to request a value with the type `any`.
 			// ... core.InputValueById[any](...)

nodes/file-compress@v1.go

@@ -72,6 +72,7 @@ func (n *FileZipNode) ExecuteImpl(c *core.ExecutionState, inputId core.InputId,
 
 	dsf := core.DataStreamFactory{
 		Reader: reader,
+		Length: core.GetReaderLength(reader),
 	}
 
 	err = n.Outputs.SetOutputValue(c, ni.Core_file_compress_v1_Output_data, dsf, core.SetOutputValueOpts{})

nodes/file-read@v1.go

@@ -30,12 +30,12 @@ func (n *FileReadNode) ExecuteImpl(c *core.ExecutionState, inputId core.InputId,
 		return core.CreateErr(c, err)
 	}
 
-	defer fp.Close()
-
 	dsf := core.DataStreamFactory{
 		SourcePath: path,
 		Reader:     fp,
+		Length:     core.GetReaderLength(fp),
 	}
+	defer dsf.CloseStreamAndIgnoreError()
 
 	err = n.Outputs.SetOutputValue(c, ni.Core_file_read_v1_Output_data, dsf, core.SetOutputValueOpts{})
 	if err != nil {

nodes/http@v1.go

@@ -118,6 +118,7 @@ func (n *HttpNode) ExecuteImpl(c *core.ExecutionState, inputId core.InputId, pre
 	var statusCode int
 	if resp != nil {
 		dsf.Reader = resp.Body
+		dsf.Length = resp.ContentLength
 		statusCode = resp.StatusCode
 	}
 

nodes/length@v1.go

@@ -2,6 +2,7 @@ package nodes
 
 import (
 	_ "embed"
+	"io"
 	"reflect"
 
 	"github.com/actionforge/actrun-cli/core"
@@ -23,8 +24,15 @@ func (n *LengthNode) OutputValueById(c *core.ExecutionState, outputId core.Outpu
 		return nil, err
 	}
 
-	v := reflect.ValueOf(inputs)
+	if dsf, ok := inputs.(core.DataStreamFactory); ok {
+		if dsf.Length != -1 {
+			return dsf.Length, nil
+		}
+	} else if r, ok := inputs.(io.Reader); ok {
+		return core.GetReaderLength(r), nil
+	}
 
+	v := reflect.ValueOf(inputs)
 	switch v.Kind() {
 	case reflect.Array, reflect.Map, reflect.Slice, reflect.String:
 		return v.Len(), nil

nodes/random-stream@v1.go

@@ -122,6 +122,7 @@ func (n *RandomStreamNode) OutputValueById(c *core.ExecutionState, outputId core
 	reader := NewRandomStringReader(length, includeNumbers, includeCharacters, includeUppercase, includeSpecial, seed)
 	return core.DataStreamFactory{
 		Reader: reader,
+		Length: core.GetReaderLength(reader),
 	}, nil
 }
 

nodes/storage-download@v1.go

@@ -38,6 +38,7 @@ func (n *StorageDownloadNode) ExecuteImpl(c *core.ExecutionState, inputId core.I
 		SourcePath:     path,
 		SourceProvider: provider,
 		Reader:         reader,
+		Length:         core.GetReaderLength(reader),
 	}
 
 	err = n.Outputs.SetOutputValue(c, ni.Core_storage_download_v1_Output_provider, provider, core.SetOutputValueOpts{})

tests_e2e/references/reference_group-port-collision.sh_l13

@@ -23,7 +23,7 @@ stack trace:
 github.com/actionforge/actrun-cli/nodes.init.39.func1
 	group@v1.go:128
 github.com/actionforge/actrun-cli/core.NewNodeInstance
-	base.go:566
+	base.go:598
 github.com/actionforge/actrun-cli/core.LoadNode
 	graph.go:599
 github.com/actionforge/actrun-cli/core.LoadNodes

tests_e2e/references/reference_length.sh_l10

@@ -0,0 +1,25 @@
+build hasn't expired yet
+looking for value: 'env_file'
+  no value (is optional) found for: 'env_file'
+looking for value: 'config_file'
+  no value (is optional) found for: 'config_file'
+looking for value: 'concurrency'
+  no value (is optional) found for: 'concurrency'
+looking for value: 'graph_file'
+  found value in: 'env (shell)'
+  evaluated to: 'length.act'
+looking for value: 'session_token'
+  no value (is optional) found for: 'session_token'
+looking for value: 'create_debug_session'
+  found value in flags
+  evaluated to: 'false'
+PushNodeVisit: start, execute: true
+🟢 Execute 'File Read (core-file-read-v1-banana-gold-fox)'
+PushNodeVisit: core-file-read-v1-banana-gold-fox, execute: true
+🟢 Execute 'Print (core-print-v1-apricot-teal-purple)'
+PushNodeVisit: core-print-v1-apricot-teal-purple, execute: true
+PushNodeVisit: (cached) core-file-read-v1-banana-gold-fox, execute: false
+PushNodeVisit: core-length-v1-grape-lobster-camel, execute: false
+PushNodeVisit: (cached) core-file-read-v1-banana-gold-fox, execute: false
+hello world
+11

tests_e2e/references/reference_run-python-embedded.sh_l13

@@ -27,7 +27,7 @@ stack trace:
 github.com/actionforge/actrun-cli/nodes.init.50.func1
 	nrun-python-embedded@v1.go:16
 github.com/actionforge/actrun-cli/core.NewNodeInstance
-	base.go:566
+	base.go:598
 github.com/actionforge/actrun-cli/core.LoadNode
 	graph.go:599
 github.com/actionforge/actrun-cli/core.LoadNodes

tests_e2e/scripts/length.act

@@ -0,0 +1,63 @@
+editor:
+  version:
+    created: v1.34.0
+entry: start
+type: standard
+nodes:
+  - id: start
+    type: core/start@v1
+    position:
+      x: -520
+      y: -110
+  - id: core-file-read-v1-banana-gold-fox
+    type: core/file-read@v1
+    position:
+      x: -170
+      y: -190
+    inputs:
+      path: length.txt
+  - id: core-print-v1-apricot-teal-purple
+    type: core/print@v1
+    position:
+      x: 620
+      y: -220
+    inputs:
+      values[0]: null
+      values[1]: null
+  - id: core-length-v1-grape-lobster-camel
+    type: core/length@v1
+    position:
+      x: 270
+      y: 40
+connections:
+  - src:
+      node: core-file-read-v1-banana-gold-fox
+      port: data
+    dst:
+      node: core-length-v1-grape-lobster-camel
+      port: input
+  - src:
+      node: core-length-v1-grape-lobster-camel
+      port: result
+    dst:
+      node: core-print-v1-apricot-teal-purple
+      port: values[1]
+  - src:
+      node: core-file-read-v1-banana-gold-fox
+      port: data
+    dst:
+      node: core-print-v1-apricot-teal-purple
+      port: values[0]
+executions:
+  - src:
+      node: core-file-read-v1-banana-gold-fox
+      port: exec-success
+    dst:
+      node: core-print-v1-apricot-teal-purple
+      port: exec
+  - src:
+      node: start
+      port: exec
+    dst:
+      node: core-file-read-v1-banana-gold-fox
+      port: exec

tests_e2e/scripts/length.sh

@@ -0,0 +1,11 @@
+echo "Test Length"
+
+TEST_NAME=length
+GRAPH_FILE="${ACT_GRAPH_FILES_DIR}${PATH_SEPARATOR}${TEST_NAME}.act"
+TEST_FILE="${ACT_GRAPH_FILES_DIR}${PATH_SEPARATOR}${TEST_NAME}.txt"
+cp $GRAPH_FILE $TEST_NAME.act
+cp $TEST_FILE $TEST_NAME.txt
+export ACT_GRAPH_FILE=$TEST_NAME.act
+
+#! test actrun
+

tests_e2e/scripts/length.txt

@@ -0,0 +1 @@
+hello world