From f33a9dc226117587b095c9a12552140f03789047 Mon Sep 17 00:00:00 2001
From: Konstantin Grachev <me@grachevko.ru>
Date: Mon, 18 Jan 2021 16:46:31 +0300
Subject: [PATCH] Initial commit

---
 .gitignore                               |   2 +
 LICENSE                                  |  21 +++
 composer.json                            |  29 +++++
 src/DependencyInjection/NsqExtension.php |  20 +++
 src/Messenger/NsqReceivedStamp.php       |  21 +++
 src/Messenger/NsqTransport.php           | 155 +++++++++++++++++++++++
 src/Messenger/NsqTransportFactory.php    |  52 ++++++++
 src/NsqBundle.php                        |  10 ++
 src/Resources/config/services.yml        |   3 +
 9 files changed, 313 insertions(+)
 create mode 100644 .gitignore
 create mode 100644 LICENSE
 create mode 100644 composer.json
 create mode 100644 src/DependencyInjection/NsqExtension.php
 create mode 100644 src/Messenger/NsqReceivedStamp.php
 create mode 100644 src/Messenger/NsqTransport.php
 create mode 100644 src/Messenger/NsqTransportFactory.php
 create mode 100644 src/NsqBundle.php
 create mode 100644 src/Resources/config/services.yml

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..4fbb073
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,2 @@
+/vendor/
+/composer.lock
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..0a55468
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2021 nsqphp
+
+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.
diff --git a/composer.json b/composer.json
new file mode 100644
index 0000000..0bfb952
--- /dev/null
+++ b/composer.json
@@ -0,0 +1,29 @@
+{
+    "name": "nsq/nsq-bundle",
+    "type": "library",
+    "description": "Symfony Integration for NsqPHP",
+    "homepage": "https://github.com/nsqphp/NsqBundle",
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Konstantin Grachev",
+            "email": "me@grachevko.ru"
+        }
+    ],
+    "require": {
+        "php": ">=7.4",
+        "ext-json": "*",
+        "nsq/nsq": "0.1",
+        "symfony/framework-bundle": "^5.0",
+        "symfony/messenger": "^5.0"
+    },
+    "require-dev": {
+        "ergebnis/composer-normalize": "^2.13"
+    },
+    "autoload": {
+        "psr-4": {
+            "NsqPHP\\NsqBundle\\": "src/"
+        }
+    },
+    "minimum-stability": "dev"
+}
diff --git a/src/DependencyInjection/NsqExtension.php b/src/DependencyInjection/NsqExtension.php
new file mode 100644
index 0000000..fc070bd
--- /dev/null
+++ b/src/DependencyInjection/NsqExtension.php
@@ -0,0 +1,20 @@
+<?php
+
+namespace NsqPHP\NsqBundle\DependencyInjection;
+
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\Config\FileLocator;
+use Symfony\Component\HttpKernel\DependencyInjection\Extension;
+use Symfony\Component\DependencyInjection\Loader;
+
+class NsqExtension extends Extension
+{
+    /**
+     * {@inheritDoc}
+     */
+    public function load(array $configs, ContainerBuilder $container): void
+    {
+        $loader = new Loader\YamlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
+        $loader->load('services.yml');
+    }
+}
diff --git a/src/Messenger/NsqReceivedStamp.php b/src/Messenger/NsqReceivedStamp.php
new file mode 100644
index 0000000..7bb8224
--- /dev/null
+++ b/src/Messenger/NsqReceivedStamp.php
@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace NsqPHP\NsqBundle\Messenger;
+
+use Nsq\Envelope;
+use Symfony\Component\Messenger\Stamp\StampInterface;
+
+/**
+ * @psalm-immutable
+ */
+final class NsqReceivedStamp implements StampInterface
+{
+    public Envelope $envelope;
+
+    public function __construct(Envelope $envelope)
+    {
+        $this->envelope = $envelope;
+    }
+}
diff --git a/src/Messenger/NsqTransport.php b/src/Messenger/NsqTransport.php
new file mode 100644
index 0000000..79f240e
--- /dev/null
+++ b/src/Messenger/NsqTransport.php
@@ -0,0 +1,155 @@
+<?php
+
+declare(strict_types=1);
+
+namespace NsqPHP\NsqBundle\Messenger;
+
+use Generator;
+use JsonException;
+use LogicException;
+use Nsq\Connection;
+use Nsq\Envelope as NsqEnvelope;
+use Nsq\Reader;
+use Nsq\Subscriber;
+use Nsq\Writer;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
+use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
+use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+use function json_decode;
+use function json_encode;
+use const JSON_THROW_ON_ERROR;
+
+final class NsqTransport implements TransportInterface
+{
+    private Connection $connection;
+
+    private SerializerInterface $serializer;
+
+    private ?Generator $generator = null;
+
+    private string $topic;
+
+    private string $channel;
+
+    public function __construct(
+        Connection $connection,
+        string $topic,
+        string $channel,
+        SerializerInterface $serializer = null
+    ) {
+        $this->connection = $connection;
+        $this->topic = $topic;
+        $this->channel = $channel;
+        $this->serializer = $serializer ?? new PhpSerializer();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function send(Envelope $envelope): Envelope
+    {
+        $nsqEnvelope = $this->getNsqEnvelope($envelope);
+
+        $encodedMessage = $this->serializer->encode($envelope->withoutAll(NsqReceivedStamp::class));
+
+        $this->getPublisher()->pub($this->topic, json_encode($encodedMessage, JSON_THROW_ON_ERROR));
+
+        if (null !== $nsqEnvelope) {
+            $nsqEnvelope->ack();
+        }
+
+        return $envelope;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function get(): iterable
+    {
+        $generator = $this->generator;
+        if (null === $generator) {
+            $this->generator = $generator = $this->getSubscriber()->subscribe($this->topic, $this->channel);
+        } else {
+            $generator->next();
+        }
+
+        /** @var NsqEnvelope|null $nsqEnvelope */
+        $nsqEnvelope = $generator->current();
+
+        if (null === $nsqEnvelope) {
+            return [];
+        }
+
+        try {
+            $encodedEnvelope = json_decode($nsqEnvelope->message->body, true, 512, JSON_THROW_ON_ERROR);
+        } catch (JsonException $e) {
+            $nsqEnvelope->ack();
+
+            throw new MessageDecodingFailedException('', 0, $e);
+        }
+
+        try {
+            $envelope = $this->serializer->decode($encodedEnvelope);
+        } catch (MessageDecodingFailedException  $e) {
+            $nsqEnvelope->ack();
+
+            throw $e;
+        }
+
+        return [
+            $envelope->with(
+                new NsqReceivedStamp($nsqEnvelope),
+                new TransportMessageIdStamp($nsqEnvelope->message->id),
+            ),
+        ];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function ack(Envelope $envelope): void
+    {
+        $message = $this->getNsqEnvelope($envelope);
+        if (!$message instanceof NsqEnvelope) {
+            throw new LogicException('Returned envelop doesn\'t related to NsqMessage.');
+        }
+
+        $message->ack();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function reject(Envelope $envelope): void
+    {
+        $message = $this->getNsqEnvelope($envelope);
+        if (!$message instanceof NsqEnvelope) {
+            throw new LogicException('Returned envelop doesn\'t related to NsqMessage.');
+        }
+
+        $message->ack();
+    }
+
+    private function getNsqEnvelope(Envelope $envelope): ?NsqEnvelope
+    {
+        $stamp = $envelope->last(NsqReceivedStamp::class);
+        if (!$stamp instanceof NsqReceivedStamp) {
+            return null;
+        }
+
+        return $stamp->envelope;
+    }
+
+    private function getPublisher(): Writer
+    {
+        return $this->publisher ??= new Writer($this->connection);
+    }
+
+    private function getSubscriber(): Subscriber
+    {
+        return $this->publisher ??= new Subscriber(new Reader($this->connection));
+    }
+}
diff --git a/src/Messenger/NsqTransportFactory.php b/src/Messenger/NsqTransportFactory.php
new file mode 100644
index 0000000..ab84134
--- /dev/null
+++ b/src/Messenger/NsqTransportFactory.php
@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace NsqPHP\NsqBundle\Messenger;
+
+use Nsq\Config;
+use Nsq\Connection;
+use Symfony\Component\Messenger\Exception\InvalidArgumentException;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+use function dump;
+use function parse_str;
+use function parse_url;
+use function sprintf;
+use function strpos;
+
+final class NsqTransportFactory implements TransportFactoryInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
+    {
+        if (false === $parsedUrl = parse_url($dsn)) {
+            throw new InvalidArgumentException(sprintf('The given Nsq DSN "%s" is invalid.', $dsn));
+        }
+
+        $nsqOptions = [];
+        if (isset($parsedUrl['query'])) {
+            parse_str($parsedUrl['query'], $nsqOptions);
+        }
+
+        $address = sprintf('tcp://%s:%s', $parsedUrl['host'] ?? 'nsqd', $parsedUrl['port'] ?? 4150);
+
+        return new NsqTransport(
+            Connection::connect(new Config($address)),
+            $nsqOptions['topic'] ?? 'symfony-messenger',
+            $nsqOptions['channel'] ?? 'default',
+            $serializer
+        );
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function supports(string $dsn, array $options): bool
+    {
+        return 0 === strpos($dsn, 'nsq://');
+    }
+}
diff --git a/src/NsqBundle.php b/src/NsqBundle.php
new file mode 100644
index 0000000..908988f
--- /dev/null
+++ b/src/NsqBundle.php
@@ -0,0 +1,10 @@
+<?php
+declare(strict_types=1);
+
+namespace NsqPHP\NsqBundle;
+
+use Symfony\Component\HttpKernel\Bundle\Bundle;
+
+final class NsqBundle extends Bundle
+{
+}
diff --git a/src/Resources/config/services.yml b/src/Resources/config/services.yml
new file mode 100644
index 0000000..f9fdb68
--- /dev/null
+++ b/src/Resources/config/services.yml
@@ -0,0 +1,3 @@
+services:
+    NsqPHP\NsqBundle\Messenger\NsqTransportFactory:
+        tags: [ 'messenger.transport_factory' ]