Add length field to DataStreamFactory

Author: sebastianrath

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{})