Implementation of retry mechanism

contributte · Jul 18, 2023 · 18f6fc8 · 18f6fc8
1 parent bddfa49
commit 18f6fc8
Show file tree

Hide file tree

Showing 4 changed files with 75 additions and 24 deletions.
diff --git a/src/DI/MessengerExtension.php b/src/DI/MessengerExtension.php
@@ -19,6 +19,7 @@
 use Nette\Schema\Expect;
 use Nette\Schema\Schema;
 use Nette\Utils\ArrayHash;
+use PhpOffice\PhpSpreadsheet\Calculation\MathTrig\Exp;
 use stdClass;
 
 /**
@@ -32,6 +33,7 @@ class MessengerExtension extends CompilerExtension
 	public const FAILURE_TRANSPORT_TAG = 'contributte.messenger.failure_transport';
 	public const BUS_TAG = 'contributte.messenger.bus';
 	public const HANDLER_TAG = 'contributte.messenger.handler';
+	public const RETRY_STRATEGY_TAG = 'contributte.messenger.retry_strategy';
 
 	/** @var AbstractPass[] */
 	protected array $passes = [];
@@ -101,6 +103,22 @@ public function getConfigSchema(): Schema
 					'options' => Expect::array(),
 					'serializer' => $expectService,
 					'failureTransport' => Expect::string(),
+                    'retryStrategy' => Expect::anyOf(
+						null,
+						Expect::structure([
+							'maxRetries' => Expect::int(),
+							'delay' => Expect::int(),
+							'multiplier' => Expect::anyOf(Expect::int(), Expect::float())->castTo('float'),
+							'maxDelay' => Expect::int(),
+							'service' => $expectService->nullable(),
+						]),
+					)->default(ArrayHash::from([
+						'maxRetries' => 3,
+						'delay' => 1000,
+						'multiplier' => 1,
+						'maxDelay' => 0,
+						'service' => null,
+					])),
 				]),
 				Expect::string()->required()
 			),

diff --git a/src/DI/Pass/EventPass.php b/src/DI/Pass/EventPass.php
@@ -33,11 +33,6 @@ public function loadPassConfiguration(): void
 		$builder->addDefinition($this->prefix('retry_strategy.container'))
 			->setFactory(NetteContainer::class)
 			->setAutowired(false);
-
-		// Register container for retry strategies
-		$builder->addDefinition($this->prefix('sender.container'))
-			->setFactory(NetteContainer::class)
-			->setAutowired(false);
 	}
 
 	/**
@@ -53,21 +48,7 @@ public function beforePassCompile(): void
 
 		/** @var ServiceDefinition $retryStrategyContainerDef */
 		$retryStrategyContainerDef = $builder->getDefinition($this->prefix('retry_strategy.container'));
-		$retryStrategyContainerDef->setArgument(0,
-			// TODO to be dynamic
-			[
-				'test' => ''
-			],
-		);
-
-		/** @var ServiceDefinition $sendersContainerDef */
-		$sendersContainerDef = $builder->getDefinition($this->prefix('sender.container'));
-		$sendersContainerDef->setArgument(0,
-			// TODO to be dynamic
-			[
-				'test' => ''
-			],
-		);
+		$retryStrategyContainerDef->setArgument(0, BuilderMan::of($this)->getRetryStrategies());
 
 		$dispatcherServiceName = $builder->getByType(EventDispatcherInterface::class);
 
@@ -99,16 +80,14 @@ private function getSubscribers(): array
 	{
 		$subscribers = [
 			new Statement(DispatchPcntlSignalListener::class),
-			/*
 			new Statement(
 				SendFailedMessageForRetryListener::class,
 				[
-					$this->prefix('@sender.container'),
+                    $this->prefix('@transport.container'),
 					$this->prefix('@retry_strategy.container'),
 					$this->prefix('@logger.logger'),
 				]
 			),
-			*/
 			new Statement(
 				SendFailedMessageToFailureTransportListener::class,
 				[
@@ -122,7 +101,7 @@ private function getSubscribers(): array
 		if (class_exists(StopWorkerOnSignalsListener::class)) {
 			$subscribers[] = new Statement(StopWorkerOnSignalsListener::class);
 		} else {
-			$subscribers[] = new Statement(StopWorkerOnSigtermSignalListener::class);
+			$subscribers[] = new Statement(StopWorkerOnSigtermSignalListener::class); // @phpstan-ignore-line
 		}
 
 		return $subscribers;

diff --git a/src/DI/Pass/TransportPass.php b/src/DI/Pass/TransportPass.php
@@ -6,7 +6,9 @@
 use Contributte\Messenger\DI\MessengerExtension;
 use Contributte\Messenger\DI\Utils\BuilderMan;
 use Contributte\Messenger\DI\Utils\ServiceMan;
+use Nette\DI\Definitions\Definition;
 use Nette\DI\Definitions\ServiceDefinition;
+use Symfony\Component\Messenger\Retry\MultiplierRetryStrategy;
 
 class TransportPass extends AbstractPass
 {
@@ -33,6 +35,8 @@ public function loadPassConfiguration(): void
 			if ($transport->failureTransport) {
 				$transportDef->addTag(MessengerExtension::FAILURE_TRANSPORT_TAG, $transport->failureTransport);
 			}
+
+            $this->registerRetryStrategyForTransport($transportDef, $transport->retryStrategy);
 		}
 
 		// Register transports container
@@ -53,4 +57,34 @@ public function beforePassCompile(): void
 		$transportContainerDef->setArgument(0, BuilderMan::of($this)->getTransports());
 	}
 
+    /**
+     * @param null|object{service: null|string, maxRetries: int, delay: int, multiplier: int, maxDelay: int} $strategy
+     */
+    private function registerRetryStrategyForTransport(Definition $transportDev, null|object $strategy): void
+    {
+        if ($strategy === null) {
+            return;
+        }
+
+        $builder = $this->getContainerBuilder();
+
+        // Already prefixed, do not need to prefix again
+        $strategyServiceName = sprintf('%s.retry_strategy', $transportDev->getName());
+        $strategyDefinition = $builder->addDefinition($strategyServiceName);
+
+        $transportDev->addTag(MessengerExtension::RETRY_STRATEGY_TAG, $strategyServiceName);
+
+        if ($strategy->service !== null) {
+            $strategyDefinition->setFactory($strategy->service);
+
+            return;
+        }
+
+        $strategyDefinition->setFactory(MultiplierRetryStrategy::class, [
+            $strategy->maxRetries,
+            $strategy->delay,
+            (float) $strategy->multiplier,
+            $strategy->maxDelay,
+        ]);
+    }
 }
diff --git a/src/DI/Utils/BuilderMan.php b/src/DI/Utils/BuilderMan.php
@@ -106,4 +106,24 @@ public function getServiceNames(string $tag): array
 		return $definitions;
 	}
 
+    /**
+     * @return array<string, string>
+     */
+    public function getRetryStrategies(): array
+    {
+        $builder = $this->pass->getContainerBuilder();
+		$definitions = $this->getServiceDefinitions(MessengerExtension::TRANSPORT_TAG);
+
+		$strategies = [];
+		foreach ($definitions as $transport) {
+			$transportName = $transport->getTag(MessengerExtension::TRANSPORT_TAG);
+			$retryService = $transport->getTag(MessengerExtension::RETRY_STRATEGY_TAG);
+
+			if ($retryService !== null) {
+				$strategies[$transportName] = $retryService;
+			}
+		}
+
+		return $strategies;
+    }
 }