Scrapped msgpack task message serialization option.

Because for both msgpack PHP extension or `rybakit/msgpack` Composer
package there seem to be unsolvable problems with differentiating of
how to serialize empty lists vs empty dicts (PHP only has arrays
for both use cases).

.gitignore

@@ -6,5 +6,8 @@
 tests/**/output/**
 !tests/**/output/.gitkeep
 
+# Tests - Python compiled bytecode cache.
+*.pyc
+
 # Installed dependencies.
 vendor

README.md

@@ -5,7 +5,7 @@
 A modern PHP client library for [Celery - Distributed Task Queue](https://docs.celeryq.dev).
 
 ## Requirements
-- PHP 8.0+
+- PHP 8.1+
 
 ## Example
 
@@ -36,7 +36,9 @@ $task = new TaskSignature(
 // Send the task into Celery.
 $asyncResult = $celery->sendTask($task);
 
-// Wait for the result and retrieve it.
+// Wait for the result (up to 10 seconds by default) and return it.
+// Alternatively a \Smuuf\CeleryForPhp\Exc\CeleryTimeoutException exception will
+// be thrown if the task won't finish in time.
 $result = $asyncResult->get();
 // $result === 9
 ```

bin/tests.sh

@@ -4,13 +4,18 @@ set -e
 
 cd $(dirname $0)/..
 
+TEST_PATH="${1:-./tests/suite}"
+
+# Test configuration
+export CELERYFORPHP_TASK_SERIALIZER='json'
+export CELERYFORPHP_TASK_MESSAGE_PROTOCOL_VERSION=2
+
 function _info {
 	echo -e "▄"
 	echo -e "█ $@"
 	echo -e "▀"
 }
 
-
 function _compose {
 	docker compose -f ./tests/infra/docker/test-services.yml $@
 }
@@ -21,12 +26,17 @@ _compose up \
 	--quiet-pull \
 	--build
 
+_info "Tests will use this configuration:"
+echo CELERYFORPHP_TASK_SERIALIZER=$CELERYFORPHP_TASK_SERIALIZER
+echo CELERYFORPHP_TASK_MESSAGE_PROTOCOL_VERSION=$CELERYFORPHP_TASK_MESSAGE_PROTOCOL_VERSION
+
 _info "Running tests"
 php ./vendor/nette/tester/src/tester \
+		-C `# Use system-wide php-ini` \
 		--coverage ./tests/output/coverage.html \
 		--coverage-src ./src \
 		-p phpdbg \
-		./tests/suite;
+		$TEST_PATH;
 
 _info "Stopping test infrastructure services"
 _compose down --timeout 2

composer.json

@@ -28,6 +28,6 @@
 		"predis/predis": "^2"
 	},
 	"suggest": {
-		"predis/predis": "Adds support for the predis, PHP client library for Redis, backend"
+		"predis/predis": "Adds support for Predis - a PHP library providing a Redis result backend"
 	}
 }

src/Config.php

@@ -25,15 +25,15 @@ public static function createDefault(): self {
 	 *     Name of the Celery control exchange (used for sending control commands,
 	 *     such as 'revoke', to Celery workers).
 	 *     Default value is 'celery' but can be specified to some other name.
-	 * @param int $messageProtocolVersion
-	 *     Celery message protocol version to use.
+	 * @param int $taskMessageProtocolVersion
+	 *     Celery task message protocol version to use.
 	 *     See https://docs.celeryq.dev/en/stable/internals/protocol.html
 	 * @param ISerializer $taskSerializer
 	 *     Serializer to use when serializing task message body. If not
 	 *     specified, `JsonSerializer` is used by default.
 	 */
 	public function __construct(
-		private int $messageProtocolVersion = MessageBuilder::MESSAGE_PROTOCOL_V2,
+		private int $taskMessageProtocolVersion = MessageBuilder::MESSAGE_PROTOCOL_V2,
 		private ?ISerializer $taskSerializer = null,
 		private ?ITaskIdFactory $taskIdFactory = null,
 		private string $controlExchangeName = 'celery',
@@ -46,8 +46,8 @@ public function __construct(
 			throw new InvalidArgumentException('Control exchange name must be a non-empty string');
 		}
 
-		if (!in_array($this->messageProtocolVersion, MessageBuilder::VALID_MESSAGE_PROTOCOL_VERSIONS, true)) {
-			throw new InvalidArgumentException("Invalid message protocol version '{$this->messageProtocolVersion}'");
+		if (!in_array($this->taskMessageProtocolVersion, MessageBuilder::VALID_MESSAGE_PROTOCOL_VERSIONS, true)) {
+			throw new InvalidArgumentException("Invalid message protocol version '{$this->taskMessageProtocolVersion}'");
 		}
 
 	}
@@ -61,7 +61,7 @@ public function getControlExchangeName(): string {
 	}
 
 	public function getTaskProtocolVersion(): int {
-		return $this->messageProtocolVersion;
+		return $this->taskMessageProtocolVersion;
 	}
 
 	public function getTaskSerializer(): ISerializer {

tests/bootstrap.php

@@ -1,21 +1,53 @@
 <?php
 
+use Predis\Client as PredisClient;
+
+use Smuuf\CeleryForPhp\Celery;
+use Smuuf\CeleryForPhp\Config;
+use Smuuf\CeleryForPhp\Brokers\RedisBroker;
+use Smuuf\CeleryForPhp\Drivers\PredisDriver;
+use Smuuf\CeleryForPhp\Backends\RedisBackend;
+use Smuuf\CeleryForPhp\Serializers\JsonSerializer;
+
 require __DIR__ . '/../vendor/autoload.php';
 
 \Tester\Environment::setup();
 
-class TestEnv {
+class CeleryFactory {
+
+	public static function getCelery(): Celery {
+
+		$envConfig = self::readEnv();
+
+		$config = new Config(
+			taskMessageProtocolVersion: $envConfig['task_message_protocol_version'],
+			taskSerializer: match ($envConfig['serializer']) {
+				'json' => new JsonSerializer(),
+			},
+		);
 
-	private static string $redisUri;
+		$predis = new PredisClient(['host' => '127.0.0.1']);
+		$redisDriver = new PredisDriver($predis);
+		$broker = new RedisBroker($redisDriver);
+		$backend = new RedisBackend($redisDriver);
+
+		return new Celery($broker, $backend, $config);
 
-	public static function init(): void {
-		self::$redisUri = '127.0.0.1';
 	}
 
-	public static function getRedisUri(): string {
-		return self::$redisUri;
+	/**
+	 * @return array{
+	 *     serializer: string,
+	 *     message_protocol: int,
+	 * }
+	 */
+	private static function readEnv(): array {
+
+		return [
+			'serializer' => getenv('CELERYFORPHP_TASK_SERIALIZER') ?: '',
+			'task_message_protocol_version' => (int) getenv('CELERYFORPHP_TASK_MESSAGE_PROTOCOL_VERSION'),
+		];
+
 	}
 
 }
-
-TestEnv::init();

tests/infra/celery-app/Dockerfile

@@ -3,10 +3,10 @@ FROM python:3.11-alpine
 RUN adduser my_user --disabled-password
 USER my_user
 
-RUN pip install --user celery "celery[redis,msgpack]"
+RUN pip install --user celery "celery[redis]"
 
-RUN mkdir ~/celery-app
-COPY main.py start.sh ~/celery-app/
+RUN mkdir /home/my_user/celery-app
+COPY main.py start.sh /home/my_user/celery-app/
 
-WORKDIR ~/celery-app
+WORKDIR /home/my_user/celery-app
 ENTRYPOINT ["sh", "./start.sh"]

tests/infra/celery-app/main.py

@@ -6,12 +6,10 @@
 
 CONNECTION = 'redis://127.0.0.1'
 BROKER = CONNECTION
-#BROKER = 'amqp://myuser:mypassword@localhost:5672/myvhost'
 
 
 class Config:
     task_serializer = 'json'
-    #task_serializer = 'msgpack'
 
 
 number = Union[int, float]

tests/suite/Backends.RedisBackend.phpt

@@ -12,7 +12,7 @@ use Smuuf\CeleryForPhp\Drivers\PredisDriver;
 
 require __DIR__ . '/../bootstrap.php';
 
-$predis = new PredisClient(['host' => TestEnv::getRedisUri()]);
+$predis = new PredisClient(['host' => '127.0.0.1']);
 $redisDriver = new PredisDriver($predis);
 
 $x = serialize($predis);

tests/suite/Brokers.RedisBroker.phpt

@@ -4,9 +4,8 @@ use Tester\Assert;
 
 use Predis\Client as PredisClient;
 
-use Smuuf\CeleryForPhp\TaskSignature;
-use Smuuf\CeleryForPhp\Brokers\RedisBroker;
 use Smuuf\CeleryForPhp\DeliveryInfo;
+use Smuuf\CeleryForPhp\Brokers\RedisBroker;
 use Smuuf\CeleryForPhp\Drivers\PredisDriver;
 use Smuuf\CeleryForPhp\Messaging\CeleryMessage;
 use Smuuf\CeleryForPhp\Serializers\JsonSerializer;
@@ -24,7 +23,7 @@ function _prepare_random_queue(): string {
 
 }
 
-$predis = new PredisClient(['host' => TestEnv::getRedisUri()]);
+$predis = new PredisClient(['host' => '127.0.0.1']);
 $redisDriver = new PredisDriver($predis);
 $broker = new RedisBroker($redisDriver);
 

tests/suite/Integration/RealTask.countdown.phpt

@@ -1,26 +1,15 @@
 <?php
 
 use Tester\Assert;
-use Predis\Client as PredisClient;
 
-use Smuuf\CeleryForPhp\Celery;
 use Smuuf\CeleryForPhp\AsyncResult;
 use Smuuf\CeleryForPhp\TaskSignature;
-use Smuuf\CeleryForPhp\Backends\RedisBackend;
-use Smuuf\CeleryForPhp\Brokers\RedisBroker;
-use Smuuf\CeleryForPhp\Drivers\PredisDriver;
 use Smuuf\CeleryForPhp\Exc\InvalidArgumentException;
 use Smuuf\CeleryForPhp\State;
 
 require __DIR__ . '/../../bootstrap.php';
 
-$predis = new PredisClient(['host' => TestEnv::getRedisUri()]);
-$redisDriver = new PredisDriver($predis);
-
-$c = new Celery(
-	new RedisBroker($redisDriver),
-	new RedisBackend($redisDriver),
-);
+$c = CeleryFactory::getCelery();
 
 // Call real-life Python Celery's task.
 $ts = new TaskSignature('main.add');

tests/suite/Integration/RealTask.eta.phpt

@@ -14,7 +14,7 @@ use Smuuf\CeleryForPhp\State;
 
 require __DIR__ . '/../../bootstrap.php';
 
-$predis = new PredisClient(['host' => TestEnv::getRedisUri()]);
+$predis = new PredisClient(['host' => '127.0.0.1']);
 $redisDriver = new PredisDriver($predis);
 
 $c = new Celery(
@@ -49,13 +49,13 @@ function test_task_with_eta($eta, int $wait): void {
 
 }
 
-test_task_with_eta('now + 3 seconds', 6);
+test_task_with_eta('now + 3 seconds', 9);
 test_task_with_eta(new \DateTime('now + 2 seconds'), 6);
 
 Assert::exception(function() {
-	test_task_with_eta('just some garbage', 1);
+	test_task_with_eta('just some garbage', 3);
 }, InvalidArgumentException::class, '#cannot convert#i');
 
 Assert::exception(function() {
-	test_task_with_eta(['wtf lol'], 1);
+	test_task_with_eta(['wtf lol'], 3);
 }, \TypeError::class);

tests/suite/Integration/RealTask.task.justWait.phpt

@@ -1,25 +1,14 @@
 <?php
 
 use Tester\Assert;
-use Predis\Client as PredisClient;
 
-use Smuuf\CeleryForPhp\Celery;
+use Smuuf\CeleryForPhp\State;
 use Smuuf\CeleryForPhp\TaskSignature;
 use Smuuf\CeleryForPhp\Exc\CeleryTimeoutException;
-use Smuuf\CeleryForPhp\Backends\RedisBackend;
-use Smuuf\CeleryForPhp\Brokers\RedisBroker;
-use Smuuf\CeleryForPhp\Drivers\PredisDriver;
-use Smuuf\CeleryForPhp\State;
 
 require __DIR__ . '/../../bootstrap.php';
 
-$predis = new PredisClient(['host' => TestEnv::getRedisUri()]);
-$redisDriver = new PredisDriver($predis);
-
-$c = new Celery(
-	new RedisBroker($redisDriver),
-	new RedisBackend($redisDriver),
-);
+$c = CeleryFactory::getCelery();
 
 // Call real-life Python Celery's task.
 $ts = new TaskSignature('main.just_wait');

tests/suite/Integration/RealTask.task.justWait.revoke.phpt

@@ -14,7 +14,7 @@ use Smuuf\CeleryForPhp\Helpers\Signals;
 
 require __DIR__ . '/../../bootstrap.php';
 
-$predis = new PredisClient(['host' => TestEnv::getRedisUri()]);
+$predis = new PredisClient(['host' => '127.0.0.1']);
 $redisDriver = new PredisDriver($predis);
 
 $c = new Celery(