Add beanstalkd port to docker-compose

bin/patch-type-declarations

@@ -0,0 +1,119 @@
+#!/usr/bin/env php
+<?php
+
+/**
+ * Proxy PHP file generated by Composer
+ *
+ * This file includes the referenced bin path (../vendor/symfony/error-handler/Resources/bin/patch-type-declarations)
+ * using a stream wrapper to prevent the shebang from being output on PHP<8
+ *
+ * @generated
+ */
+
+namespace Composer;
+
+$GLOBALS['_composer_bin_dir'] = __DIR__;
+$GLOBALS['_composer_autoload_path'] = __DIR__ . '/..'.'/vendor/autoload.php';
+
+if (PHP_VERSION_ID < 80000) {
+    if (!class_exists('Composer\BinProxyWrapper')) {
+        /**
+         * @internal
+         */
+        final class BinProxyWrapper
+        {
+            private $handle;
+            private $position;
+            private $realpath;
+
+            public function stream_open($path, $mode, $options, &$opened_path)
+            {
+                // get rid of phpvfscomposer:// prefix for __FILE__ & __DIR__ resolution
+                $opened_path = substr($path, 17);
+                $this->realpath = realpath($opened_path) ?: $opened_path;
+                $opened_path = $this->realpath;
+                $this->handle = fopen($this->realpath, $mode);
+                $this->position = 0;
+
+                return (bool) $this->handle;
+            }
+
+            public function stream_read($count)
+            {
+                $data = fread($this->handle, $count);
+
+                if ($this->position === 0) {
+                    $data = preg_replace('{^#!.*\r?\n}', '', $data);
+                }
+
+                $this->position += strlen($data);
+
+                return $data;
+            }
+
+            public function stream_cast($castAs)
+            {
+                return $this->handle;
+            }
+
+            public function stream_close()
+            {
+                fclose($this->handle);
+            }
+
+            public function stream_lock($operation)
+            {
+                return $operation ? flock($this->handle, $operation) : true;
+            }
+
+            public function stream_seek($offset, $whence)
+            {
+                if (0 === fseek($this->handle, $offset, $whence)) {
+                    $this->position = ftell($this->handle);
+                    return true;
+                }
+
+                return false;
+            }
+
+            public function stream_tell()
+            {
+                return $this->position;
+            }
+
+            public function stream_eof()
+            {
+                return feof($this->handle);
+            }
+
+            public function stream_stat()
+            {
+                return array();
+            }
+
+            public function stream_set_option($option, $arg1, $arg2)
+            {
+                return true;
+            }
+
+            public function url_stat($path, $flags)
+            {
+                $path = substr($path, 17);
+                if (file_exists($path)) {
+                    return stat($path);
+                }
+
+                return false;
+            }
+        }
+    }
+
+    if (
+        (function_exists('stream_get_wrappers') && in_array('phpvfscomposer', stream_get_wrappers(), true))
+        || (function_exists('stream_wrapper_register') && stream_wrapper_register('phpvfscomposer', 'Composer\BinProxyWrapper'))
+    ) {
+        return include("phpvfscomposer://" . __DIR__ . '/..'.'/vendor/symfony/error-handler/Resources/bin/patch-type-declarations');
+    }
+}
+
+return include __DIR__ . '/..'.'/vendor/symfony/error-handler/Resources/bin/patch-type-declarations';

composer.json

@@ -89,6 +89,7 @@
       "Enqueue\\Gps\\": "pkg/gps/",
       "Enqueue\\JobQueue\\": "pkg/job-queue/",
       "Enqueue\\Mongodb\\": "pkg/mongodb/",
+      "Enqueue\\Natsjs\\": "pkg/natsjs/",
       "Enqueue\\Null\\": "pkg/null/",
       "Enqueue\\Pheanstalk\\": "pkg/pheanstalk/",
       "Enqueue\\RdKafka\\": "pkg/rdkafka/",

docker-compose.yml

@@ -72,6 +72,8 @@ services:
 
   beanstalkd:
       image: 'jonbaldie/beanstalkd'
+      ports:
+        - "11300:11300"
 
   gearmand:
       image: 'artefactual/gearmand'

pkg/natsjs/.gitattributes

@@ -0,0 +1,5 @@
+/Tests export-ignore
+.gitattributes export-ignore
+.gitignore export-ignore
+.travis.yml export-ignore
+phpunit.xml.dist export-ignore

pkg/natsjs/.github/workflows/ci.yml

@@ -0,0 +1,29 @@
+name: CI
+on:
+  pull_request:
+  push:
+    branches:
+      - master
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        php: ['8.1', '8.2', '8.3']
+
+    name: PHP ${{ matrix.php }} tests
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: shivammathur/setup-php@v2
+        with:
+          php-version: ${{ matrix.php }}
+          coverage: none
+
+      - uses: "ramsey/composer-install@v1"
+        with:
+          composer-options: "--prefer-source"
+
+      - run: vendor/bin/phpunit --exclude-group=functional

pkg/natsjs/LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+Copyright (c) 2017 Kotliar Maksym
+Copyright (c) 2024 g41797
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is furnished
+to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

pkg/natsjs/NatsjsConnectionFactory.php

@@ -0,0 +1,91 @@
+<?php
+
+/*
+    Copyright (c) 2024 g41797
+    SPDX-License-Identifier: MIT
+*/
+
+namespace Enqueue\Natsjs;
+
+use Basis\Nats\Client;
+use Basis\Nats\Configuration;
+use Interop\Queue\ConnectionFactory;
+use Interop\Queue\Context;
+
+class NatsjsConnectionFactory implements ConnectionFactory
+{
+    private Client $broker;
+    private array $config;
+
+    /**
+     * The config could be an array, string DSN or null. In case of null it will attempt to connect to localhost with default settings.
+     * see defaultConfig() and parseDsn()).
+     *
+     * @param array|string $config
+     */
+    public function __construct($config = 'nats:')
+    {
+        if (empty($config) || 'nats:' === $config) {
+            $config = [];
+        } elseif (is_string($config)) {
+            $config = $this->parseDsn($config);
+        } elseif (is_array($config)) {
+        } else {
+            throw new \LogicException('The config must be either an array of options, a DSN string or null');
+        }
+
+        $this->config = array_replace($this->defaultConfig(), $config);
+    }
+
+    public function createContext(): Context
+    {
+        return new NatsjsContext($this->createBroker());
+    }
+
+    private function defaultConfig(): array
+    {
+        return [
+            'scheme' => 'nats',
+            'host' => 'localhost',
+            'port' => 4222,
+            'user' => null,
+            'pass' => null,
+            'pedantic' => false,
+            'reconnect' => true,
+        ];
+    }
+
+    /**
+     * format of dns string for nats:
+     * 'nats://username:password@hostname:port'
+     * part of components may be omitted
+     * 'nats:'      connect to 'localhost:4022'
+     */
+    private function parseDsn(string $dsn): array
+    {
+        $dsnConfig = parse_url($dsn);
+
+        if (false === $dsnConfig) {
+            throw new \LogicException(sprintf('Failed to parse DSN "%s"', $dsn));
+        }
+
+        $dsnConfig = array_replace($this->defaultConfig(), $dsnConfig);
+
+        if ('nats' !== $dsnConfig['scheme']) {
+            throw new \LogicException(sprintf('The given DSN scheme "%s" is not supported. Could be "nats" only.', $dsnConfig['scheme']));
+        }
+
+        return $dsnConfig;
+    }
+
+    private function createBroker(): Client
+    {
+        if ($this->broker) {
+            return $this->broker;
+        }
+
+        $this->broker = new Client(new Configuration($this->config));
+
+        return $this->broker;
+    }
+}

pkg/natsjs/NatsjsContext.php

@@ -0,0 +1,79 @@
+<?php
+
+/*
+    Copyright (c) 2024 g41797
+    SPDX-License-Identifier: MIT
+*/
+
+namespace Enqueue\Natsjs;
+
+use Basis\Nats\Client;
+use Basis\Nats\Connection;
+use Interop\Queue\Consumer;
+use Interop\Queue\Context;
+use Interop\Queue\Destination;
+use Interop\Queue\Exception\SubscriptionConsumerNotSupportedException;
+use Interop\Queue\Exception\TemporaryQueueNotSupportedException;
+use Interop\Queue\Message;
+use Interop\Queue\Producer;
+use Interop\Queue\Queue;
+use Interop\Queue\SubscriptionConsumer;
+use Interop\Queue\Topic;
+
+class NatsjsContext implements Context
+{
+    private Client $broker;
+
+    public function __construct(Client $broker)
+    {
+        $this->broker = $broker;
+    }
+
+    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    {
+        return new NatsjsMessage($body, $properties, $headers);
+    }
+
+    public function createTopic(string $topicName): Topic
+    {
+        return new NatsjsDestination($topicName);
+    }
+
+    public function createQueue(string $queueName): Queue
+    {
+        return new NatsjsDestination($queueName);
+    }
+
+    public function createTemporaryQueue(): Queue
+    {
+        throw TemporaryQueueNotSupportedException::providerDoestNotSupportIt();
+    }
+
+    public function createProducer(): Producer
+    {
+        // TODO: Implement createProducer() method.
+    }
+
+    public function createConsumer(Destination $destination): Consumer
+    {
+        // TODO: Implement createConsumer() method.
+    }
+
+    public function createSubscriptionConsumer(): SubscriptionConsumer
+    {
+        throw SubscriptionConsumerNotSupportedException::providerDoestNotSupportIt();
+    }
+
+    public function purgeQueue(Queue $queue): void
+    {
+        $this->broker->getApi()->getStream($queue->getQueueName())->purge();
+    }
+
+    public function close(): void
+    {
+        if ($this->broker->ping()) {
+            $property = new \ReflectionProperty(Connection::class, 'socket');
+            fclose($property->getValue($this->broker->connection));
+        }
+    }
+}