transactional example

/** * {@inheritdoc} */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $query = $this->createQuery();

        $helper = new ConsoleProgressHelper($output);
        $helper->start($query->fetchCount(), 'Start indexing stream search data');

        $this->container->get(\Doctrine\DBAL\Connection::class)->transactional(
            function D) use ($helper, $query) {
                $this->container->get(\Doctrine\DBAL\Connection::class)->executeUpdate('DELETE FROM s_customer_search_index');

                $indexer = $this->container->get(\Shopware\Bundle\CustomerSearchBundleDBAL\Indexing\SearchIndexerInterface::class);

                while ($ids = $query->fetch()) {
                    $indexer->populate($ids);
                    $helper->advance(\count($ids));
                }
            }
        );

        

    private static function retry(Connection $connection, \Closure $closure, int $counter)
    {
        ++$counter;

        try {
            return $connection->transactional($closure);
        } catch (RetryableException $retryableException) {
            if ($connection->getTransactionNestingLevel() > 0) {
                // If this RetryableTransaction was executed inside another transaction, do not retry this nested                 // transaction. Remember that the whole (outermost) transaction was already rolled back by the database                 // when any RetryableException is thrown.                 // Rethrow the exception here so only the outermost transaction is retried which in turn includes this                 // nested transaction.                 throw $retryableException;
            }

            if ($counter > 10) {
                

    /** * @return bool */
    public function refresh()
    {
        $helper = new CronJobProgressHelper();

        $query = $this->createQuery();

        $this->connection->transactional(function D) use ($query) {
            $this->connection->executeUpdate('DELETE FROM s_customer_search_index');

            while ($ids = $query->fetch()) {
                $this->searchIndexer->populate($ids);
            }
        });

        $streams = $this->fetchStreams();

        if (empty($streams)) {
            return true;
        }

public function __construct(Connection $connection, CustomerProviderInterface $provider)
    {
        $this->connection = $connection;
        $this->provider = $provider;
    }

    /** * @param int[] $ids */
    public function populate(array $ids)
    {
        $this->connection->transactional(function D) use ($ids) {
            $insert = $this->createInsertQuery();
            $customers = $this->provider->get($ids);
            foreach ($customers as $customer) {
                $insert->execute($this->buildData($customer));
            }
        });
    }

    /** * @deprecated in 5.6, will be removed in 5.8 without replacement */
    

foreach ($notCreatedDefaultFolders as $notCreatedDefaultFolder) {
            $this->createDefaultFolder(
                $connection,
                $notCreatedDefaultFolder['default_folder_id'],
                $notCreatedDefaultFolder['entity']
            );
        }
    }

    private function createDefaultFolder(Connection $connection, string $defaultFolderId, string $entity): void
    {
        $connection->transactional(function DConnection $connection) use ($defaultFolderId, $entity): void {
            $configurationId = Uuid::randomBytes();
            $folderId = Uuid::randomBytes();
            $folderName = $this->getMediaFolderName($entity);
            $private = 0;
            if ($entity === 'document') {
                $private = 1;
            }
            $connection->executeStatement(' INSERT INTO `media_folder_configuration` (`id`, `thumbnail_quality`, `create_thumbnails`, `private`, created_at) VALUES (:id, 80, 1, :private, :createdAt) ', [
                

 {
        $this->criteriaFactory = $criteriaFactory;
        $this->numberSearch = $numberSearch;
        $this->connection = $connection;
    }

    /** * @param int $streamId */
    public function populate($streamId, ProgressHelperInterface $helper)
    {
        $this->connection->transactional(function D) use ($streamId, $helper) {
            $criteria = $this->criteriaFactory->createCriteria($streamId);
            $criteria->offset(0)
                ->limit(100);

            $customers = $this->numberSearch->search($criteria);
            $helper->start($customers->getTotal(), 'Start indexing customers');

            $this->clearStreamIndex($streamId);

            $insert = $this->connection->prepare(
                'INSERT INTO s_customer_streams_mapping (stream_id, customer_id) VALUES (:streamId, :customerId)'
            );

/** * saves signed basket * * @param string $signature * @param array $basket * * @throws Exception */
    public function persist($signature, $basket)
    {
        $this->connection->transactional(
            function D) use ($signature, $basket) {
                $createdAt = new DateTime();

                $this->delete($signature);

                $this->connection->insert(self::DBAL_TABLE, [
                    'signature' => $signature,
                    'basket' => json_encode($basket),
                    'created_at' => $createdAt->format('Y-m-d'),
                ]);
            }
        );

return $this->reflectionHelper->unserialize(
            json_decode($stream->getConditions() ?? '', true),
            'Serialization error in Customer Stream'
        );
    }

    private function insertCustomers(array $customerIds, int $streamId): void
    {
        $connection = $this->connection;

        $connection->transactional(function D) use ($connection, $customerIds, $streamId) {
            $connection->executeStatement(
                'DELETE FROM s_customer_streams_mapping WHERE stream_id = :streamId',
                [':streamId' => $streamId]
            );

            $insert = $connection->prepare('INSERT INTO s_customer_streams_mapping (stream_id, customer_id) VALUES (:streamId, :customerId)');
            $customerIds = array_keys(array_flip($customerIds));

            foreach ($customerIds as $customerId) {
                $insert->executeStatement([
                    ':streamId' => $streamId,