Added locking capabilities to process visits command

composer.json

@@ -32,6 +32,7 @@
         "roave/security-advisories": "dev-master",
         "symfony/console": "^4.1",
         "symfony/filesystem": "^4.1",
+        "symfony/lock": "^4.1",
         "symfony/process": "^4.1",
         "theorchard/monolog-cascade": "^0.4",
         "zendframework/zend-config": "^3.0",

config/autoload/locks.global.php

@@ -0,0 +1,25 @@
+<?php
+declare(strict_types=1);
+
+use Symfony\Component\Lock;
+use Zend\ServiceManager\AbstractFactory\ConfigAbstractFactory;
+
+return [
+
+    'locks' => [
+        'locks_dir' => __DIR__ . '/../../data/locks',
+    ],
+
+    'dependencies' => [
+        'factories' => [
+            Lock\Store\FlockStore::class => ConfigAbstractFactory::class,
+            Lock\Factory::class => ConfigAbstractFactory::class,
+        ],
+    ],
+
+    ConfigAbstractFactory::class => [
+        Lock\Store\FlockStore::class => ['config.locks.locks_dir'],
+        Lock\Factory::class => [Lock\Store\FlockStore::class],
+    ],
+
+];

data/locks/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

module/CLI/config/dependencies.config.php

@@ -9,6 +9,7 @@ use Shlinkio\Shlink\Common\Service\PreviewGenerator;
 use Shlinkio\Shlink\Core\Service;
 use Shlinkio\Shlink\Rest\Service\ApiKeyService;
 use Symfony\Component\Console\Application;
+use Symfony\Component\Lock;
 use Zend\I18n\Translator\Translator;
 use Zend\ServiceManager\AbstractFactory\ConfigAbstractFactory;
 
@@ -68,6 +69,7 @@ return [
         Command\Visit\ProcessVisitsCommand::class => [
             Service\VisitService::class,
             IpLocationResolverInterface::class,
+            Lock\Factory::class,
             'translator',
         ],
         Command\Visit\UpdateDbCommand::class => [DbUpdater::class, 'translator'],

module/CLI/src/Command/Visit/ProcessVisitsCommand.php

@@ -14,12 +14,14 @@ use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\Lock\Factory as Locker;
 use Zend\I18n\Translator\TranslatorInterface;
 use function sprintf;
 
 class ProcessVisitsCommand extends Command
 {
     public const NAME = 'visit:process';
+    private const LOCK_NAME = 'visit_process';
 
     /**
      * @var VisitServiceInterface
@@ -33,41 +35,80 @@ class ProcessVisitsCommand extends Command
      * @var TranslatorInterface
      */
     private $translator;
+    /**
+     * @var Locker
+     */
+    private $locker;
+    /**
+     * @var OutputInterface
+     */
+    private $output;
 
     public function __construct(
         VisitServiceInterface $visitService,
         IpLocationResolverInterface $ipLocationResolver,
+        Locker $locker,
         TranslatorInterface $translator
     ) {
         $this->visitService = $visitService;
         $this->ipLocationResolver = $ipLocationResolver;
         $this->translator = $translator;
+        $this->locker = $locker;
         parent::__construct();
     }
 
     protected function configure(): void
     {
-        $this->setName(self::NAME)
-             ->setDescription(
-                 $this->translator->translate('Processes visits where location is not set yet')
-             );
+        $this
+            ->setName(self::NAME)
+            ->setDescription($this->translator->translate('Processes visits where location is not set yet'));
     }
 
     protected function execute(InputInterface $input, OutputInterface $output): void
     {
-        $this->visitService->locateVisits(function (Visit $visit) use ($output) {
-            if (! $visit->hasRemoteAddr()) {
-                $output->writeln(
-                    sprintf('<comment>%s</comment>', $this->translator->translate('Ignored visit with no IP address')),
-                    OutputInterface::VERBOSITY_VERBOSE
+        $this->output = $output;
+        $io = new SymfonyStyle($input, $output);
+
+        $lock = $this->locker->createLock(self::LOCK_NAME);
+        if (! $lock->acquire()) {
+            $io->warning(sprintf(
+                $this->translator->translate('There is already an instance of the "%s" command in execution'),
+                self::NAME
+            ));
+            return;
+        }
+
+        try {
+            $this->visitService->locateVisits(
+                [$this, 'getGeolocationDataForVisit'],
+                function (VisitLocation $location) use ($output) {
+                    $output->writeln(sprintf(
+                        ' [<info>' . $this->translator->translate('Address located at "%s"') . '</info>]',
+                        $location->getCountryName()
+                    ));
+                }
             );
+
+            $io->success($this->translator->translate('Finished processing all IPs'));
+        } finally {
+            $lock->release();
+        }
+    }
+
+    public function getGeolocationDataForVisit(Visit $visit): array
+    {
+        if (! $visit->hasRemoteAddr()) {
+            $this->output->writeln(sprintf(
+                '<comment>%s</comment>',
+                $this->translator->translate('Ignored visit with no IP address')
+            ), OutputInterface::VERBOSITY_VERBOSE);
             throw new IpCannotBeLocatedException('Ignored visit with no IP address');
         }
 
         $ipAddr = $visit->getRemoteAddr();
-            $output->write(sprintf('%s <fg=blue>%s</>', $this->translator->translate('Processing IP'), $ipAddr));
+        $this->output->write(sprintf('%s <fg=blue>%s</>', $this->translator->translate('Processing IP'), $ipAddr));
         if ($ipAddr === IpAddress::LOCALHOST) {
-                $output->writeln(
+            $this->output->writeln(
                 sprintf(' [<comment>%s</comment>]', $this->translator->translate('Ignored localhost address'))
             );
             throw new IpCannotBeLocatedException('Ignored localhost address');
@@ -76,26 +117,17 @@ class ProcessVisitsCommand extends Command
         try {
             return $this->ipLocationResolver->resolveIpLocation($ipAddr);
         } catch (WrongIpException $e) {
-                $output->writeln(
+            $this->output->writeln(
                 sprintf(
                     ' [<fg=red>%s</>]',
                     $this->translator->translate('An error occurred while locating IP. Skipped')
                 )
             );
-                if ($output->isVerbose()) {
-                    $this->getApplication()->renderException($e, $output);
+            if ($this->output->isVerbose()) {
+                $this->getApplication()->renderException($e, $this->output);
             }
 
             throw new IpCannotBeLocatedException('An error occurred while locating IP', 0, $e);
         }
-        }, function (VisitLocation $location) use ($output) {
-            $output->writeln(sprintf(
-                ' [<info>' . $this->translator->translate('Address located at "%s"') . '</info>]',
-                $location->getCountryName()
-            ));
-        });
-
-        $io = new SymfonyStyle($input, $output);
-        $io->success($this->translator->translate('Finished processing all IPs'));
     }
 }

module/CLI/test/Command/Visit/ProcessVisitsCommandTest.php

@@ -19,9 +19,11 @@ use Shlinkio\Shlink\Core\Service\VisitService;
 use Symfony\Component\Console\Application;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Tester\CommandTester;
+use Symfony\Component\Lock;
 use Throwable;
 use Zend\I18n\Translator\Translator;
 use function array_shift;
+use function sprintf;
 
 class ProcessVisitsCommandTest extends TestCase
 {
@@ -37,15 +39,31 @@ class ProcessVisitsCommandTest extends TestCase
      * @var ObjectProphecy
      */
     private $ipResolver;
+    /**
+     * @var ObjectProphecy
+     */
+    private $locker;
+    /**
+     * @var ObjectProphecy
+     */
+    private $lock;
 
     public function setUp()
     {
         $this->visitService = $this->prophesize(VisitService::class);
         $this->ipResolver = $this->prophesize(IpApiLocationResolver::class);
 
+        $this->locker = $this->prophesize(Lock\Factory::class);
+        $this->lock = $this->prophesize(Lock\LockInterface::class);
+        $this->lock->acquire()->willReturn(true);
+        $this->lock->release()->will(function () {
+        });
+        $this->locker->createLock(Argument::type('string'))->willReturn($this->lock->reveal());
+
         $command = new ProcessVisitsCommand(
             $this->visitService->reveal(),
             $this->ipResolver->reveal(),
+            $this->locker->reveal(),
             Translator::factory([])
         );
         $app = new Application();
@@ -160,4 +178,28 @@ class ProcessVisitsCommandTest extends TestCase
             $resolveIpLocation->shouldHaveBeenCalledOnce();
         }
     }
+
+    /**
+     * @test
+     */
+    public function noActionIsPerformedIfLockIsAcquired()
+    {
+        $this->lock->acquire()->willReturn(false);
+
+        $locateVisits = $this->visitService->locateVisits(Argument::cetera())->will(function () {
+        });
+        $resolveIpLocation = $this->ipResolver->resolveIpLocation(Argument::any())->willReturn([]);
+
+        $this->commandTester->execute([
+            'command' => 'visit:process',
+        ], ['verbosity' => OutputInterface::VERBOSITY_VERBOSE]);
+        $output = $this->commandTester->getDisplay();
+
+        $this->assertContains(
+            sprintf('There is already an instance of the "%s" command', ProcessVisitsCommand::NAME),
+            $output
+        );
+        $locateVisits->shouldNotHaveBeenCalled();
+        $resolveIpLocation->shouldNotHaveBeenCalled();
+    }
 }