Jenner\SimpleFork\Queue\PipeQueue

Complexity

Total Complexity

14

Size/Duplication

Total Lines	93
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	1

Test Coverage

Coverage

58.54%

Importance

Changes	10
Bugs	0	Features	10

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	put()	0	11	2
D	get()	0	33	10
A	remove()	0	5	1

<?php
/**
 * @author Jenner <hypxm@qq.com>
 * @blog http://www.huyanping.cn
 * @license https://opensource.org/licenses/MIT MIT
 * @datetime: 2015/11/24 18:38
 */

namespace Jenner\SimpleFork\Queue;

There must be one blank line after the namespace declaration

class PipeQueue implements QueueInterface
{
    /**
     * @var Pipe
     */
    protected $pipe;

    /**
     * @var bool
     */
    protected $block;

    /**
     * @param string $filename fifo filename
     * @param int $mode
     * @param bool $block if blocking

There is no parameter named $block. Was it maybe removed?

     */
    public function __construct($filename = '/tmp/simple-fork.pipe', $mode = 0666)
    {
        $this->pipe = new Pipe($filename, $mode);
        $this->block = false;
        $this->pipe->setBlock($this->block);
    }

    /**
     * put value into the queue of channel
     *
     * @param $value
     * @return bool
     */
    public function put($value)
    {
        $len = strlen($value);
        if ($len > 2147483647) {
            throw new \RuntimeException("value is too long");
        }
        $raw = pack('N', $len) . $value;
        $write_len = $this->pipe->write($raw);

        return $write_len == strlen($raw);
    }

    /**
     * get value from the queue of channel
     *
     * @param bool $block if block when the queue is empty
     * @return bool|string
     */
    public function get($block = false)
    {
        if ($this->block != $block) {
            $this->pipe->setBlock($block);
            $this->block = $block;
        }
        $len = $this->pipe->read(4);
        if ($len === false) {
            throw new \RuntimeException("read pipe failed");
        }

        if (strlen($len) === 0) {
            return null;
        }
        $len = unpack('N', $len);
        if(empty($len) || !array_key_exists(1, $len) || empty($len[1])){
            throw new \RuntimeException("data protocol error");
        }
        $len = intval($len[1]);

        $value = '';
        while (true) {
            $temp = $this->pipe->read($len);
            if (strlen($temp) == $len) {
                return $temp;
            }
            $value .= $temp;
            $len -= strlen($temp);
            if ($len == 0) {
                return $value;
            }
        }
    }

    /**
     * remove the queue resource
     *
     * @return bool
     */
    public function remove()
    {
        $this->pipe->close();
        $this->pipe->remove();
    }
}