Skip to content

Commit 23a5a61

Browse files
azjezzazjezz
authored
feat(channel): introduce new channel component (#283)
Signed-off-by: azjezz <azjezz@protonmail.com>
1 parent 8595a53 commit 23a5a61

24 files changed

+1136
-4
lines changed

‎CHANGELOG.md

+1
Original file line numberDiff line numberDiff line change
@@ -23,3 +23,4 @@
2323
* introduced a new `Psl\Network` component.
2424
* introduced a new `Psl\TCP` component.
2525
* introduced a new `Psl\Unix` component.
26+
* introduced a new `Psl\Channel` component.

‎docs/README.md

+1
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88

99
- [Psl](./component/psl.md)
1010
- [Psl\Async](./component/async.md)
11+
- [Psl\Channel](./component/channel.md)
1112
- [Psl\Class](./component/class.md)
1213
- [Psl\Collection](./component/collection.md)
1314
- [Psl\DataStructure](./component/data-structure.md)

‎docs/component/channel.md

+23
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
<!--
2+
This markdown file was generated using `docs/documenter.php`.
3+
4+
Any edits to it will likely be lost.
5+
-->
6+
7+
[*index](./../README.md)
8+
9+
---
10+
11+
### `Psl\Channel` Component
12+
13+
#### `Functions`
14+
15+
- [bounded](./../../src/Psl/Channel/bounded.php#L18)
16+
- [unbounded](./../../src/Psl/Channel/unbounded.php#L16)
17+
18+
#### `Interfaces`
19+
20+
- [ReceiverInterface](./../../src/Psl/Channel/ReceiverInterface.php#L12)
21+
- [SenderInterface](./../../src/Psl/Channel/SenderInterface.php#L12)
22+
23+

‎docs/component/data-structure.md

+2-2
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,8 @@
1818

1919
#### `Classes`
2020

21-
- [PriorityQueue](./../../src/Psl/DataStructure/PriorityQueue.php#L18)
22-
- [Queue](./../../src/Psl/DataStructure/Queue.php#L19)
21+
- [PriorityQueue](./../../src/Psl/DataStructure/PriorityQueue.php#L20)
22+
- [Queue](./../../src/Psl/DataStructure/Queue.php#L21)
2323
- [Stack](./../../src/Psl/DataStructure/Stack.php#L19)
2424

2525

‎docs/documenter.php

+1
Original file line numberDiff line numberDiff line change
@@ -184,6 +184,7 @@ function get_all_components(): array
184184
$components = [
185185
'Psl',
186186
'Psl\\Async',
187+
'Psl\\Channel',
187188
'Psl\\Class',
188189
'Psl\\Collection',
189190
'Psl\\DataStructure',

‎examples/channel/main.php

+25
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Psl\Example\IO;
6+
7+
use Psl\Async;
8+
use Psl\Channel;
9+
use Psl\IO;
10+
11+
require __DIR__ . '/../../vendor/autoload.php';
12+
13+
/**
14+
* @var Channel\ReceiverInterface<string> $receiver
15+
* @var Channel\SenderInterface<string> $sender
16+
*/
17+
[$receiver, $sender] = Channel\unbounded();
18+
19+
Async\Scheduler::delay(1, static function () use ($sender) {
20+
$sender->send('Hello, World!');
21+
});
22+
23+
$message = $receiver->receive();
24+
25+
IO\output_handle()->writeAll($message);

‎src/Psl/Channel/ChannelInterface.php

+59
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Psl\Channel;
6+
7+
use Countable;
8+
9+
/**
10+
* @template T
11+
*/
12+
interface ChannelInterface extends Countable
13+
{
14+
/**
15+
* Returns the channel capacity if it’s bounded.
16+
*
17+
* @mutation-free
18+
*/
19+
public function getCapacity(): ?int;
20+
21+
/**
22+
* Closes the channel.
23+
*
24+
* The remaining messages can still be received.
25+
*/
26+
public function close(): void;
27+
28+
/**
29+
* Returns true if the channel is closed.
30+
*
31+
* @mutation-free
32+
*/
33+
public function isClosed(): bool;
34+
35+
/**
36+
* Returns the number of messages in the channel.
37+
*
38+
* @return positive-int|0
39+
*
40+
* @mutation-free
41+
*/
42+
public function count(): int;
43+
44+
/**
45+
* Returns true if the channel is full.
46+
*
47+
* Unbounded channels are never full.
48+
*
49+
* @mutation-free
50+
*/
51+
public function isFull(): bool;
52+
53+
/**
54+
* Returns true if the channel is empty.
55+
*
56+
* @mutation-free
57+
*/
58+
public function isEmpty(): bool;
59+
}

‎src/Psl/Channel/Exception/ClosedChannelException.php

+34
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Psl\Channel\Exception;
6+
7+
use Psl\Channel;
8+
use Psl\Exception\RuntimeException;
9+
10+
/**
11+
* This exception is thrown when attempting to send or receive a message on a closed channel.
12+
*
13+
* @see Channel\SenderInterface::send()
14+
* @see Channel\SenderInterface::trySend()
15+
* @see Channel\ReceiverInterface::receive()
16+
* @see Channel\ReceiverInterface::tryReceive()
17+
*/
18+
final class ClosedChannelException extends RuntimeException implements ExceptionInterface
19+
{
20+
public function __construct(string $message = 'Channel has been closed')
21+
{
22+
parent::__construct($message);
23+
}
24+
25+
public static function forSending(): ClosedChannelException
26+
{
27+
return new self('Attempted to send a message to a closed channel.');
28+
}
29+
30+
public static function forReceiving(): ClosedChannelException
31+
{
32+
return new self('Attempted to receive a message from a closed empty channel.');
33+
}
34+
}

‎src/Psl/Channel/Exception/EmptyChannelException.php

+19
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Psl\Channel\Exception;
6+
7+
use OutOfBoundsException;
8+
use Psl\Channel;
9+
10+
/**
11+
* This exception is throw when calling {@see Channel\ReceiverInterface::tryReceive()} on an empty channel.
12+
*/
13+
final class EmptyChannelException extends OutOfBoundsException implements ExceptionInterface
14+
{
15+
public static function create(): EmptyChannelException
16+
{
17+
return new self('Attempted to receiver from an empty channel.');
18+
}
19+
}

‎src/Psl/Channel/Exception/ExceptionInterface.php

+11
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Psl\Channel\Exception;
6+
7+
use Psl\Exception;
8+
9+
interface ExceptionInterface extends Exception\ExceptionInterface
10+
{
11+
}

‎src/Psl/Channel/Exception/FullChannelException.php

+20
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Psl\Channel\Exception;
6+
7+
use OutOfBoundsException;
8+
use Psl\Channel;
9+
use Psl\Str;
10+
11+
/**
12+
* This exception is throw when calling {@see Channel\SenderInterface::trySend()} on a full channel.
13+
*/
14+
final class FullChannelException extends OutOfBoundsException implements ExceptionInterface
15+
{
16+
public static function ofCapacity(int $capacity): FullChannelException
17+
{
18+
return new self(Str\format('Channel has reached its full capacity of %d.', $capacity));
19+
}
20+
}

‎src/Psl/Channel/Internal/ChannelState.php

+128
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Psl\Channel\Internal;
6+
7+
use Psl\Channel\ChannelInterface;
8+
use Psl\Channel\Exception;
9+
use Psl\DataStructure\Queue;
10+
use Psl\DataStructure\QueueInterface;
11+
12+
/**
13+
* @template T
14+
*
15+
* @implements ChannelInterface<T>
16+
*
17+
* @internal
18+
*/
19+
final class ChannelState implements ChannelInterface
20+
{
21+
/**
22+
* @var QueueInterface<T>
23+
*/
24+
private QueueInterface $messages;
25+
26+
private bool $closed = false;
27+
28+
public function __construct(
29+
private ?int $capacity = null,
30+
) {
31+
$this->messages = new Queue();
32+
}
33+
34+
/**
35+
* {@inheritDoc}
36+
*/
37+
public function getCapacity(): ?int
38+
{
39+
return $this->capacity;
40+
}
41+
42+
/**
43+
* {@inheritDoc}
44+
*/
45+
public function close(): void
46+
{
47+
$this->closed = true;
48+
}
49+
50+
/**
51+
* {@inheritDoc}
52+
*/
53+
public function isClosed(): bool
54+
{
55+
return $this->closed;
56+
}
57+
58+
/**
59+
* {@inheritDoc}
60+
*/
61+
public function count(): int
62+
{
63+
return $this->messages->count();
64+
}
65+
66+
/**
67+
* {@inheritDoc}
68+
*/
69+
public function isFull(): bool
70+
{
71+
if (null === $this->capacity) {
72+
return false;
73+
}
74+
75+
return $this->capacity === $this->count();
76+
}
77+
78+
/**
79+
* {@inheritDoc}
80+
*/
81+
public function isEmpty(): bool
82+
{
83+
return 0 === $this->messages->count();
84+
}
85+
86+
/**
87+
* @param T $message
88+
*
89+
* @throws Exception\ClosedChannelException If the channel is closed.
90+
* @throws Exception\FullChannelException If the channel is full.
91+
*/
92+
public function send(mixed $message): void
93+
{
94+
if ($this->closed) {
95+
throw Exception\ClosedChannelException::forSending();
96+
}
97+
98+
if (null === $this->capacity || $this->capacity > $this->count()) {
99+
$this->messages->enqueue($message);
100+
101+
return;
102+
}
103+
104+
throw Exception\FullChannelException::ofCapacity($this->capacity);
105+
}
106+
107+
/**
108+
* @throws Exception\ClosedChannelException If the channel is closed, and there's no more messages to receive.
109+
* @throws Exception\EmptyChannelException If the channel is empty.
110+
*
111+
* @return T
112+
*/
113+
public function receive(): mixed
114+
{
115+
$empty = 0 === $this->count();
116+
$closed = $this->closed;
117+
if ($closed && $empty) {
118+
throw Exception\ClosedChannelException::forReceiving();
119+
}
120+
121+
if ($empty) {
122+
throw Exception\EmptyChannelException::create();
123+
}
124+
125+
/** @psalm-suppress MissingThrowsDocblock */
126+
return $this->messages->dequeue();
127+
}
128+
}

0 commit comments

Comments
 (0)