Add integration test

integration_tests/computation/basic_math_test.go → integration_tests/computation/math_test.go

@@ -10,7 +10,7 @@ import (
 	"testing"
 )
 
-func Test_BasicMath(t *testing.T) {
+func Test_Math(t *testing.T) {
 	tests := []struct {
 		name       string
 		setupFM    func() *fmesh.FMesh

integration_tests/piping/multiplexing_test.go

@@ -0,0 +1,154 @@
+package integration_tests
+
+import (
+	"github.com/hovsep/fmesh"
+	"github.com/hovsep/fmesh/component"
+	"github.com/hovsep/fmesh/cycle"
+	"github.com/hovsep/fmesh/port"
+	"github.com/hovsep/fmesh/signal"
+	"github.com/stretchr/testify/assert"
+	"math/rand"
+	"testing"
+	"time"
+)
+
+func Test_Multiplexing(t *testing.T) {
+	tests := []struct {
+		name       string
+		setupFM    func() *fmesh.FMesh
+		setInputs  func(fm *fmesh.FMesh)
+		assertions func(t *testing.T, fm *fmesh.FMesh, cycles cycle.Collection, err error)
+	}{
+		{
+			name: "fan-out (3 pipes from 1 source port)",
+			setupFM: func() *fmesh.FMesh {
+				fm := fmesh.New("fan-out").WithComponents(
+					component.NewComponent("producer").
+						WithInputs("start").
+						WithOutputs("o1").
+						WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+							outputs.ByName("o1").PutSignal(signal.New(time.Now()))
+							return nil
+						}),
+
+					component.NewComponent("consumer1").
+						WithInputs("i1").
+						WithOutputs("o1").
+						WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+							//Bypass received signal to output
+							port.ForwardSignal(inputs.ByName("i1"), outputs.ByName("o1"))
+							return nil
+						}),
+
+					component.NewComponent("consumer2").
+						WithInputs("i1").
+						WithOutputs("o1").
+						WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+							//Bypass received signal to output
+							port.ForwardSignal(inputs.ByName("i1"), outputs.ByName("o1"))
+							return nil
+						}),
+
+					component.NewComponent("consumer3").
+						WithInputs("i1").
+						WithOutputs("o1").
+						WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+							//Bypass received signal to output
+							port.ForwardSignal(inputs.ByName("i1"), outputs.ByName("o1"))
+							return nil
+						}),
+				)
+
+				fm.Components().ByName("producer").Outputs().ByName("o1").PipeTo(
+					fm.Components().ByName("consumer1").Inputs().ByName("i1"),
+					fm.Components().ByName("consumer2").Inputs().ByName("i1"),
+					fm.Components().ByName("consumer3").Inputs().ByName("i1"))
+
+				return fm
+			},
+			setInputs: func(fm *fmesh.FMesh) {
+				//Fire the mesh
+				fm.Components().ByName("producer").Inputs().ByName("start").PutSignal(signal.New(struct{}{}))
+			},
+			assertions: func(t *testing.T, fm *fmesh.FMesh, cycles cycle.Collection, err error) {
+				//All consumers received a signal
+				c1, c2, c3 := fm.Components().ByName("consumer1"), fm.Components().ByName("consumer2"), fm.Components().ByName("consumer3")
+				assert.True(t, c1.Outputs().ByName("o1").HasSignal())
+				assert.True(t, c2.Outputs().ByName("o1").HasSignal())
+				assert.True(t, c3.Outputs().ByName("o1").HasSignal())
+
+				//All 3 signals are the same (literally the same address in memory)
+				sig1, sig2, sig3 := c1.Outputs().ByName("o1").Signal(), c2.Outputs().ByName("o1").Signal(), c3.Outputs().ByName("o1").Signal()
+				assert.Equal(t, sig1.Payload(), sig2.Payload())
+				assert.Equal(t, sig2.Payload(), sig3.Payload())
+			},
+		},
+		{
+			name: "multiplexing",
+			setupFM: func() *fmesh.FMesh {
+				producer1 := component.NewComponent("producer1").
+					WithInputs("start").
+					WithOutputs("o1").
+					WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+						outputs.ByName("o1").PutSignal(signal.New(rand.Int()))
+						return nil
+					})
+
+				producer2 := component.NewComponent("producer2").
+					WithInputs("start").
+					WithOutputs("o1").
+					WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+						outputs.ByName("o1").PutSignal(signal.New(rand.Int()))
+						return nil
+					})
+
+				producer3 := component.NewComponent("producer3").
+					WithInputs("start").
+					WithOutputs("o1").
+					WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+						outputs.ByName("o1").PutSignal(signal.New(rand.Int()))
+						return nil
+					})
+				consumer := component.NewComponent("consumer").
+					WithInputs("i1").
+					WithOutputs("o1").
+					WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+						//Bypass
+						port.ForwardSignal(inputs.ByName("i1"), outputs.ByName("o1"))
+						return nil
+					})
+
+				producer1.Outputs().ByName("o1").PipeTo(consumer.Inputs().ByName("i1"))
+				producer2.Outputs().ByName("o1").PipeTo(consumer.Inputs().ByName("i1"))
+				producer3.Outputs().ByName("o1").PipeTo(consumer.Inputs().ByName("i1"))
+
+				return fmesh.New("multiplexer").WithComponents(producer1, producer2, producer3, consumer)
+			},
+			setInputs: func(fm *fmesh.FMesh) {
+				fm.Components().ByName("producer1").Inputs().ByName("start").PutSignal(signal.New(struct{}{}))
+				fm.Components().ByName("producer2").Inputs().ByName("start").PutSignal(signal.New(struct{}{}))
+				fm.Components().ByName("producer3").Inputs().ByName("start").PutSignal(signal.New(struct{}{}))
+			},
+			assertions: func(t *testing.T, fm *fmesh.FMesh, cycles cycle.Collection, err error) {
+				//Consumer received a signal
+				assert.True(t, fm.Components().ByName("consumer").Outputs().ByName("o1").HasSignal())
+
+				//The signal is combined and consist of 3 payloads
+				resultSignal := fm.Components().ByName("consumer").Outputs().ByName("o1").Signal()
+				assert.Equal(t, resultSignal.Len(), 3)
+
+				//And they are all different
+				assert.NotEqual(t, resultSignal.Payloads()[0], resultSignal.Payloads()[1])
+				assert.NotEqual(t, resultSignal.Payloads()[1], resultSignal.Payloads()[2])
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fm := tt.setupFM()
+			tt.setInputs(fm)
+			cycles, err := fm.Run()
+			tt.assertions(t, fm, cycles, err)
+		})
+	}
+}