diff options
Diffstat (limited to 'plugins/jobs/plugin.go')
| -rw-r--r-- | plugins/jobs/plugin.go | 15 |
1 files changed, 10 insertions, 5 deletions
diff --git a/plugins/jobs/plugin.go b/plugins/jobs/plugin.go index 9d68a95a..2eb35f14 100644 --- a/plugins/jobs/plugin.go +++ b/plugins/jobs/plugin.go @@ -117,11 +117,6 @@ func (p *Plugin) Serve() chan error { //nolint:gocognit // pipeline name (ie test-local, sqs-aws, etc) name := key.(string) - // skip pipelines which are not initialized to consume - if _, ok := p.consume[name]; !ok { - return true - } - // pipeline associated with the name pipe := value.(*pipeline.Pipeline) // driver for the pipeline (ie amqp, ephemeral, etc) @@ -149,6 +144,16 @@ func (p *Plugin) Serve() chan error { //nolint:gocognit errCh <- errors.E(op, errors.Errorf("pipe register failed for the driver: %s with pipe name: %s", pipe.Driver(), pipe.Name())) return false } + + // if pipeline initialized to be consumed, call Consume on it + if _, ok := p.consume[name]; ok { + err = initializedDriver.Consume(pipe) + if err != nil { + errCh <- errors.E(op, err) + return false + } + return true + } } return true |