Added locking mechanism

composer.json

@@ -11,9 +11,10 @@
     ],
     "require": {
         "php": "~5.5 || ~7.0",
-        "php-task/php-task": "dev-develop",
+        "php-task/php-task": "dev-master",
         "symfony/http-kernel": "^2.6 || ^3.0",
         "symfony/dependency-injection": "^2.6 || ^3.0",
+        "symfony/expression-language": "^2.6 || ^3.0",
         "symfony/config": "^2.6 || ^3.0",
         "symfony/console": "^2.6 || ^3.0",
         "doctrine/orm": "^2.5"

src/DependencyInjection/Configuration.php

@@ -19,6 +19,19 @@
  */
 class Configuration implements ConfigurationInterface
 {
+    /**
+     * @var string[]
+     */
+    private $lockingStorageAliases = [];
+
+    /**
+     * @param \string[] $lockingStorageAliases
+     */
+    public function __construct(array $lockingStorageAliases)
+    {
+        $this->lockingStorageAliases = $lockingStorageAliases;
+    }
+
     /**
      * {@inheritdoc}
      */
@@ -43,9 +56,28 @@ public function getConfigTreeBuilder()
                 ->arrayNode('run')
                     ->addDefaultsIfNotSet()
                     ->children()
-                        ->enumNode('mode')
-                            ->values(['off', 'listener'])
-                            ->defaultValue('off')
+                        ->enumNode('mode')->values(['off', 'listener'])->defaultValue('off')->end()
+                    ->end()
+                ->end()
+                ->arrayNode('locking')
+                    ->canBeEnabled()
+                    ->addDefaultsIfNotSet()
+                    ->children()
+                        ->enumNode('storage')
+                            ->values(array_keys($this->lockingStorageAliases))
+                            ->defaultValue('file')
+                        ->end()
+                        ->integerNode('ttl')->defaultValue(600)->end()
+                        ->arrayNode('storages')
+                            ->addDefaultsIfNotSet()
+                            ->children()
+                                ->arrayNode('file')
+                                    ->addDefaultsIfNotSet()
+                                    ->children()
+                                        ->scalarNode('directory')->defaultValue('%kernel.cache_dir%/tasks')->end()
+                                    ->end()
+                                ->end()
+                            ->end()
                         ->end()
                     ->end()
                 ->end()
@@ -63,4 +95,16 @@ public function getConfigTreeBuilder()
 
         return $treeBuilder;
     }
+
+    /**
+     * Returns id for given storage-alias.
+     *
+     * @param string $alias
+     *
+     * @return string
+     */
+    public function getLockingStorageId($alias)
+    {
+        return $this->lockingStorageAliases[$alias];
+    }
 }

src/DependencyInjection/TaskExtension.php

@@ -12,6 +12,7 @@
 namespace Task\TaskBundle\DependencyInjection;
 
 use Symfony\Component\Config\FileLocator;
+use Symfony\Component\Config\Loader\LoaderInterface;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
 use Symfony\Component\DependencyInjection\ContainerInterface;
 use Symfony\Component\DependencyInjection\Definition;
@@ -31,23 +32,29 @@ class TaskExtension extends Extension
      */
     public function load(array $configs, ContainerBuilder $container)
     {
-        $configuration = new Configuration();
+        $configuration = $this->getConfiguration($configs, $container);
         $config = $this->processConfiguration($configuration, $configs);
 
         $container->setParameter('task.system_tasks', $config['system_tasks']);
-        $container->setParameter('task.storage', $config['storage']);
+
+        $container->setAlias('task.lock.storage', $configuration->getLockingStorageId($config['locking']['storage']));
+        foreach (array_keys($config['locking']['storages']) as $key) {
+            $container->setParameter('task.lock.storages.' . $key, $config['locking']['storages'][$key]);
+        }
 
         $loader = new Loader\XmlFileLoader($container, new FileLocator(__DIR__ . '/../Resources/config'));
         $loader->load(sprintf('storage/%s.xml', $config['storage']));
         $loader->load('task_event_listener.xml');
         $loader->load('scheduler.xml');
         $loader->load('command.xml');
+        $loader->load('locking/services.xml');
 
         if ($config['run']['mode'] === 'listener') {
             $loader->load('listener.xml');
         }
 
         $this->loadDoctrineAdapter($config['adapters']['doctrine'], $container);
+        $this->loadLockingComponent($config['locking'], $container, $loader);
     }
 
     /**
@@ -70,4 +77,53 @@ private function loadDoctrineAdapter(array $config, ContainerBuilder $container)
             $container->setDefinition('task.adapter.doctrine.execution_listener', $definition);
         }
     }
+
+    /**
+     * Load services for locking component.
+     *
+     * @param array $config
+     * @param LoaderInterface $loader
+     * @param ContainerBuilder $container
+     */
+    private function loadLockingComponent(array $config, ContainerBuilder $container, LoaderInterface $loader)
+    {
+        if (!$config['enabled']) {
+            return $loader->load('locking/null.xml');
+        }
+
+        $loader->load('locking/services.xml');
+        $container->setParameter('task.lock.ttl', $config['ttl']);
+    }
+
+    /**
+     * Find storage aliases and related ids.
+     *
+     * @param ContainerBuilder $container
+     *
+     * @return array
+     */
+    private function getLockingStorageAliases(ContainerBuilder $container)
+    {
+        $taggedServices = $container->findTaggedServiceIds('task.lock.storage');
+
+        $result = [];
+        foreach ($taggedServices as $id => $tags) {
+            foreach ($tags as $tag) {
+                $result[$tag['alias']] = $id;
+            }
+        }
+
+        return $result;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getConfiguration(array $config, ContainerBuilder $container)
+    {
+        $loader = new Loader\XmlFileLoader($container, new FileLocator(__DIR__ . '/../Resources/config'));
+        $loader->load('locking/storages.xml');
+
+        return new Configuration($this->getLockingStorageAliases($container));
+    }
 }

src/Entity/TaskExecutionRepository.php

@@ -129,16 +129,26 @@ public function findByTaskUuid($taskUuid)
     /**
      * {@inheritdoc}
      */
-    public function findScheduled()
+    public function findNextScheduled(\DateTime $dateTime = null, array $skippedExecutions = [])
     {
-        $query = $this->createQueryBuilder('e')
+        $queryBuilder = $this->createQueryBuilder('e')
             ->innerJoin('e.task', 't')
             ->where('e.status = :status')
             ->andWhere('e.scheduleTime < :date')
-            ->setParameter('date', new \DateTime())
+            ->setParameter('date', $dateTime ?: new \DateTime())
             ->setParameter('status', TaskStatus::PLANNED)
-            ->getQuery();
+            ->setMaxResults(1);
 
-        return $query->getResult();
+        $expr = $queryBuilder->expr();
+        if (!empty($skippedExecutions)) {
+            $queryBuilder->andWhere($expr->not($expr->in('e.uuid', ':skipped')))
+                ->setParameter('skipped', $skippedExecutions);
+        }
+
+        try {
+            return $queryBuilder->getQuery()->getSingleResult();
+        } catch (NoResultException $exception) {
+            return null;
+        }
     }
 }

src/Locking/NullLock.php

@@ -0,0 +1,52 @@
+<?php
+
+/*
+ * This file is part of php-task library.
+ *
+ * (c) php-task
+ *
+ * This source file is subject to the MIT license that is bundled
+ * with this source code in the file LICENSE.
+ */
+
+namespace Task\TaskBundle\Locking;
+
+use Task\Lock\LockInterface;
+
+/**
+ * Implements LockInterface which does nothing.
+ */
+class NullLock implements LockInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function acquire($task)
+    {
+        return true;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function refresh($task)
+    {
+        return true;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function release($task)
+    {
+        return true;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function isAcquired($task)
+    {
+        return false;
+    }
+}

src/Resources/config/locking/null.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" ?>
+<container xmlns="http://symfony.com/schema/dic/services"
+           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+           xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+    <services>
+        <service id="task.lock" class="Task\TaskBundle\Locking\NullLock"/>
+    </services>
+</container>

src/Resources/config/locking/services.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" ?>
+<container xmlns="http://symfony.com/schema/dic/services"
+           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+           xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+    <services>
+        <service id="task.lock" class="Task\Lock\Lock">
+            <argument type="service" id="task.lock.storage"/>
+            <argument>%task.lock.ttl%</argument>
+        </service>
+    </services>
+</container>

src/Resources/config/locking/storages.xml

@@ -0,0 +1,12 @@
+<?xml version="1.0" ?>
+<container xmlns="http://symfony.com/schema/dic/services"
+           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+           xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+    <services>
+        <service id="task.lock.storage.file" class="Task\Lock\Storage\FileLockStorage">
+            <argument type="expression">parameter('task.lock.storages.file')['directory']</argument>
+
+            <tag name="task.lock.storage" alias="file"/>
+        </service>
+    </services>
+</container>

src/Resources/config/scheduler.xml

@@ -19,7 +19,9 @@
         <service id="task.runner" class="Task\Runner\TaskRunner">
             <argument type="service" id="task.storage.task_execution"/>
             <argument type="service" id="task.handler.factory"/>
+            <argument type="service" id="task.lock"/>
             <argument type="service" id="event_dispatcher"/>
+            <argument type="service" id="logger" on-invalid="ignore"/>
         </service>
     </services>
 </container>

tests/Functional/Command/ScheduleTaskCommandTest.php

@@ -94,7 +94,7 @@ public function testExecuteWithWorkloadAndIntervalAndEndDate()
         $this->assertEquals(TestHandler::class, $tasks[0]->getHandlerClass());
         $this->assertEquals('Test workload 1', $tasks[0]->getWorkload());
         $this->assertEquals('0 * * * *', $tasks[0]->getInterval());
-        $this->assertEquals($date, $tasks[0]->getLastExecution());
+        $this->assertEquals($date, $tasks[0]->getLastExecution(), '', 2);
     }
 
     public function testExecuteWithExecutionDate()

tests/Functional/Entity/TaskExecutionRepositoryTest.php

@@ -154,10 +154,8 @@ public function testFindScheduledPast()
 
         $execution = $this->save($task, new \DateTime('-1 hour'));
 
-        $result = $this->taskExecutionRepository->findScheduled();
-
-        $this->assertCount(1, $result);
-        $this->assertEquals($execution->getUuid(), $result[0]->getUuid());
+        $result = $this->taskExecutionRepository->findNextScheduled();
+        $this->assertEquals($execution->getUuid(), $result->getUuid());
     }
 
     public function testFindScheduledFuture()
@@ -167,7 +165,17 @@ public function testFindScheduledFuture()
 
         $this->save($task, new \DateTime('+1 hour'));
 
-        $this->assertEmpty($this->taskExecutionRepository->findScheduled());
+        $this->assertNull($this->taskExecutionRepository->findNextScheduled());
+    }
+
+    public function testFindScheduledSkipped()
+    {
+        $task = $this->createTask();
+        $this->taskRepository->save($task);
+
+        $this->save($task, new \DateTime('+1 hour'));
+
+        $this->assertNull($this->taskExecutionRepository->findNextScheduled());
     }
 
     /**

tests/Functional/Handler/TaskHandlerFactoryTest.php

@@ -12,7 +12,6 @@
 namespace Task\TaskBundle\Tests\Functional\Handler;
 
 use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
-use Task\Handler\TaskHandlerNotExistsException;
 use Task\TaskBundle\Handler\TaskHandlerFactory;
 use Task\TaskBundle\Tests\Functional\TestHandler;
 
@@ -39,10 +38,11 @@ public function testCreate()
         $this->assertInstanceOf(TestHandler::class, $this->taskHandlerFactory->create(TestHandler::class));
     }
 
+    /**
+     * @expectedException \Task\Handler\TaskHandlerNotExistsException
+     */
     public function testCreateNotExists()
     {
-        $this->setExpectedException(TaskHandlerNotExistsException::class);
-
         $this->taskHandlerFactory->create(\stdClass::class);
     }
 }

tests/app/config/config.yml

@@ -1,2 +1,8 @@
 parameters:
     kernel.secret: 12345
+
+task:
+    locking:
+        storages:
+            file:
+                directory: %kernel.cache_dir%/locks