improvements

Author: thorewi

README.md

@@ -104,7 +104,7 @@ Ve všech ostatních případech se záznam uloží jako úspěšně dokončený
 
 ### 2.2 Commandy
 
-`background-queue:process` Zpracuje všechny záznamy ve stavu `READY`, `TEMPORARILY_FAILED`, `WAITING` a `AMQP_FAILED`, v případě využití AMQP brokera pouze záznamy ve stavu `TEMPORARILY_FAILED` a `WAITING`. Command je ideální spouštět cronem každou minutu. V případě použití AMQP brokeru je záznam ve stavu `TEMPORARILY_FAILED` a `WAITING` zařazen znovu do AMQP brokera.
+`background-queue:process` Zpracuje všechny záznamy ve stavu `READY`, `TEMPORARILY_FAILED` a `WAITING`, v případě využití AMQP brokera pouze záznamy ve stavu `TEMPORARILY_FAILED` a `WAITING`. Command je ideální spouštět cronem každou minutu. V případě použití AMQP brokeru je záznam ve stavu `TEMPORARILY_FAILED` a `WAITING` zařazen znovu do AMQP brokera.
 
 `background-queue:clear-finished` Smaže všechny úspěšně zpracované záznamy.
 

src/BackgroundQueue.php

@@ -121,7 +121,7 @@ public function publishToBroker(BackgroundJob $entity): void
 		} catch (Exception $e) {
 			$this->logException('Unexpected error occurred.', $entity, $e);
 
-			$entity->setState(BackgroundJob::STATE_AMQP_FAILED)
+			$entity->setProcessByBroker(false)
 				->setErrorMessage($e->getMessage());
 			$this->save($entity);
 		}
@@ -396,6 +396,9 @@ public function save(BackgroundJob $entity): void
 		$this->updateSchema();
 
 		if (!$entity->getId()) {
+			if ($this->producer) {
+				$entity->setProcessByBroker(true);
+			}
 			$this->connection->insert($this->config['tableName'], $entity->getDatabaseValues());
 			$entity->setId($this->connection->lastInsertId());
 		} else {
@@ -448,6 +451,7 @@ public function updateSchema(): void
 		$table->addColumn('identifier', Types::STRING)->setNotnull(false);
 		$table->addColumn('is_unique', Types::BOOLEAN)->setNotnull(true)->setDefault(0);
 		$table->addColumn('available_at', Types::DATETIME_IMMUTABLE)->setNotnull(false);
+		$table->addColumn('process_by_broker', Types::BOOLEAN)->setNotnull(true)->setDefault(0);
 
 		$table->setPrimaryKey(['id']);
 		$table->addIndex(['identifier']);

src/Console/ProcessCommand.php

@@ -44,7 +44,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 			if (
 				$this->backgroundQueue->getConfig()['producer']
 				&&
-				$_entity->getState() !== BackgroundJob::STATE_AMQP_FAILED
+				$_entity->getProcessByBroker()
 			) {
 				$_entity->setState(BackgroundJob::STATE_READY);
 				$this->backgroundQueue->save($_entity);

src/Entity/BackgroundJob.php

@@ -14,13 +14,11 @@ final class BackgroundJob
 	const STATE_PERMANENTLY_FAILED = 5; // kritická chyba (např. chyba v implementaci)
 	const STATE_WAITING = 6; // ceka na pristi zpracovani
 	const STATE_REDUNDANT = 7; // je nadbytecny (kdyz isUnique = true)
-	const STATE_AMQP_FAILED = 8; // nepodarilo se zpracovat pomoci AMQP
 
 	const READY_TO_PROCESS_STATES = [
 		self::STATE_READY => self::STATE_READY,
 		self::STATE_TEMPORARILY_FAILED => self::STATE_TEMPORARILY_FAILED,
 		self::STATE_WAITING => self::STATE_WAITING,
-		self::STATE_AMQP_FAILED => self::STATE_AMQP_FAILED
 	];
 
 	const FINISHED_STATES = [
@@ -41,6 +39,7 @@ final class BackgroundJob
 	private ?string $identifier = null;
 	private bool $isUnique = false;
 	private ?DateTimeImmutable $availableAt = null;
+	private bool $processByBroker = false;
 
 	public function __construct()
 	{
@@ -182,6 +181,17 @@ public function setAvailableAt(?DateTimeImmutable $availableAt): self
 		return $this;
 	}
 
+	public function getProcessByBroker(): bool
+	{
+		return $this->processByBroker;
+	}
+
+	public function setProcessByBroker(bool $processByBroker): self
+	{
+		$this->processByBroker = $processByBroker;
+		return $this;
+	}
+
 	public function isReadyForProcess(): bool
 	{
 		return isset(self::READY_TO_PROCESS_STATES[$this->state]);
@@ -206,7 +216,8 @@ public static function createEntity(array $values): BackgroundJob
 		$entity->identifier = $values['identifier'];
 		$entity->isUnique = $values['is_unique'];
 		$entity->availableAt = $values['available_at'] ? new DateTimeImmutable($values['available_at']) : null;
-		
+		$entity->processByBroker = $values['process_by_broker'];
+
 		return $entity;
 	}
 
@@ -225,6 +236,7 @@ public function getDatabaseValues(): array
 			'identifier' => $this->identifier,
 			'is_unique' => (int) $this->isUnique,
 			'available_at' => $this->availableAt ? $this->availableAt->format('Y-m-d H:i:s') : null,
+			'process_by_broker' => (int) $this->processByBroker,
 		];
 	}