V.0.1.0

README.md

@@ -8,25 +8,26 @@
 
 <h1>What is it?</h1>
 <p>F-Mesh is a simplistic FBP-inspired framework in Go. 
-It allows you to express your program as a mesh of interconnected components.</p>
+It allows you to express your program as a mesh of interconnected components.
+You can think of it as a simple functions orchestrator.
+</p>
 <h3>Main concepts:</h3>
 <ul>
 <li>F-Mesh consists of multiple <b>Components</b> - the main building blocks</li>
-<li>Components have unlimited number of input and output <p>Ports</p>. </li>
+<li>Components have unlimited number of input and output <p>Ports</p></li>
 <li>The main job of each component is to read inputs and provide outputs</li>
-<li>Each port of the component can be connected to one or multiple ports of any other component. Such connections are called <b>Pipes</b></li>
-<li>It does not matter whether port is input or output, any port can be connected to any other port</li>
+<li>Any output port can be connected to any input port via <b>Pipes</b></li>
 <li>The component behaviour is defined by its <b>Activation function</b></li>
 <li>The framework checks when components are ready to be activated and calls their activation functions concurrently</li>
 <li>One such iteration is called <b>Activation cycle</b></li>
-<li>On each activation cycle the framework does same things: activates all the components ready to be activated, flushes the data through pipes and disposes processed <b>Signals (the data chunks flowing between components)</b></li>
+<li>On each activation cycle the framework does same things: activates all the components ready for activation, flushes the data through pipes and disposes input <b>Signals (the data chunks flowing between components)</b></li>
 <li>Ports and pipes are type agnostic, any data can be transferred or aggregated on any port</li>
 <li>The framework works in discrete time, not it wall time. The quant of time is 1 activation cycle, which gives you "logical parallelism" out of the box</li>
 <li>F-Mesh is suitable for logical wireframing, simulation, functional-style computations and implementing simple concurrency patterns without using the concurrency primitives like channels or any sort of locks</li>
 </ul>
 
 <h1>What it is not?</h1>
-<p>F-mesh is not a classical FBP implementation, and it is not fully async. It does not support long-running components or wall-time events (like timers and tickers).</p>
+<p>F-mesh is not a classical FBP implementation, and it is not fully async. It does not support long-running components or wall-time events (like timers and tickers)</p>
 <p>The framework is not suitable for implementing complex concurrent systems</p>
 
 <h2>Example:</h2>
@@ -54,7 +55,10 @@ It allows you to express your program as a mesh of interconnected components.</p
 					outputs.ByName("res").PutSignals(signal.New(strings.ToTitle(inputString)))
 					return nil
 				})).
-		WithErrorHandlingStrategy(fmesh.StopOnFirstErrorOrPanic)
+                .WithConfig(fmesh.Config{
+                                    ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic,
+                                    CyclesLimit:           10,
+                                    })
 
 	fm.Components().ByName("concat").Outputs().ByName("res").PipeTo(
 		fm.Components().ByName("case").Inputs().ByName("i1"),

component/activation_result.go

@@ -1,6 +1,7 @@
 package component
 
 import (
+	"errors"
 	"fmt"
 )
 
@@ -30,6 +31,12 @@
 
 	// ActivationCodePanicked : component is activated, but panicked
 	ActivationCodePanicked
+
+	// ActivationCodeWaitingForInputs : component waits for specific inputs, but all input signals in current activation cycle may be cleared (default behaviour)
+	ActivationCodeWaitingForInputsClear
+
+	// ActivationCodeWaitingForInputsKeep : component waits for specific inputs, but wants to keep current input signals for the next cycle
+	ActivationCodeWaitingForInputsKeep
 )
 
 // NewActivationResult creates a new activation result for given component
@@ -60,13 +67,13 @@
 	return ar.code
 }
 
-// HasError returns true when activation result has an error
-func (ar *ActivationResult) HasError() bool {
+// IsError returns true when activation result has an error
+func (ar *ActivationResult) IsError() bool {
 	return ar.code == ActivationCodeReturnedError && ar.Error() != nil
 }
 
-// HasPanic returns true when activation result is derived from panic
-func (ar *ActivationResult) HasPanic() bool {
+// IsPanic returns true when activation result is derived from panic
+func (ar *ActivationResult) IsPanic() bool {
 	return ar.code == ActivationCodePanicked && ar.Error() != nil
 }
 
@@ -125,3 +132,23 @@
 		WithActivationCode(ActivationCodePanicked).
 		WithError(err)
 }
+
+func (c *Component) newActivationResultWaitingForInputs(err error) *ActivationResult {
+	activationCode := ActivationCodeWaitingForInputsClear
+	if errors.Is(err, errWaitingForInputsKeep) {
+		activationCode = ActivationCodeWaitingForInputsKeep
+	}
+	return NewActivationResult(c.Name()).
+		SetActivated(true).
+		WithActivationCode(activationCode).
+		WithError(err)
+}
+
+func IsWaitingForInput(activationResult *ActivationResult) bool {
+	return activationResult.Code() == ActivationCodeWaitingForInputsClear ||
+		activationResult.Code() == ActivationCodeWaitingForInputsKeep
+}
+
+func WantsToKeepInputs(activationResult *ActivationResult) bool {
+	return activationResult.Code() == ActivationCodeWaitingForInputsKeep
+}

component/activation_result_collection.go

@@ -19,7 +19,7 @@ func (collection ActivationResultCollection) Add(activationResults ...*Activatio
 // HasErrors tells whether the collection contains at least one activation result with error and respective code
 func (collection ActivationResultCollection) HasErrors() bool {
 	for _, ar := range collection {
-		if ar.HasError() {
+		if ar.IsError() {
 			return true
 		}
 	}
@@ -29,7 +29,7 @@ func (collection ActivationResultCollection) HasErrors() bool {
 // HasPanics tells whether the collection contains at least one activation result with panic and respective code
 func (collection ActivationResultCollection) HasPanics() bool {
 	for _, ar := range collection {
-		if ar.HasPanic() {
+		if ar.IsPanic() {
 			return true
 		}
 	}

component/collection.go

@@ -3,8 +3,8 @@ package component
 // Collection is a collection of components with useful methods
 type Collection map[string]*Component
 
-// NewComponentCollection creates empty collection
-func NewComponentCollection() Collection {
+// NewCollection creates empty collection
+func NewCollection() Collection {
 	return make(Collection)
 }
 
@@ -13,8 +13,8 @@ func (collection Collection) ByName(name string) *Component {
 	return collection[name]
 }
 
-// Add adds components to existing collection
-func (collection Collection) Add(components ...*Component) Collection {
+// With adds components and returns the collection
+func (collection Collection) With(components ...*Component) Collection {
 	for _, component := range components {
 		collection[component.Name()] = component
 	}

component/collection_test.go

@@ -18,7 +18,7 @@ func TestCollection_ByName(t *testing.T) {
 	}{
 		{
 			name:       "component found",
-			components: NewComponentCollection().Add(New("c1"), New("c2")),
+			components: NewCollection().With(New("c1"), New("c2")),
 			args: args{
 				name: "c2",
 			},
@@ -32,7 +32,7 @@ func TestCollection_ByName(t *testing.T) {
 		},
 		{
 			name:       "component not found",
-			components: NewComponentCollection().Add(New("c1"), New("c2")),
+			components: NewCollection().With(New("c1"), New("c2")),
 			args: args{
 				name: "c3",
 			},
@@ -46,7 +46,7 @@ func TestCollection_ByName(t *testing.T) {
 	}
 }
 
-func TestCollection_Add(t *testing.T) {
+func TestCollection_With(t *testing.T) {
 	type args struct {
 		components []*Component
 	}
@@ -58,7 +58,7 @@ func TestCollection_Add(t *testing.T) {
 	}{
 		{
 			name:       "adding nothing to empty collection",
-			collection: NewComponentCollection(),
+			collection: NewCollection(),
 			args: args{
 				components: nil,
 			},
@@ -68,7 +68,7 @@ func TestCollection_Add(t *testing.T) {
 		},
 		{
 			name:       "adding to empty collection",
-			collection: NewComponentCollection(),
+			collection: NewCollection(),
 			args: args{
 				components: []*Component{New("c1"), New("c2")},
 			},
@@ -81,7 +81,7 @@ func TestCollection_Add(t *testing.T) {
 		},
 		{
 			name:       "adding to non-empty collection",
-			collection: NewComponentCollection().Add(New("c1"), New("c2")),
+			collection: NewCollection().With(New("c1"), New("c2")),
 			args: args{
 				components: []*Component{New("c3"), New("c4")},
 			},
@@ -97,9 +97,9 @@ func TestCollection_Add(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			tt.collection.Add(tt.args.components...)
+			collectionAfter := tt.collection.With(tt.args.components...)
 			if tt.assertions != nil {
-				tt.assertions(t, tt.collection)
+				tt.assertions(t, collectionAfter)
 			}
 		})
 	}

component/component.go

@@ -1,6 +1,7 @@
 package component
 
 import (
+	"errors"
 	"fmt"
 	"github.com/hovsep/fmesh/port"
 )
@@ -89,10 +90,6 @@ func (c *Component) hasActivationFunction() bool {
 // MaybeActivate tries to run the activation function if all required conditions are met
 // @TODO: hide this method from user
 func (c *Component) MaybeActivate() (activationResult *ActivationResult) {
-	defer func() {
-		c.Inputs().Clear()
-	}()
-
 	defer func() {
 		if r := recover(); r != nil {
 			activationResult = c.newActivationResultPanicked(fmt.Errorf("panicked with: %v", r))
@@ -102,7 +99,6 @@ func (c *Component) MaybeActivate() (activationResult *ActivationResult) {
 	if !c.hasActivationFunction() {
 		//Activation function is not set (maybe useful while the mesh is under development)
 		activationResult = c.newActivationResultNoFunction()
-
 		return
 	}
 
@@ -115,14 +111,17 @@ func (c *Component) MaybeActivate() (activationResult *ActivationResult) {
 	//Invoke the activation func
 	err := c.f(c.Inputs(), c.Outputs())
 
+	if errors.Is(err, errWaitingForInputs) {
+		activationResult = c.newActivationResultWaitingForInputs(err)
+		return
+	}
+
 	if err != nil {
 		activationResult = c.newActivationResultReturnedError(err)
-
 		return
 	}
 
 	activationResult = c.newActivationResultOK()
-
 	return
 }
 
@@ -132,3 +131,8 @@ func (c *Component) FlushOutputs() {
 		out.Flush()
 	}
 }
+
+// ClearInputs clears all input ports
+func (c *Component) ClearInputs() {
+	c.Inputs().Clear()
+}

component/component_test.go

@@ -501,10 +501,10 @@ func TestComponent_MaybeActivate(t *testing.T) {
 			assert.Equal(t, tt.wantActivationResult.Activated(), gotActivationResult.Activated())
 			assert.Equal(t, tt.wantActivationResult.ComponentName(), gotActivationResult.ComponentName())
 			assert.Equal(t, tt.wantActivationResult.Code(), gotActivationResult.Code())
-			if tt.wantActivationResult.HasError() {
+			if tt.wantActivationResult.IsError() {
 				assert.EqualError(t, gotActivationResult.Error(), tt.wantActivationResult.Error().Error())
 			} else {
-				assert.False(t, gotActivationResult.HasError())
+				assert.False(t, gotActivationResult.IsError())
 			}
 
 		})

component/errors.go

@@ -0,0 +1,19 @@
+package component
+
+import (
+	"errors"
+	"fmt"
+)
+
+var (
+	errWaitingForInputs     = errors.New("component is waiting for some inputs")
+	errWaitingForInputsKeep = fmt.Errorf("%w: do not clear input ports", errWaitingForInputs)
+)
+
+// NewErrWaitForInputs returns respective error
+func NewErrWaitForInputs(keepInputs bool) error {
+	if keepInputs {
+		return errWaitingForInputsKeep
+	}
+	return errWaitingForInputs
+}

fmesh.go

@@ -32,7 +32,7 @@ type FMesh struct {
 func New(name string) *FMesh {
 	return &FMesh{
 		name:       name,
-		components: component.NewComponentCollection(),
+		components: component.NewCollection(),
 		config:     defaultConfig,
 	}
 }
@@ -60,7 +60,7 @@ func (fm *FMesh) WithDescription(description string) *FMesh {
 // WithComponents adds components to f-mesh
 func (fm *FMesh) WithComponents(components ...*component.Component) *FMesh {
 	for _, c := range components {
-		fm.components.Add(c)
+		fm.components = fm.components.With(c)
 	}
 	return fm
 }
@@ -106,7 +106,17 @@ func (fm *FMesh) drainComponents(cycle *cycle.Cycle) {
 			continue
 		}
 
+		if component.IsWaitingForInput(activationResult) {
+			if !component.WantsToKeepInputs(activationResult) {
+				c.ClearInputs()
+			}
+			// Components waiting for inputs are not flushed
+			continue
+		}
+
+		// Normally components are fully drained
 		c.FlushOutputs()
+		c.ClearInputs()
 	}
 }
 
@@ -127,7 +137,7 @@ func (fm *FMesh) Run() (cycle.Collection, error) {
 }
 
 func (fm *FMesh) mustStop(cycleResult *cycle.Cycle, cycleNum int) (bool, error) {
-	if (fm.config.CyclesLimit > 0) && (cycleNum >= fm.config.CyclesLimit) {
+	if (fm.config.CyclesLimit > 0) && (cycleNum > fm.config.CyclesLimit) {
 		return true, ErrReachedMaxAllowedCycles
 	}
 

fmesh_test.go

@@ -601,10 +601,10 @@ func TestFMesh_Run(t *testing.T) {
 					assert.Equal(t, tt.want[i].ActivationResults()[componentName].ComponentName(), gotActivationResult.ComponentName())
 					assert.Equal(t, tt.want[i].ActivationResults()[componentName].Code(), gotActivationResult.Code())
 
-					if tt.want[i].ActivationResults()[componentName].HasError() {
+					if tt.want[i].ActivationResults()[componentName].IsError() {
 						assert.EqualError(t, tt.want[i].ActivationResults()[componentName].Error(), gotActivationResult.Error().Error())
 					} else {
-						assert.False(t, gotActivationResult.HasError())
+						assert.False(t, gotActivationResult.IsError())
 					}
 				}
 			}