DatabaseQueue.php

Go to the documentation of this file.

<?php

namespace Elgg\Queue;

use Elgg\Database\Delete;
use Elgg\Database\Insert;
use Elgg\Database\Select;
use Elgg\Database\Update;
use Elgg\Traits\TimeUsing;

class DatabaseQueue implements \Elgg\Queue\Queue {

    use TimeUsing;
    
    const TABLE_NAME = 'queue';
    
    protected $name;

    protected $db;

    protected $workerId;

    public function __construct(string $name, \Elgg\Database $db) {
        $this->db = $db;
        $this->name = $name;
        $this->workerId = md5(microtime() . getmypid());
    }

    public function enqueue($item) {
        $insert = Insert::intoTable(self::TABLE_NAME);
        $insert->values([
            'name' => $insert->param($this->name, ELGG_VALUE_STRING),
            'data' => $insert->param(serialize($item), ELGG_VALUE_STRING),
            'timestamp' => $insert->param($this->getCurrentTime()->getTimestamp(), ELGG_VALUE_TIMESTAMP),
        ]);
        
        return $this->db->insertData($insert) !== false;
    }

    public function dequeue() {
        // get a record for processing
        $select = Select::fromTable(self::TABLE_NAME);
        $select->select('*')
            ->where($select->compare('name', '=', $this->name, ELGG_VALUE_STRING))
            ->andWhere($select->expr()->isNull('worker'))
            ->orderBy('id', 'ASC')
            ->setMaxResults(1);
        
        $row = $this->db->getDataRow($select);
        if (empty($row)) {
            return;
        }
        
        // lock a record for processing
        $update = Update::table(self::TABLE_NAME);
        $update->set('worker', $update->param($this->workerId, ELGG_VALUE_STRING))
            ->where($update->compare('name', '=', $this->name, ELGG_VALUE_STRING))
            ->andWhere($update->compare('id', '=', $row->id, ELGG_VALUE_ID))
            ->andWhere($update->expr()->isNull('worker'));
        
        if ($this->db->updateData($update, true) !== 1) {
            return;
        }
        
        // remove locked record from database
        $delete = Delete::fromTable(self::TABLE_NAME);
        $delete->where($delete->compare('id', '=', $row->id, ELGG_VALUE_ID));
        
        $this->db->deleteData($delete);
        
        return unserialize($row->data);
    }

    public function clear() {
        $delete = Delete::fromTable(self::TABLE_NAME);
        $delete->where($delete->compare('name', '=', $this->name, ELGG_VALUE_STRING));
        
        $this->db->deleteData($delete);
    }

    public function size() {
        $select = Select::fromTable(self::TABLE_NAME);
        $select->select('COUNT(*) AS total')
            ->where($select->compare('name', '=', $this->name, ELGG_VALUE_STRING));
        
        $result = $this->db->getDataRow($select);
        return (int) $result->total;
    }
}