AMQP job driver...

Update main driver's interface, add Consume(*pipeline) method. Implement
it on the amqp and ephemeral drivers.
Fix error with incorrect order of Register <-> Consume method calls.
Implement rabbitMQ driver, add timeouts, dead-letter-exchange,
packing-unpacking of the amqp messages.
Implement AMQP redialer in case of network error as well as channels
re-creation.
Update drawio diagram.
Update .rr.yaml jobs configuration, add all amqp options. Implement
Ack/Nack.

Signed-off-by: Valery Piashchynski <[email protected]>

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