Merge pull request #168 from mbonneau/priority-queue-patch

Remove cancelled items from PriorityQueue

Author: davidwdan

src/Scheduler/PriorityQueue.php

@@ -22,6 +22,10 @@ public function enqueue(ScheduledItem $item)
 
     public function remove($item)
     {
+        if ($this->count() === 0) {
+            return false;
+        }
+
         if ($this->peek() === $item) {
             $this->dequeue();
             return true;

src/Scheduler/VirtualTimeScheduler.php

@@ -5,6 +5,7 @@
 namespace Rx\Scheduler;
 
 use Rx\AsyncSchedulerInterface;
+use Rx\Disposable\CallbackDisposable;
 use Rx\Disposable\EmptyDisposable;
 use Rx\Disposable\SerialDisposable;
 use Rx\DisposableInterface;
@@ -87,7 +88,10 @@ public function scheduleAbsoluteWithState($state, int $dueTime, callable $action
 
         $this->queue->enqueue($scheduledItem);
 
-        return $scheduledItem->getDisposable();
+        return new CallbackDisposable(function () use ($scheduledItem) {
+            $scheduledItem->getDisposable()->dispose();
+            $this->queue->remove($scheduledItem);
+        });
     }
 
     public function scheduleRelativeWithState($state, $dueTime, $action): DisposableInterface

test/Rx/Scheduler/PriorityQueueTest.php

@@ -130,4 +130,21 @@ public function can_remove_scheduled_items_out_of_order()
         $this->assertSame($scheduledItem, $queue->dequeue());
         $this->assertSame($scheduledItem3, $queue->dequeue());
     }
+
+    /**
+     * @test
+     */
+    public function should_not_remove_nonexistent_item()
+    {
+        $queue = new PriorityQueue();
+        $queue->remove(
+            new ScheduledItem(
+                $this->createMock(ScheduledItem::class),
+                null,
+                function () {
+                },
+                0
+            )
+        );
+    }
 }