Merge pull request #41 from hovsep/v.0.1.0

V.0.1.0

component/component.go

@@ -34,13 +34,13 @@ func (c *Component) WithDescription(description string) *Component {
 
 // WithInputs ads input ports
 func (c *Component) WithInputs(portNames ...string) *Component {
- c.inputs.Add(port.NewPortGroup(portNames...)...)
+ c.inputs = c.inputs.Add(port.NewPortGroup(portNames...)...)
  return c
 }
 
 // WithOutputs adds output ports
 func (c *Component) WithOutputs(portNames ...string) *Component {
- c.outputs.Add(port.NewPortGroup(portNames...)...)
+ c.outputs = c.outputs.Add(port.NewPortGroup(portNames...)...)
  return c
 }
 
@@ -92,7 +92,6 @@ func (c *Component) MaybeActivate() (activationResult *ActivationResult) {
  return
  }
 
- //@TODO:: https://github.com/hovsep/fmesh/issues/15
  if !c.inputs.AnyHasSignal() {
  //No inputs set, stop here
  activationResult = c.newActivationCodeNoInput()
@@ -103,10 +102,10 @@ func (c *Component) MaybeActivate() (activationResult *ActivationResult) {
  //Run the computation
  err := c.f(c.inputs, c.outputs)
 
- if IsWaitingForInputError(err) {
+ if errors.Is(err, errWaitingForInputs) {
  activationResult = c.newActivationCodeWaitingForInput()
 
- if !errors.Is(err, ErrWaitingForInputKeepInputs) {
+ if !errors.Is(err, errWaitingForInputsKeep) {
  c.inputs.ClearSignal()
  }
 

component/component_test.go

@@ -171,8 +171,8 @@ func TestComponent_Inputs(t *testing.T) {
  name: "with inputs",
  component: NewComponent("c1").WithInputs("i1", "i2"),
  want: port.Collection{
- "i1": port.NewPort("i1"),
- "i2": port.NewPort("i2"),
+ port.NewPort("i1"),
+ port.NewPort("i2"),
  },
  },
  }
@@ -200,8 +200,8 @@ func TestComponent_Outputs(t *testing.T) {
  name: "with outputs",
  component: NewComponent("c1").WithOutputs("o1", "o2"),
  want: port.Collection{
- "o1": port.NewPort("o1"),
- "o2": port.NewPort("o2"),
+ port.NewPort("o1"),
+ port.NewPort("o2"),
  },
  },
  }
@@ -306,8 +306,8 @@ func TestComponent_WithInputs(t *testing.T) {
  name: "c1",
  description: "",
  inputs: port.Collection{
- "p1": port.NewPort("p1"),
- "p2": port.NewPort("p2"),
+ port.NewPort("p1"),
+ port.NewPort("p2"),
  },
  outputs: port.Collection{},
  f: nil,
@@ -358,8 +358,8 @@ func TestComponent_WithOutputs(t *testing.T) {
  description: "",
  inputs: port.Collection{},
  outputs: port.Collection{
- "p1": port.NewPort("p1"),
- "p2": port.NewPort("p2"),
+ port.NewPort("p1"),
+ port.NewPort("p2"),
  },
  f: nil,
  },
@@ -420,7 +420,7 @@ func TestComponent_MaybeActivate(t *testing.T) {
  WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
 
  if !inputs.ByNames("i1", "i2").AllHaveSignal() {
- return ErrWaitingForInputResetInputs
+ return NewErrWaitForInputs(false)
  }
 
  return nil
@@ -432,14 +432,35 @@ func TestComponent_MaybeActivate(t *testing.T) {
  WithActivationCode(ActivationCodeNoInput),
  },
  {
- name: "component is waiting for input",
+ name: "component is waiting for input, reset inputs",
  getComponent: func() *Component {
  c := NewComponent("c1").
  WithInputs("i1", "i2").
  WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
 
  if !inputs.ByNames("i1", "i2").AllHaveSignal() {
- return ErrWaitingForInputResetInputs
+ return NewErrWaitForInputs(false)
+ }
+
+ return nil
+ })
+ //Only one input set
+ c.Inputs().ByName("i1").PutSignal(signal.New(123))
+ return c
+ },
+ wantActivationResult: NewActivationResult("c1").
+ SetActivated(false).
+ WithActivationCode(ActivationCodeWaitingForInput),
+ },
+ {
+ name: "component is waiting for input, keep inputs",
+ getComponent: func() *Component {
+ c := NewComponent("c1").
+ WithInputs("i1", "i2").
+ WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
+
+ if !inputs.ByNames("i1", "i2").AllHaveSignal() {
+ return NewErrWaitForInputs(true)
  }
 
  return nil

component/errors.go

@@ -1,13 +1,19 @@
 package component
 
-import "errors"
+import (
+ "errors"
+ "fmt"
+)
 
 var (
- //@TODO: provide wrapper methods so exact input can be specified within error
- ErrWaitingForInputResetInputs = errors.New("component is waiting for one or more inputs. All inputs will be reset")
- ErrWaitingForInputKeepInputs = errors.New("component is waiting for one or more inputs. All inputs will be kept")
+ errWaitingForInputs = errors.New("component is waiting for some inputs")
+ errWaitingForInputsKeep = fmt.Errorf("%w: do not clear input ports", errWaitingForInputs)
 )
 
-func IsWaitingForInputError(err error) bool {
- return errors.Is(err, ErrWaitingForInputResetInputs) || errors.Is(err, ErrWaitingForInputKeepInputs)
+// NewErrWaitForInputs returns respective error
+func NewErrWaitForInputs(keepInputs bool) error {
+ if keepInputs {
+ return errWaitingForInputsKeep
+ }
+ return errWaitingForInputs
 }

fmesh.go

@@ -53,45 +53,30 @@ func (fm *FMesh) WithErrorHandlingStrategy(strategy ErrorHandlingStrategy) *FMes
  return fm
 }
 
-// runCycle runs one activation cycle (tries to activate all components)
+// runCycle runs one activation cycle (tries to activate ready components)
 func (fm *FMesh) runCycle() *cycle.Cycle {
- cycleResult := cycle.New()
+ newCycle := cycle.New()
 
  if len(fm.components) == 0 {
- return cycleResult
+ return newCycle
  }
 
- activationResultsChan := make(chan *component.ActivationResult) //@TODO: close the channel
- doneChan := make(chan struct{}) //@TODO: close the channel
-
  var wg sync.WaitGroup
 
- go func() {
- for {
- select {
- case aRes := <-activationResultsChan:
- //@TODO :check for closed channel
- cycleResult.Lock()
- cycleResult.ActivationResults().Add(aRes)
- cycleResult.Unlock()
- case <-doneChan:
- return
- }
- }
- }()
-
  for _, c := range fm.components {
  wg.Add(1)
- c := c //@TODO: check if this needed
- go func() {
+
+ go func(component *component.Component, cycle *cycle.Cycle) {
  defer wg.Done()
- activationResultsChan <- c.MaybeActivate()
- }()
+
+ cycle.Lock()
+ cycle.ActivationResults().Add(c.MaybeActivate())
+ cycle.Unlock()
+ }(c, newCycle)
  }
 
  wg.Wait()
- doneChan <- struct{}{} //@TODO: no need to send close signal, just close the channel
- return cycleResult
+ return newCycle
 }
 
 // DrainComponents drains the data from all components outputs

fmesh_test.go

@@ -669,7 +669,7 @@ func TestFMesh_runCycle(t *testing.T) {
  WithOutputs("o1").
  WithActivationFunc(func(inputs port.Collection, outputs port.Collection) error {
  if !inputs.ByNames("i1", "i2").AllHaveSignal() {
- return component.ErrWaitingForInputKeepInputs
+ return component.NewErrWaitForInputs(true)
  }
  return nil
  }),

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