Futures/FutureCollection.php at master · longjos/Futures · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
<?php
/**
 * FutureCollection
 *
 * Allows for a Collection of Futures to be acted on completion
 *
 * PHP Version 5.3
 *
 * @author Josh Long <longjos@gmail.com>
 * @copyright 2011 Josh Long
 * @license http://www.opensource.org/licenses/BSD-3-Clause BSD 3-Clause
 * @link https://github.com/spiralout/Futures
 */
class FutureCollection
{

    const VALUE_READY = 1;

    /** max value size */
    const SIZE = 100000;

    private $futures = array();


    private $collectionQueueId;

    private $collectionQueueKey;

    function __construct()
    {
        if(!($this->collectionQueueId = msg_get_queue($this->getMessageQueueId())))
        {
            throw Exception("Unable to get Collection Queue");
        }
    }

    /**
     * Add a Future to this collection
     *
     * @param Future $future
     * @return FutureCollection Return self for fluent interface
     */
    function addFuture(Future $future)
    {
        $this->futures[$future->getFutureId()] = $future;
        return $this;
    }

    /**
     * Return the next Future with VALUE_READY
     *
     * @return Future
     */
    function getNextFuture()
    {
         if (!(msg_receive($this->collectionQueueId, self::VALUE_READY, $msgType, self::SIZE, $futureId, true, 0, $error))) {
                return false;
            }
         $finishedFuture = $this->futures[$futureId];
         unset($this->futures[$futureId]);
         return $finishedFuture;;
    }

    /**
     * Return the number of futures in this collection
     *
     * @return int
     */
    function count()
    {
        return count($this->futures);
    }

    /**
     * Return the key for this collection
     *
     * @return type
     */
    function getCollectionId()
    {
        return $this->getMessageQueueId();
    }

    /**
     * Get a unique ID for our collection queue
     *
     * @return int
     */
    private function getMessageQueueId()
    {
        if(!isset($this->collectionQueueKey))
        {
            $this->collectionQueueKey = floor(microtime(true) * 100000);
        }
        return $this->collectionQueueKey;
    }
}