diff --git a/cmd/slang/main.go b/cmd/slang/main.go index 4efe2a45..1523c601 100644 --- a/cmd/slang/main.go +++ b/cmd/slang/main.go @@ -148,6 +148,14 @@ func runHttpPost(operator *core.Operator, bind string) { }) + // In order to generate a response to a HTTP-Request we need to have + // the out port fully connected. In some cases the out port is a map of multiple + // types and we need them all connected to their `src`. Otherwise we `pull` on the port + // and wait forever to return something. Which would make the client timeout. + if err := operator.Main().Out().FullyConnected(); err != nil { + log.Fatal(err) + } + operator.Main().Out().Bufferize() operator.Start() log.Print("started as httpPost") diff --git a/pkg/core/port.go b/pkg/core/port.go index c0669eba..babd15be 100644 --- a/pkg/core/port.go +++ b/pkg/core/port.go @@ -740,6 +740,29 @@ func (p *Port) Bufferize() { } } +// Check port has a `src` +// What is strange though for an operator with a connected delegate this would +// return an error. For instance if you take `controlIterate` and connect the delegate. +// And nothing more it would still work when you `push` and `pull` on `controlIterateId` +func (p *Port) FullyConnected() error { + if p.PrimitiveType() { + if p.src == nil && p.strSrc.src == nil { + return fmt.Errorf("%v is not connected to a source", p.Name()) + } else { + return nil + } + } else if p.itemType == TYPE_MAP { + for _, sub := range p.subs { + if err := sub.FullyConnected(); err != nil { + return err + } + } + } else if p.itemType == TYPE_STREAM { + return p.sub.FullyConnected() + } + return nil +} + // PRIVATE METHODS func setParentStreams(p *Port, parent *Port) { diff --git a/pkg/elem/control_iterate_test.go b/pkg/elem/control_iterate_test.go index 877f0a89..47146345 100644 --- a/pkg/elem/control_iterate_test.go +++ b/pkg/elem/control_iterate_test.go @@ -132,6 +132,7 @@ func Test_CtrlIterate__SimpleAggregation(t *testing.T) { // Connect require.NoError(t, ao.Delegate("iterator").Out().Connect(fo.Main().In())) require.NoError(t, fo.Main().Out().Connect(ao.Delegate("iterator").In())) + require.NoError(t, ao.Main().Out().FullyConnected()) ao.Main().Out().Bufferize()