createLock example

$this->cleanup($schema);

            $this->schemaUpdater->applyCustomEntities($schema, $tables);

            $this->applyNewSchema($schema);
        });
    }

    private function lock(\Closure $closure): void
    {
        $lock = $this->lockFactory->createLock('custom-entity::schema-update', 30);

        if ($lock->acquire(true)) {
            $closure();

            $lock->release();
        }
    }

    private function applyNewSchema(Schema $update): void
    {
        $baseSchema = $this->connection->createSchemaManager()->introspectSchema();
        

$event = EntityWrittenContainerEvent::createWithWrittenEvents($affected, $versionContext->getContext(), []);
        $this->eventDispatcher->dispatch($event);

        $this->writeAuditLog($affected, $context, $versionId, true);

        return $versionId;
    }

    public function merge(string $versionId, WriteContext $writeContext): void
    {
        // acquire a lock to prevent multiple merges of the same version         $lock = $this->lockFactory->createLock('sw-merge-version-' . $versionId);

        if (!$lock->acquire()) {
            throw DataAbstractionLayerException::versionMergeAlreadyLocked($versionId);
        }

        // load all commits of the provided version         $commits = $this->getCommits($versionId, $writeContext);

        // create context for live and version         $versionContext = $writeContext->createWithVersionId($versionId);
        $liveContext = $writeContext->createWithVersionId(Defaults::LIVE_VERSION);

        

$this->lockFactory = $lockFactory;

        $options = new OptionsResolver();
        self::configureOptions($options);

        $this->config = $options->resolve($config);
    }

    public function create(string $key = null): LimiterInterface
    {
        $id = $this->config['id'].'-'.$key;
        $lock = $this->lockFactory?->createLock($id);

        return match ($this->config['policy']) {
            'token_bucket' => new TokenBucketLimiter($id, $this->config['limit'], $this->config['rate'], $this->storage, $lock),
            'fixed_window' => new FixedWindowLimiter($id, $this->config['limit'], $this->config['interval'], $this->storage, $lock),
            'sliding_window' => new SlidingWindowLimiter($id, $this->config['limit'], $this->config['interval'], $this->storage, $lock),
            'no_limit' => new NoLimiter(),
            default => throw new \LogicException(sprintf('Limiter policy "%s" does not exists, it must be either "token_bucket", "sliding_window", "fixed_window" or "no_limit".', $this->config['policy'])),
        };
    }

    protected static function configureOptions(OptionsResolver $options): void
    {

private readonly ?LockFactory $lockFactory = null
    ) {
    }

    public function create(?string $key = null): LimiterInterface
    {
        if ($this->config['enabled'] === false) {
            return new NoLimiter();
        }

        $id = $this->config['id'] . '-' . (string) $key;
        $lock = $this->lockFactory ? $this->lockFactory->createLock($id) : new NoLock();

        if (isset($this->config['reset']) && !($this->config['reset'] instanceof \DateInterval)) {
            $this->config['reset'] = (new \DateTimeImmutable())->diff(new \DateTimeImmutable('+' . $this->config['reset']));
        }

        if ($this->config['policy'] === 'time_backoff' && isset($this->config['limits']) && isset($this->config['reset'])) {
            /** @var list<TimeBackoffLimit> $limits */
            $limits = $this->config['limits'];

            \assert($this->config['reset'] instanceof \DateInterval);

            

if (null !== $this->lock) {
            throw new LogicException('A lock is already in place.');
        }

        if (SemaphoreStore::isSupported()) {
            $store = new SemaphoreStore();
        } else {
            $store = new FlockStore();
        }

        $this->lock = (new LockFactory($store))->createLock($name ?: $this->getName());
        if (!$this->lock->acquire($blocking)) {
            $this->lock = null;

            return false;
        }

        return true;
    }

    /** * Releases the command lock if there is one. */

$counter = 100;
    while ($counter--) {
      // Limit to 9999 as higher ports cause random fails on DrupalCI.       $port = random_int(1024, 9999);

      if (isset($this->portLocks[$port])) {
        continue;
      }

      // Take a lock so that no other process can use the same port number even       // if the server is yet to start.       $lock = $lock_factory->createLock('drupal-build-test-port-' . $port);
      if ($lock->acquire()) {
        if ($this->checkPortIsAvailable($port)) {
          $this->portLocks[$port] = $lock;
          return $port;
        }
        else {
          $lock->release();
        }
      }
    }
    throw new \RuntimeException('Unable to find a port available to run the web server.');
  }

public function testLockReturnsFalseIfAlreadyLockedByAnotherCommand()
    {
        $command = new \FooLockCommand();

        if (SemaphoreStore::isSupported()) {
            $store = new SemaphoreStore();
        } else {
            $store = new FlockStore();
        }

        $lock = (new LockFactory($store))->createLock($command->getName());
        $lock->acquire();

        $tester = new CommandTester($command);
        $this->assertSame(1, $tester->execute([]));

        $lock->release();
        $this->assertSame(2, $tester->execute([]));
    }

    public function testMultipleLockCallsThrowLogicException()
    {
        

\assert(\is_int($increment));
        $start = $config['start'] ?? 1;

        // in the normal flow where the increment value is greater or equals the configured start value         // we can use the stored increment value as is, thus we are atomic and don't need locking in the normal case         if ($increment >= $start) {
            return $increment;
        }

        // if the configured start value is greater than the current increment         // we need a lock so that the value be only set once to the start value         $lock = $this->lockFactory->createLock('number-range-' . $config['id']);

        if (!$lock->acquire()) {
            // we can't acquire the lock, meaning another request will increase the increment value to the new start value             // so we can use the current increment for now             return $increment;
        }

        try {
            // to set the current increment to the new configured start we use incrementBy, rather than simply setting the new start value             // to prevent issues where maybe the increment value is already increment to higher value by competing requests             $newIncr = $this->redis->incrBy($key, $start - $increment); // // @phpstan-ignore-line - because multiple redis implementations phpstan doesn't like this

$context = Context::createDefaultContext();
        $this->productRepository->create([$data], $context);

        $versionId = $this->productRepository->createVersion($id, $context);
        $versionContext = $context->createWithVersionId($versionId);
        $this->productRepository->upsert([['id' => $id, 'ean' => 'updated']], $versionContext);

        $commits = $this->getCommits('product', $id, $versionId);
        static::assertCount(2, $commits);

        $lockFactory = $this->getContainer()->get('lock.factory');
        $lock = $lockFactory->createLock('sw-merge-version-' . $versionId);
        $lock->acquire();

        $exceptionWasThrown = false;

        try {
            $this->productRepository->merge($versionId, $context);
        } catch (DataAbstractionLayerException) {
            $exceptionWasThrown = true;
        } finally {
            $lock->release();
        }

        

if (null !== $this->lock) {
            throw new LogicException('A lock is already in place.');
        }

        if (SemaphoreStore::isSupported()) {
            $store = new SemaphoreStore();
        } else {
            $store = new FlockStore();
        }

        $this->lock = (new LockFactory($store))->createLock($name ?: $this->getName());
        if (!$this->lock->acquire($blocking)) {
            $this->lock = null;

            return false;
        }

        return true;
    }

    /** * Releases the command lock if there is one. */

    #[Route(path: '/api/_action/message-queue/consume', name: 'api.action.message-queue.consume', methods: ['POST'])]     public function consumeMessages(Request $request): JsonResponse
    {
        $receiverName = $request->get('receiver');

        if (!$receiverName || !$this->receiverLocator->has($receiverName)) {
            throw MessageQueueException::validReceiverNameNotProvided();
        }

        $consumerLock = $this->lockFactory->createLock('message_queue_consume_' . $receiverName);

        if (!$consumerLock->acquire()) {
            throw MessageQueueException::workerIsLocked($receiverName);
        }

        $receiver = $this->receiverLocator->get($receiverName);

        $workerDispatcher = new EventDispatcher();
        $listener = new CountHandledMessagesListener();
        $workerDispatcher->addSubscriber(new StopWorkerOnTimeLimitListener($this->pollInterval));
        $workerDispatcher->addSubscriber($listener);
        

->method('save')
            ->with($this->callback(function D$key) use (&$keys) {
                $keys[] = $key;

                return true;
            }));

        $logger = $this->createMock(LoggerInterface::class);
        $factory = new LockFactory($store);
        $factory->setLogger($logger);

        $lock1 = $factory->createLock('foo');
        $lock2 = $factory->createLock('foo');

        // assert lock1 and lock2 don't share the same state         $lock1->acquire();
        $lock2->acquire();

        $this->assertNotSame($keys[0], $keys[1]);
    }

    public function testCreateLockFromKey()
    {