Filesystem transport

bin/dev

@@ -21,7 +21,7 @@ while getopts "bustefc" OPTION; do
       ./bin/php-cs-fixer fix
       ;;
     t)
-      COMPOSE_PROJECT_NAME=mqdev docker-compose run --workdir="/mqdev" --rm dev ./bin/test
+      COMPOSE_PROJECT_NAME=mqdev docker-compose run --workdir="/mqdev" --rm dev ./bin/test "$2"
       ;;
     c)
       COMPOSE_PROJECT_NAME=mqdev docker-compose run --workdir="/mqdev" --rm generate-changelog github_changelog_generator --future-release "$2" --simple-list

bin/subtree-split

@@ -47,6 +47,7 @@ remote psr-queue [email protected]:php-enqueue/psr-queue.git
 remote enqueue [email protected]:php-enqueue/enqueue.git
 remote stomp [email protected]:php-enqueue/stomp.git
 remote amqp-ext [email protected]:php-enqueue/amqp-ext.git
+remote fs [email protected]:php-enqueue/fs.git
 remote enqueue-bundle [email protected]:php-enqueue/enqueue-bundle.git
 remote job-queue [email protected]:php-enqueue/job-queue.git
 remote test [email protected]:php-enqueue/test.git
@@ -55,6 +56,7 @@ split 'pkg/psr-queue' psr-queue
 split 'pkg/enqueue' enqueue
 split 'pkg/stomp' stomp
 split 'pkg/amqp-ext' amqp-ext
+split 'pkg/fs' fs
 split 'pkg/enqueue-bundle' enqueue-bundle
 split 'pkg/job-queue' job-queue
 split 'pkg/test' test

composer.json

@@ -8,6 +8,7 @@
     "enqueue/enqueue": "*@dev",
     "enqueue/stomp": "*@dev",
     "enqueue/amqp-ext": "*@dev",
+    "enqueue/fs": "*@dev",
     "enqueue/enqueue-bundle": "*@dev",
     "enqueue/job-queue": "*@dev",
     "enqueue/test": "*@dev"
@@ -51,6 +52,10 @@
     {
       "type": "path",
       "url": "pkg/job-queue"
+    },
+    {
+      "type": "path",
+      "url": "pkg/fs"
     }
   ]
 }

docs/bundle/config_reference.md

@@ -1,7 +1,8 @@
 # Config reference
 
+You can get this info by running `./bin/console config:dump-reference enqueue` command.
+
 ```yaml
-# Default configuration for extension with alias: "enqueue"
 enqueue:
     transport:            # Required
         default:
@@ -91,6 +92,16 @@ enqueue:
 
             # The option tells whether RabbitMQ broker has delay plugin installed or not
             delay_plugin_installed: false
+        fs:
+
+            # The store directory where all queue\topics files will be created and messages are stored
+            store_dir:            ~ # Required
+
+            # The option tells how many messages should be read from file at once. The feature save resources but could lead to bigger messages lose.
+            pre_fetch_count:      1
+
+            # The queue files are created with this given permissions if not exist.
+            chmod:                384
     client:
         traceable_producer:   false
         prefix:               enqueue

docs/client/supported_brokers.md

@@ -8,7 +8,7 @@ Here's the list of protocols and Client features supported by them
 | RabbitMQ AMQP  |   Yes    |    Yes*  |    Yes     |     Yes      |     Yes     |
 | STOMP          |   No     |    No    |    Yes     |     No       |     Yes**   |
 | RabbitMQ STOMP |   Yes    |    Yes*  |    Yes     |     Yes***   |     Yes**   |
-
+| Filesystem     |   No     |    No    |    No      |     Yes      |     No      |
 
 * \* Possible if a RabbitMQ delay plugin is installed.
 * \*\* Possible if topics (exchanges) are configured on broker side manually.

docs/filesystem_transport.md

@@ -0,0 +1,82 @@
+# Filesystem transport
+
+Use files on local filesystem as queues. 
+It creates a file per queue\topic. 
+A message is a line inside the file.
+**Limitations** It works only in auto ack mode. Local by nature therefor messages are not visible on other servers.  
+
+* [Create context](#create-context)
+* [Declare topic](#declare-topic)
+* [Declare queue](#decalre-queue)
+* [Bind queue to topic](#bind-queue-to-topic)
+* [Send message to topic](#send-message-to-topic)
+* [Send message to queue](#send-message-to-queue)
+* [Consume message](#consume-message)
+* [Purge queue messages](#purge-queue-messages)
+
+## Create context
+
+```php
+<?php
+use Enqueue\Fs\FsConnectionFactory;
+
+$connectionFactory = new FsConnectionFactory([
+    'store_dir' => '/tmp'
+]);
+
+$psrContext = $connectionFactory->createContext();
+```
+
+## Send message to topic
+
+```php
+<?php
+/** @var \Enqueue\Fs\FsContext $psrContext */
+
+$fooTopic = $psrContext->createTopic('aTopic');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooTopic, $message);
+```
+
+## Send message to queue 
+
+```php
+<?php
+/** @var \Enqueue\Fs\FsContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$message = $psrContext->createMessage('Hello world!');
+
+$psrContext->createProducer()->send($fooQueue, $message);
+```
+
+## Consume message:
+
+```php
+<?php
+/** @var \Enqueue\Fs\FsContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+$consumer = $psrContext->createConsumer($fooQueue);
+
+$message = $consumer->receive();
+
+// process a message
+
+$consumer->acknowledge($message);
+// $consumer->reject($message);
+```
+
+## Purge queue messages:
+
+```php
+<?php
+/** @var \Enqueue\Fs\FsContext $psrContext */
+
+$fooQueue = $psrContext->createQueue('aQueue');
+
+$psrContext->purge($fooQueue);
+```
+
+[back to index](index.md)

docs/index.md

@@ -2,9 +2,10 @@
 
 * [Quick tour](quick_tour.md)
 * Transports
-    - [AMQP](amqp_transport.md)
-    - [STOMP](stomp_transport.md)
-    - [NULL](null_transport.md)
+    - [Amqp](amqp_transport.md)
+    - [Stomp](stomp_transport.md)
+    - [Filesystem](filesystem_transport.md)
+    - [Null](null_transport.md)
 * Consumption
     - [Extensions](consumption/extensions.md)
 * Client

phpunit.xml.dist

@@ -29,6 +29,10 @@
             <directory>pkg/amqp-ext/Tests</directory>
         </testsuite>
 
+        <testsuite name="fs">
+            <directory>pkg/fs/Tests</directory>
+        </testsuite>
+
         <testsuite name="enqueue-bundle">
             <directory>pkg/enqueue-bundle/Tests</directory>
         </testsuite>

pkg/amqp-ext/Tests/Symfony/AmqpTransportFactoryTest.php

@@ -121,5 +121,14 @@ public function testShouldCreateDriver()
 
         $driver = $container->getDefinition($serviceId);
         $this->assertSame(AmqpDriver::class, $driver->getClass());
+
+        $this->assertInstanceOf(Reference::class, $driver->getArgument(0));
+        $this->assertEquals('enqueue.transport.amqp.context', (string) $driver->getArgument(0));
+
+        $this->assertInstanceOf(Reference::class, $driver->getArgument(1));
+        $this->assertEquals('enqueue.client.config', (string) $driver->getArgument(1));
+
+        $this->assertInstanceOf(Reference::class, $driver->getArgument(2));
+        $this->assertEquals('enqueue.client.meta.queue_meta_registry', (string) $driver->getArgument(2));
     }
 }

pkg/enqueue-bundle/EnqueueBundle.php

@@ -11,6 +11,8 @@
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildQueueMetaRegistryPass;
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildTopicMetaSubscribersPass;
 use Enqueue\Bundle\DependencyInjection\EnqueueExtension;
+use Enqueue\Fs\FsContext;
+use Enqueue\Fs\Symfony\FsTransportFactory;
 use Enqueue\Stomp\StompContext;
 use Enqueue\Stomp\Symfony\RabbitMqStompTransportFactory;
 use Enqueue\Stomp\Symfony\StompTransportFactory;
@@ -46,5 +48,9 @@ public function build(ContainerBuilder $container)
             $extension->addTransportFactory(new AmqpTransportFactory());
             $extension->addTransportFactory(new RabbitMqAmqpTransportFactory());
         }
+
+        if (class_exists(FsContext::class)) {
+            $extension->addTransportFactory(new FsTransportFactory());
+        }
     }
 }

pkg/enqueue-bundle/Tests/Unit/EnqueueBundleTest.php

@@ -11,6 +11,7 @@
 use Enqueue\Bundle\DependencyInjection\Compiler\BuildTopicMetaSubscribersPass;
 use Enqueue\Bundle\DependencyInjection\EnqueueExtension;
 use Enqueue\Bundle\EnqueueBundle;
+use Enqueue\Fs\Symfony\FsTransportFactory;
 use Enqueue\Stomp\Symfony\RabbitMqStompTransportFactory;
 use Enqueue\Stomp\Symfony\StompTransportFactory;
 use Enqueue\Symfony\DefaultTransportFactory;
@@ -139,6 +140,23 @@ public function testShouldRegisterAmqpAndRabbitMqAmqpTransportFactories()
         $bundle->build($container);
     }
 
+    public function testShouldRegisterFSTransportFactory()
+    {
+        $extensionMock = $this->createEnqueueExtensionMock();
+
+        $container = new ContainerBuilder();
+        $container->registerExtension($extensionMock);
+
+        $extensionMock
+            ->expects($this->at(6))
+            ->method('addTransportFactory')
+            ->with($this->isInstanceOf(FsTransportFactory::class))
+        ;
+
+        $bundle = new EnqueueBundle();
+        $bundle->build($container);
+    }
+
     /**
      * @return \PHPUnit_Framework_MockObject_MockObject|EnqueueExtension
      */

pkg/fs/.gitignore

@@ -0,0 +1,6 @@
+*~
+/composer.lock
+/composer.phar
+/phpunit.xml
+/vendor/
+/.idea/

pkg/fs/.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 1
+
+language: php
+
+php:
+    - '5.6'
+    - '7.0'
+
+cache:
+    directories:
+        - $HOME/.composer/cache
+
+install:
+    - composer self-update
+    - composer install --prefer-source
+
+script:
+    - vendor/bin/phpunit --exclude-group=functional