Update consumers, tests stabilization

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

3 files changed, 34 insertions, 193 deletions

diff --git a/tests/plugins/jobs/configs/.rr-jobs-init.yaml b/tests/plugins/jobs/configs/.rr-jobs-init.yaml
index e31e4441..550e55cc 100644
--- a/tests/plugins/jobs/configs/.rr-jobs-init.yaml
+++ b/tests/plugins/jobs/configs/.rr-jobs-init.yaml
@@ -9,11 +9,16 @@ server:
 amqp:
   addr: amqp://guest:guest@localhost:5672/
 
-  # beanstalk configuration
+# beanstalk configuration
+#
 beanstalk:
+  # beanstalk address
   addr: tcp://localhost:11300
+  # connect timeout
+  timeout: 10s
 
-  # amazon sqs configuration
+# amazon sqs configuration
+# General section
 sqs:
   key: api-key
   secret: api-secret
@@ -42,32 +47,47 @@ jobs:
     test-local:
       driver: ephemeral
       priority: 10
-      pipeline_size: 10000
+      prefetch: 10000
 
     test-local-2:
       driver: ephemeral
       priority: 1
-      pipeline_size: 10000
+      prefetch: 10000
 
     test-local-3:
       driver: ephemeral
       priority: 2
-      pipeline_size: 10000
+      prefetch: 10000
 
     test-1:
       driver: amqp
-      priority: 1
-      pipeline_size: 1000000
+      # QoS
+      prefetch: 1000000
+      # Queue name
       queue: test-1-queue
+      # Pipeline jobs priority, 1 - highest
+      priority: 1
+      # Exchange
       exchange: default
-      exclusive: false
+      # Exchange type: direct, topic, fanout
       exchange_type: direct
+      # Routing key for the queue
       routing_key: test
+      # Declare a queue exclusive at the exchange
+      exclusive: false
+      # When multiple is true, this delivery and all prior unacknowledged deliveries
+      # on the same channel will be acknowledged.  This is useful for batch processing
+      # of deliveries
+      multiple_ack: false
+      # When multiple is true, this delivery and all prior unacknowledged deliveries
+      # on the same channel will be acknowledged.  This is useful for batch processing
+      # of deliveries
+      requeue_on_fail: false
 
     test-2-amqp:
       driver: amqp
       priority: 2
-      pipeline_size: 100000
+      prefetch: 1000000
       queue: test-2-queue
       exchange: default
       exchange_type: direct
@@ -77,11 +97,10 @@ jobs:
       driver: beanstalk
       priority: 11
       tube: default
-      pipeline_size: 1000000
 
     test-3:
       driver: sqs
-      pipeline_size: 1000000
+      prefetch: 1000000
       queue: default
       attributes:
         MessageRetentionPeriod: 86400
diff --git a/tests/plugins/jobs/configs/.rr-jobs-test.yaml b/tests/plugins/jobs/configs/.rr-jobs-test.yaml
deleted file mode 100644
index 8213d72a..00000000
--- a/tests/plugins/jobs/configs/.rr-jobs-test.yaml
+++ /dev/null
@@ -1,106 +0,0 @@
-rpc:
-  listen: unix:///tmp/rr.sock
-
-server:
-  command: "php ../../client.php echo pipes"
-  relay: "pipes"
-  relay_timeout: "20s"
-
-amqp:
-  addr: amqp://guest:guest@localhost:5672/
-
-  # beanstalk configuration
-beanstalk:
-  addr: tcp://localhost:11300
-
-  # amazon sqs configuration
-sqs:
-  key: api-key
-  secret: api-secret
-  region: us-west-1
-  endpoint: http://localhost:9324
-  session_token: ""
-  ping_period: 10
-  attributes:
-    MessageRetentionPeriod: 86400
-
-logs:
-  level: info
-  encoding: console
-  mode: development
-
-jobs:
-  # num logical cores by default
-  num_pollers: 64
-  # 1mi by default
-  pipeline_size: 100000
-  # worker pool configuration
-  pool:
-    num_workers: 20
-    max_jobs: 0
-    allocate_timeout: 60s
-    destroy_timeout: 60s
-
-  # list of broker pipelines associated with endpoints
-  pipelines:
-#    test-local:
-#      driver: ephemeral
-#      priority: 10
-#      pipeline_size: 10000
-#
-#    test-local-2:
-#      driver: ephemeral
-#      priority: 1
-#      pipeline_size: 10000
-#
-#    test-local-3:
-#      driver: ephemeral
-#      priority: 2
-#      pipeline_size: 10000
-#
-#    test-1:
-#      driver: amqp
-#      priority: 1
-#      pipeline_size: 1000000
-#      queue: test-1-queue
-#      exchange: default
-#      exchange_type: direct
-#      routing_key: test
-#
-#    test-4:
-#      driver: amqp
-#      priority: 1
-#      pipeline_size: 1000000
-#      queue: test-1-queue
-#      exchange: default
-#      exchange_type: direct
-#      routing_key: test
-#
-#    test-2-amqp:
-#      driver: amqp
-#      priority: 2
-#      pipeline_size: 100000
-#      queue: test-2-queue
-#      exchange: default
-#      exchange_type: direct
-#      routing_key: test-2
-
-    test-2:
-      driver: beanstalk
-      priority: 11
-      tube: default
-      pipeline_size: 1000000
-      reserve_timeout: 10s
-
-#    test-3:
-#      driver: sqs
-#      pipeline_size: 1000000
-#      queue: default
-#      attributes:
-#        MessageRetentionPeriod: 86400
-#      tags:
-#        test: "tag"
-
-  # list of pipelines to be consumed by the server, keep empty if you want to start consuming manually
-  consume: [ "test-local", "test-local-2", "test-local-3", "test-1", "test-2-amqp", "test-3", "test-2" ]
-
diff --git a/tests/plugins/jobs/jobs_plugin_test.go b/tests/plugins/jobs/jobs_plugin_test.go
index d7c1075f..c10ac350 100644
--- a/tests/plugins/jobs/jobs_plugin_test.go
+++ b/tests/plugins/jobs/jobs_plugin_test.go
@@ -17,9 +17,7 @@ import (
 	"github.com/spiral/roadrunner/v2/plugins/informer"
 	"github.com/spiral/roadrunner/v2/plugins/jobs"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/drivers/amqp"
-	"github.com/spiral/roadrunner/v2/plugins/jobs/drivers/beanstalk"
 	"github.com/spiral/roadrunner/v2/plugins/jobs/drivers/ephemeral"
-	"github.com/spiral/roadrunner/v2/plugins/jobs/drivers/sqs"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
 	"github.com/spiral/roadrunner/v2/plugins/resetter"
 	rpcPlugin "github.com/spiral/roadrunner/v2/plugins/rpc"
@@ -30,80 +28,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestTEMP_INTI(t *testing.T) {
-	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel))
-	assert.NoError(t, err)
-
-	cfg := &config.Viper{
-		Path:   "configs/.rr-jobs-test.yaml",
-		Prefix: "rr",
-	}
-
-	err = cont.RegisterAll(
-		cfg,
-		&server.Plugin{},
-		&rpcPlugin.Plugin{},
-		&logger.ZapLogger{},
-		&jobs.Plugin{},
-		&resetter.Plugin{},
-		&informer.Plugin{},
-		&ephemeral.Plugin{},
-		&sqs.Plugin{},
-		&amqp.Plugin{},
-		&beanstalk.Plugin{},
-	)
-	assert.NoError(t, err)
-
-	err = cont.Init()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	ch, err := cont.Serve()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	sig := make(chan os.Signal, 1)
-	signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
-
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	stopCh := make(chan struct{}, 1)
-
-	go func() {
-		defer wg.Done()
-		for {
-			select {
-			case e := <-ch:
-				assert.Fail(t, "error", e.Error.Error())
-				err = cont.Stop()
-				if err != nil {
-					assert.FailNow(t, "error", err.Error())
-				}
-			case <-sig:
-				err = cont.Stop()
-				if err != nil {
-					assert.FailNow(t, "error", err.Error())
-				}
-				return
-			case <-stopCh:
-				// timeout
-				err = cont.Stop()
-				if err != nil {
-					assert.FailNow(t, "error", err.Error())
-				}
-				return
-			}
-		}
-	}()
-
-	time.Sleep(time.Second * 3)
-	stopCh <- struct{}{}
-	wg.Wait()
-}
-
 func TestJobsInit(t *testing.T) {
 	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel))
 	assert.NoError(t, err)
@@ -262,6 +186,10 @@ func TestJobsPauseResume(t *testing.T) {
 
 	mockLogger.EXPECT().Info("pipeline active", "pipeline", "test-local", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
 
+	// TODO delete
+	mockLogger.EXPECT().Debug("request", "body:", gomock.Any(), "context:", gomock.Any()).AnyTimes()
+	mockLogger.EXPECT().Debug("response", "body:", gomock.Any(), "context:", gomock.Any()).AnyTimes()
+
 	err = cont.RegisterAll(
 		cfg,
 		&server.Plugin{},