protected function doPop()
    {
        $this->getPersistence()->getPlatform()->beginTransaction();

        $logContext = [
            'Queue' => $this->getQueueNameWithPrefix()
        ];

        try {
            $qb = $this->getQueryBuilder()
                ->select('id, message')
                ->from($this->getTableName())
                ->andWhere('visible = :visible')
                ->orderBy('created_at')
                ->setMaxResults($this->getNumberOfTasksToReceive());

            /**
             * SELECT ... FOR UPDATE is used for locking
             *
             * @see https://dev.mysql.com/doc/refman/5.6/en/innodb-locking-reads.html
             */
            $sql = $qb->getSQL() .' '. $this->getPersistence()->getPlatForm()->getWriteLockSQL();

            if ($dbResult = $this->getPersistence()->query($sql, ['visible' => 1])->fetchAll(\PDO::FETCH_ASSOC)) {

                // set the received messages to invisible for other workers
                $qb = $this->getQueryBuilder()
                    ->update($this->getTableName())
                    ->set('visible', ':visible')
                    ->where('id IN ('. implode(',', array_column($dbResult, 'id')) .')')
                    ->setParameter('visible', 0);

                $qb->execute();

                foreach ($dbResult as $row) {
                    if ($task = $this->unserializeTask($row['message'], $row['id'], $logContext)) {
                        $task->setMetadata('RdsMessageId', $row['id']);
                        $this->pushPreFetchedMessage($task);
                    }
                }
            } else {
                $this->logDebug('No task in the queue.', $logContext);
            }

            $this->getPersistence()->getPlatform()->commit();
        } catch (\Exception $e) {
            $this->getPersistence()->getPlatform()->rollBack();
            $this->logError('Popping tasks failed with MSG: '. $e->getMessage(), $logContext);
        }
    }
protected function doPop()
    {
        $this->getPersistence()->getPlatform()->beginTransaction();

        $logContext = [
            'Queue' => $this->getQueueNameWithPrefix()
        ];

        try {
            $qb = $this->getQueryBuilder()
                ->select('id, message')
                ->from($this->getTableName())
                ->andWhere('visible = :visible')
                ->orderBy('created_at')
                ->setMaxResults($this->getNumberOfTasksToReceive());

            /**
             * SELECT ... FOR UPDATE is used for locking
             *
             * @see https://dev.mysql.com/doc/refman/5.6/en/innodb-locking-reads.html
             */
            $sql = $qb->getSQL() .' '. $this->getPersistence()->getPlatForm()->getWriteLockSQL();

            if ($dbResult = $this->getPersistence()->query($sql, ['visible' => 1])->fetchAll(\PDO::FETCH_ASSOC)) {

                // set the received messages to invisible for other workers
                $qb = $this->getQueryBuilder()
                    ->update($this->getTableName())
                    ->set('visible', ':visible')
                    ->where('id IN ('. implode(',', array_column($dbResult, 'id')) .')')
                    ->setParameter('visible', 0);

                $qb->execute();

                foreach ($dbResult as $row) {
                    if ($task = $this->unserializeTask($row['message'], $row['id'], $logContext)) {
                        $task->setMetadata('RdsMessageId', $row['id']);
                        $this->pushPreFetchedMessage($task);
                    }
                }
            } else {
                $this->logDebug('No task in the queue.', $logContext);
            }

            $this->getPersistence()->getPlatform()->commit();
        } catch (\Exception $e) {
            $this->getPersistence()->getPlatform()->rollBack();
            $this->logError('Popping tasks failed with MSG: '. $e->getMessage(), $logContext);
        }
    }
protected function doPop ( ) { $ this -> getPersistence ( ) -> getPlatform ( ) -> beginTransaction ( ) ; $ logContext = [ 'Queue' => $ this -> getQueueNameWithPrefix ( ) ] ; try { $ qb = $ this -> getQueryBuilder ( ) -> select ( 'id, message' ) -> from ( $ this -> getTableName ( ) ) -> andWhere ( 'visible = :visible' ) -> orderBy ( 'created_at' ) -> setMaxResults ( $ this -> getNumberOfTasksToReceive ( ) ) ; /**
             * SELECT ... FOR UPDATE is used for locking
             *
             * @see https://dev.mysql.com/doc/refman/5.6/en/innodb-locking-reads.html
             */ $ sql = $ qb -> getSQL ( ) . ' ' . $ this -> getPersistence ( ) -> getPlatForm ( ) -> getWriteLockSQL ( ) ; if ( $ dbResult = $ this -> getPersistence ( ) -> query ( $ sql , [ 'visible' => 1 ] ) -> fetchAll ( \ PDO :: FETCH_ASSOC ) ) { // set the received messages to invisible for other workers $ qb = $ this -> getQueryBuilder ( ) -> update ( $ this -> getTableName ( ) ) -> set ( 'visible' , ':visible' ) -> where ( 'id IN (' . implode ( ',' , array_column ( $ dbResult , 'id' ) ) . ')' ) -> setParameter ( 'visible' , 0 ) ; $ qb -> execute ( ) ; foreach ( $ dbResult as $ row ) { if ( $ task = $ this -> unserializeTask ( $ row [ 'message' ] , $ row [ 'id' ] , $ logContext ) ) { $ task -> setMetadata ( 'RdsMessageId' , $ row [ 'id' ] ) ; $ this -> pushPreFetchedMessage ( $ task ) ; } } } else { $ this -> logDebug ( 'No task in the queue.' , $ logContext ) ; } $ this -> getPersistence ( ) -> getPlatform ( ) -> commit ( ) ; } catch ( \ Exception $ e ) { $ this -> getPersistence ( ) -> getPlatform ( ) -> rollBack ( ) ; $ this -> logError ( 'Popping tasks failed with MSG: ' . $ e -> getMessage ( ) , $ logContext ) ; } }
Does the DBAL specific pop mechanism.
Does the DBAL specific pop mechanism .