[fs] Add tests for driver and transport factory.

Author: makasim

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/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/Client/FsDriver.php

@@ -5,6 +5,8 @@
 use Enqueue\Client\Config;
 use Enqueue\Client\DriverInterface;
 use Enqueue\Client\Message;
+use Enqueue\Client\MessagePriority;
+use Enqueue\Client\Meta\QueueMetaRegistry;
 use Enqueue\Fs\FsContext;
 use Enqueue\Fs\FsDestination;
 use Enqueue\Fs\FsMessage;
@@ -25,13 +27,20 @@ class FsDriver implements DriverInterface
     private $config;
 
     /**
-     * @param FsContext $context
-     * @param Config    $config
+     * @var QueueMetaRegistry
      */
-    public function __construct(FsContext $context, Config $config)
+    private $queueMetaRegistry;
+
+    /**
+     * @param FsContext         $context
+     * @param Config            $config
+     * @param QueueMetaRegistry $queueMetaRegistry
+     */
+    public function __construct(FsContext $context, Config $config, QueueMetaRegistry $queueMetaRegistry)
     {
         $this->context = $context;
         $this->config = $config;
+        $this->queueMetaRegistry = $queueMetaRegistry;
     }
 
     /**
@@ -74,8 +83,27 @@ public function sendToProcessor(Message $message)
     public function setupBroker(LoggerInterface $logger = null)
     {
         $logger = $logger ?: new NullLogger();
+        $log = function ($text, ...$args) use ($logger) {
+            $logger->debug(sprintf('[FsDriver] '.$text, ...$args));
+        };
+
+        // setup router
+        $routerTopic = $this->createRouterTopic();
+        $routerQueue = $this->createQueue($this->config->getRouterQueueName());
 
-        // TODO: implement it;
+        $log('Declare router exchange "%s" file: %s', $routerTopic->getTopicName(), $routerTopic->getFileInfo());
+        $this->context->declareDestination($routerTopic);
+
+        $log('Declare router queue "%s" file: %s', $routerQueue->getQueueName(), $routerTopic->getFileInfo());
+        $this->context->declareDestination($routerQueue);
+
+        // setup queues
+        foreach ($this->queueMetaRegistry->getQueuesMeta() as $meta) {
+            $queue = $this->createQueue($meta->getClientName());
+
+            $log('Declare processor queue "%s" file: %s', $queue->getQueueName(), $queue->getFileInfo());
+            $this->context->declareDestination($queue);
+        }
     }
 
     /**
@@ -126,6 +154,7 @@ public function createClientMessage(TransportMessage $message)
         $clientMessage->setContentType($message->getHeader('content_type'));
         $clientMessage->setMessageId($message->getMessageId());
         $clientMessage->setTimestamp($message->getTimestamp());
+        $clientMessage->setPriority(MessagePriority::NORMAL);
 
         return $clientMessage;
     }

pkg/fs/Symfony/FsTransportFactory.php

@@ -38,10 +38,15 @@ public function addConfiguration(ArrayNodeDefinition $builder)
                     ->cannotBeEmpty()
                     ->info('The store directory where all queue\topics files will be created and messages are stored')
                 ->end()
-                    ->integerNode('pre_fetch_count')
+                ->integerNode('pre_fetch_count')
                     ->min(1)
+                    ->defaultValue(1)
                     ->info('The option tells how many messages should be read from file at once. The feature save resources but could lead to bigger messages lose.')
                 ->end()
+                ->integerNode('chmod')
+                    ->defaultValue(0600)
+                    ->info('The queue files are created with this given permissions if not exist.')
+                ->end()
         ;
     }
 
@@ -84,6 +89,7 @@ public function createDriver(ContainerBuilder $container, array $config)
         $driver->setArguments([
             new Reference(sprintf('enqueue.transport.%s.context', $this->getName())),
             new Reference('enqueue.client.config'),
+            new Reference('enqueue.client.meta.queue_meta_registry'),
         ]);
 
         $driverId = sprintf('enqueue.client.%s.driver', $this->getName());