Updated dependencies and fixed State class

Author: sebprt

composer.json

@@ -32,19 +32,17 @@
   "prefer-stable": true,
   "require": {
     "php": "^8.0",
-    "psr/log": "^1.1@dev",
     "php-etl/pipeline-contracts": "^0.3.0",
-    "php-etl/bucket": "^0.2.0",
-    "bunny/bunny": "@dev"
+    "php-etl/bucket": "dev-feature/update-bucket-contracts as 0.2.0",
+    "bunny/bunny": "^0.5.0"
   },
   "require-dev": {
     "phpunit/phpunit": "^9.0",
     "phpunit/php-invoker": "*",
     "johnkary/phpunit-speedtrap": "*",
     "mybuilder/phpunit-accelerator": "*",
     "infection/infection": "^0.20",
-    "adlawson/vfs": "dev-develop",
-    "php-etl/phpunit-extension": "*"
+    "adlawson/vfs": "dev-develop"
   },
   "extra": {
     "branch-alias": {

composer.lock

@@ -13,6 +13,7 @@ final class State implements StateInterface
 
     public function __construct(
         private Client $connection,
+        private string $pipelineId,
         private string $stepCode,
         private string $stepLabel,
         private string $topic,
@@ -22,6 +23,7 @@ public function __construct(
     }
 
     public static function withoutAuthentication(
+        string $pipelineId,
         string $stepCode,
         string $stepLabel,
         string $host,
@@ -39,10 +41,11 @@ public static function withoutAuthentication(
         ]);
         $connection->connect();
 
-        return new self($connection, stepCode: $stepCode, stepLabel: $stepLabel, topic: $topic, exchange: $exchange);
+        return new self($connection, pipelineId: $pipelineId, stepCode: $stepCode, stepLabel: $stepLabel, topic: $topic, exchange: $exchange);
     }
 
     public static function withAuthentication(
+        string $pipelineId,
         string $stepCode,
         string $stepLabel,
         string $host,
@@ -62,14 +65,15 @@ public static function withAuthentication(
         ]);
         $connection->connect();
 
-        return new self($connection, stepCode: $stepCode, stepLabel: $stepLabel, topic: $topic, exchange: $exchange);
+        return new self($connection, pipelineId: $pipelineId, stepCode: $stepCode, stepLabel: $stepLabel, topic: $topic, exchange: $exchange);
     }
 
     public function initialize(int $start = 0): void
     {
         $this->metrics = [
             'accept' => 0,
             'reject' => 0,
+            'error' => 0,
         ];
 
         $this->channel->queueDeclare(
@@ -97,7 +101,9 @@ public function reject(int $step = 1): void
 
     public function error(int $step = 1): void
     {
-        // Not implemented yet
+        $this->metrics['error'] += $step;
+
+        $this->sendMessage('error');
     }
 
     public function teardown(): void
@@ -112,7 +118,7 @@ private function sendMessage(string $metricCode): void
 
         $this->channel->publish(
             \json_encode([
-                'id' => '',
+                'id' => $this->pipelineId,
                 'date' => ['date' => $date->format('c'), 'tz' => $date->getTimezone()->getName()],
                 'stepsUpdates' => [
                     [
@@ -121,7 +127,7 @@ private function sendMessage(string $metricCode): void
                         'metrics' => [
                             [
                                 'code' => $metricCode,
-                                'increment' => $step
+                                'increment' => $this->metrics[$metricCode]
                             ]
                         ]
                     ]