V.0.1.0

README.md

@@ -1,4 +1,3 @@
-
 <div align="center">
   <img src="./assets/img/logo.png" width="200" height="200" alt="f-mesh"/>
   <h1>f-mesh</h1>
@@ -7,66 +6,58 @@
 </div>
 
 <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.
-You can think of it as a simple functions orchestrator.
+<p>F-Mesh is a functions orchestrator inspired by FBP. 
+It allows you to express your program as a mesh of interconnected components (or more formally as a computational graph).
 </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 <b>Ports</b></li>
-<li>The main job of each component is to read inputs and provide outputs</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 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>
+<li>Ports and pipes are type agnostic, any data can be transferred to 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 (activation function is running in "frozen time")</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>The framework is not suitable for implementing complex concurrent systems</p>
+<p>F-mesh is not a classical FBP implementation, it does not support long-running components or wall-time events (like timers and tickers)</p>
+
 
 <h2>Example:</h2>
 
 ```go
-	// Create f-mesh
-	fm := fmesh.New("hello world").
+fm := fmesh.New("hello world").
 		WithComponents(
 			component.New("concat").
 				WithInputs("i1", "i2").
 				WithOutputs("res").
-				WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
-					word1 := inputs.ByName("i1").Signals().FirstPayload().(string)
-					word2 := inputs.ByName("i2").Signals().FirstPayload().(string)
+				WithActivationFunc(func(inputs *port.Collection, outputs *port.Collection) error {
+					word1 := inputs.ByName("i1").FirstSignalPayloadOrDefault("").(string)
+					word2 := inputs.ByName("i2").FirstSignalPayloadOrDefault("").(string)
 
 					outputs.ByName("res").PutSignals(signal.New(word1 + word2))
 					return nil
 				}),
 			component.New("case").
 				WithInputs("i1").
 				WithOutputs("res").
-				WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
-					inputString := inputs.ByName("i1").Signals().FirstPayload().(string)
+				WithActivationFunc(func(inputs *port.Collection, outputs *port.Collection) error {
+					inputString := inputs.ByName("i1").FirstSignalPayloadOrDefault("").(string)
 
 					outputs.ByName("res").PutSignals(signal.New(strings.ToTitle(inputString)))
 					return nil
 				})).
-                .WithConfig(fmesh.Config{
-                                    ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic,
-                                    CyclesLimit:           10,
-                                    })
+		WithConfig(fmesh.Config{
+			ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic,
+			CyclesLimit:           10,
+		})
 
 	fm.Components().ByName("concat").Outputs().ByName("res").PipeTo(
 		fm.Components().ByName("case").Inputs().ByName("i1"),
 	)
 
 	// Init inputs
-	fm.Components().ByName("concat").Inputs().ByName("i1").PutSignals(signal.New("hello "))
-	fm.Components().ByName("concat").Inputs().ByName("i2").PutSignals(signal.New("world !"))
+	fm.Components().ByName("concat").InputByName("i1").PutSignals(signal.New("hello "))
+	fm.Components().ByName("concat").InputByName("i2").PutSignals(signal.New("world !"))
 
 	// Run the mesh
 	_, err := fm.Run()
@@ -78,8 +69,8 @@ You can think of it as a simple functions orchestrator.
 	}
 
 	//Extract results
-	results := fm.Components().ByName("case").Outputs().ByName("res").Signals().FirstPayload()
-	fmt.Printf("Result is :%v", results)
+	results := fm.Components().ByName("case").OutputByName("res").FirstSignalPayloadOrNil()
+	fmt.Printf("Result is : %v", results)
 ```
-
-<h2>Version 0.1.0 coming soon</h2>
+See more in ```examples``` directory.
+<h2>Version 0.1.0 coming soon</h2>

common/chainable.go

@@ -0,0 +1,25 @@
+package common
+
+type Chainable struct {
+	err error
+}
+
+// NewChainable initialises new chainable
+func NewChainable() *Chainable {
+	return &Chainable{}
+}
+
+// SetErr sets chainable error
+func (c *Chainable) SetErr(err error) {
+	c.err = err
+}
+
+// HasErr returns true when chainable has error
+func (c *Chainable) HasErr() bool {
+	return c.err != nil
+}
+
+// Err returns chainable error
+func (c *Chainable) Err() error {
+	return c.err
+}

common/described_entity.go

@@ -0,0 +1,15 @@
+package common
+
+type DescribedEntity struct {
+	description string
+}
+
+// NewDescribedEntity constructor
+func NewDescribedEntity(description string) DescribedEntity {
+	return DescribedEntity{description: description}
+}
+
+// Description getter
+func (e DescribedEntity) Description() string {
+	return e.description
+}

common/described_entity_test.go

@@ -0,0 +1,65 @@
+package common
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestNewDescribedEntity(t *testing.T) {
+	type args struct {
+		description string
+	}
+	tests := []struct {
+		name string
+		args args
+		want DescribedEntity
+	}{
+		{
+			name: "empty description",
+			args: args{
+				description: "",
+			},
+			want: DescribedEntity{
+				description: "",
+			},
+		},
+		{
+			name: "with description",
+			args: args{
+				description: "component1 is used to generate logs",
+			},
+			want: DescribedEntity{
+				description: "component1 is used to generate logs",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, NewDescribedEntity(tt.args.description))
+		})
+	}
+}
+
+func TestDescribedEntity_Description(t *testing.T) {
+	tests := []struct {
+		name            string
+		describedEntity DescribedEntity
+		want            string
+	}{
+		{
+			name:            "empty description",
+			describedEntity: NewDescribedEntity(""),
+			want:            "",
+		},
+		{
+			name:            "with description",
+			describedEntity: NewDescribedEntity("component2 is used to handle errors"),
+			want:            "component2 is used to handle errors",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, tt.describedEntity.Description())
+		})
+	}
+}

common/labeled_entity.go

@@ -0,0 +1,97 @@
+package common
+
+import (
+	"errors"
+	"fmt"
+)
+
+type LabelsCollection map[string]string
+
+type LabeledEntity struct {
+	labels LabelsCollection
+}
+
+var (
+	ErrLabelNotFound = errors.New("label not found")
+)
+
+// NewLabeledEntity constructor
+func NewLabeledEntity(labels LabelsCollection) LabeledEntity {
+	return LabeledEntity{labels: labels}
+}
+
+// Labels getter
+func (e *LabeledEntity) Labels() LabelsCollection {
+	return e.labels
+}
+
+// Label returns the value of single label or nil if it is not found
+func (e *LabeledEntity) Label(label string) (string, error) {
+	value, ok := e.labels[label]
+
+	if !ok {
+		return "", fmt.Errorf("label %s not found, %w", label, ErrLabelNotFound)
+	}
+
+	return value, nil
+}
+
+// LabelOrDefault returns label value or default value in case of any error
+func (e *LabeledEntity) LabelOrDefault(label string, defaultValue string) string {
+	value, err := e.Label(label)
+	if err != nil {
+		return defaultValue
+	}
+	return value
+}
+
+// SetLabels overwrites labels collection
+func (e *LabeledEntity) SetLabels(labels LabelsCollection) {
+	e.labels = labels
+}
+
+// AddLabel adds or updates(if label already exists) single label
+func (e *LabeledEntity) AddLabel(label string, value string) {
+	if e.labels == nil {
+		e.labels = make(LabelsCollection)
+	}
+	e.labels[label] = value
+}
+
+// AddLabels adds or updates(if label already exists) multiple labels
+func (e *LabeledEntity) AddLabels(labels LabelsCollection) {
+	for label, value := range labels {
+		e.AddLabel(label, value)
+	}
+}
+
+// DeleteLabel deletes given label
+func (e *LabeledEntity) DeleteLabel(label string) {
+	delete(e.labels, label)
+}
+
+// HasLabel returns true when entity has given label or false otherwise
+func (e *LabeledEntity) HasLabel(label string) bool {
+	_, ok := e.labels[label]
+	return ok
+}
+
+// HasAllLabels checks if entity has all labels
+func (e *LabeledEntity) HasAllLabels(label ...string) bool {
+	for _, l := range label {
+		if !e.HasLabel(l) {
+			return false
+		}
+	}
+	return true
+}
+
+// HasAnyLabel checks if entity has at least one of given labels
+func (e *LabeledEntity) HasAnyLabel(label ...string) bool {
+	for _, l := range label {
+		if e.HasLabel(l) {
+			return true
+		}
+	}
+	return false
+}