1
0
mirror of https://github.com/danog/parallel.git synced 2024-12-02 17:52:14 +01:00
parallel/tests/Sync/ChannelTest.php

178 lines
4.8 KiB
PHP
Raw Normal View History

2015-08-03 07:20:06 +02:00
<?php
namespace Icicle\Tests\Concurrent\Sync;
use Icicle\Concurrent\Sync\Channel;
2015-08-03 07:58:08 +02:00
use Icicle\Coroutine;
use Icicle\Loop;
use Icicle\Socket\Stream\DuplexStream;
2015-09-03 01:29:48 +02:00
use Icicle\Stream\DuplexStreamInterface;
use Icicle\Stream\ReadableStreamInterface;
use Icicle\Stream\WritableStreamInterface;
use Icicle\Tests\Concurrent\TestCase;
2015-08-03 07:20:06 +02:00
class ChannelTest extends TestCase
2015-08-03 07:20:06 +02:00
{
2015-09-03 01:29:48 +02:00
public function testCreateSocketPair()
2015-08-03 07:20:06 +02:00
{
list($a, $b) = Channel::createSocketPair();
2015-08-03 07:20:06 +02:00
$this->assertInternalType('resource', $a);
$this->assertInternalType('resource', $b);
2015-08-03 07:20:06 +02:00
}
2015-09-03 01:29:48 +02:00
public function testIsOpen()
{
$mock = $this->getMock(DuplexStreamInterface::class);
$mock->expects($this->once())
->method('isOpen')
->will($this->returnValue(true));
$channel = new Channel($mock);
$channel->isOpen();
$readable = $this->getMock(ReadableStreamInterface::class);
$writable = $this->getMock(WritableStreamInterface::class);
$readable->expects($this->once())
->method('isOpen')
->will($this->returnValue(true));
$writable->expects($this->once())
->method('isOpen')
->will($this->returnValue(true));
$channel = new Channel($readable, $writable);
$channel->isOpen();
}
2015-08-03 07:20:06 +02:00
public function testClose()
{
2015-09-03 01:29:48 +02:00
$mock = $this->getMock(DuplexStreamInterface::class);
2015-08-03 07:20:06 +02:00
2015-09-03 01:29:48 +02:00
$mock->expects($this->once())
->method('close');
2015-08-03 07:20:06 +02:00
2015-09-03 01:29:48 +02:00
$channel = new Channel($mock);
$channel->close();
$readable = $this->getMock(ReadableStreamInterface::class);
$writable = $this->getMock(WritableStreamInterface::class);
$readable->expects($this->once())
->method('close');
$writable->expects($this->once())
->method('close');
$channel = new Channel($readable, $writable);
$channel->close();
}
/**
* @expectedException \Icicle\Concurrent\Exception\InvalidArgumentError
*/
public function testReadableWithoutWritable()
{
$mock = $this->getMock(ReadableStreamInterface::class);
2015-08-03 07:20:06 +02:00
2015-09-03 01:29:48 +02:00
$channel = new Channel($mock);
2015-08-03 07:20:06 +02:00
}
public function testSendReceive()
{
Coroutine\create(function () {
list($a, $b) = Channel::createSocketPair();
$a = new Channel(new DuplexStream($a));
$b = new Channel(new DuplexStream($b));
2015-08-03 07:20:06 +02:00
yield $a->send('hello');
$data = (yield $b->receive());
2015-08-03 07:20:06 +02:00
$this->assertEquals('hello', $data);
2015-08-28 20:38:16 +02:00
$a->close();
$b->close();
})->done();
2015-08-03 07:20:06 +02:00
Loop\run();
}
2015-09-03 01:29:48 +02:00
/**
* @depends testSendReceive
* @expectedException \Icicle\Concurrent\Exception\ChannelException
*/
public function testInvalidDataReceived()
{
Coroutine\create(function () {
list($a, $b) = Channel::createSocketPair();
$a = new Channel($stream = new DuplexStream($a));
$b = new Channel(new DuplexStream($b));
// Close $a. $b should close on next read...
yield $stream->write(pack('L', 10) . '1234567890');
$data = (yield $b->receive());
})->done();
Loop\run();
}
/**
* @depends testSendReceive
* @expectedException \Icicle\Concurrent\Exception\ChannelException
*/
public function testSendUnserializableData()
{
Coroutine\create(function () {
list($a, $b) = Channel::createSocketPair();
$a = new Channel(new DuplexStream($a));
$b = new Channel(new DuplexStream($b));
// Close $a. $b should close on next read...
yield $a->send(function () {});
$data = (yield $b->receive());
})->done();
Loop\run();
}
/**
* @depends testSendReceive
* @expectedException \Icicle\Concurrent\Exception\ChannelException
*/
public function testSendAfterClose()
{
Coroutine\create(function () {
list($a, $b) = Channel::createSocketPair();
$a = new Channel(new DuplexStream($a));
$b = new Channel(new DuplexStream($b));
$a->close();
// Close $a. $b should close on next read...
yield $a->send('hello');
})->done();
Loop\run();
}
/**
* @depends testSendReceive
* @expectedException \Icicle\Concurrent\Exception\ChannelException
*/
public function testReceiveAfterClose()
{
Coroutine\create(function () {
list($a, $b) = Channel::createSocketPair();
$a = new Channel(new DuplexStream($a));
$b = new Channel(new DuplexStream($b));
$a->close();
// Close $a. $b should close on next read...
$data = (yield $a->receive());
})->done();
Loop\run();
}
2015-08-03 07:20:06 +02:00
}