DatabaseQueue.php

Go to the documentation of this file.

<?php

namespace Elgg\Queue;

class DatabaseQueue implements \Elgg\Queue\Queue {

    protected $name;

    protected $db;

    protected $workerId;

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

    public function enqueue($item) {
        $prefix = $this->db->prefix;
        
        $query = "INSERT INTO {$prefix}queue
            (name, data, timestamp)
            VALUES
            (:name, :data, :timestamp)";
        $params = [
            ':name' => $this->name,
            ':data' => serialize($item),
            ':timestamp' => time(),
        ];
        return $this->db->insertData($query, $params) !== false;
    }

    public function dequeue() {
        $prefix = $this->db->prefix;
        $name = $this->name;
        $worker_id = $this->workerId;
        
        $update = "UPDATE {$prefix}queue
            SET worker = :worker
            WHERE name = :name AND worker IS NULL
            ORDER BY id ASC LIMIT 1";
        $update_params = [
            ':worker' => $worker_id,
            ':name' => $name,
        ];
        $num = $this->db->updateData($update, true, $update_params);
        if ($num !== 1) {
            return;
        }
        
        $select = "SELECT data
            FROM {$prefix}queue
            WHERE worker = :worker
            AND name = :name";
        $select_params = [
            ':worker' => $worker_id,
            ':name' => $name,
        ];
        $obj = $this->db->getDataRow($select, null, $select_params);
        if (empty($obj)) {
            return;
        }
        
        $delete = "DELETE FROM {$prefix}queue
            WHERE name = :name
            AND worker = :worker";
        $delete_params = [
            ':worker' => $worker_id,
            ':name' => $name,
        ];
        $this->db->deleteData($delete, $delete_params);
        
        return unserialize($obj->data);
    }

    public function clear() {
        $prefix = $this->db->prefix;
        
        $sql = "DELETE FROM {$prefix}queue
            WHERE name = :name";
        $params = [
            ':name' => $this->name,
        ];

        $this->db->deleteData($sql, $params);
    }

    public function size() {
        $prefix = $this->db->prefix;
        
        $sql = "SELECT COUNT(id) AS total
            FROM {$prefix}queue
            WHERE name = :name";
        $params = [
            ':name' => $this->name,
        ];
        
        $result = $this->db->getDataRow($sql, null, $params);
        return (int) $result->total;
    }
}