Skip to content

Commit c977540

Browse files
WyriHaximusWyriHaximus
committed
Template types
1 parent bb558e1 commit c977540

File tree

8 files changed

+417
-114
lines changed

8 files changed

+417
-114
lines changed

‎composer.json‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,16 +14,26 @@
1414
"ext-parallel": "*",
1515
"react/async": "^4.2",
1616
"react/event-loop": "^1.5",
17-
"react/promise": "^2.11 || ^3.0",
18-
"reactivex/rxphp": "^2.0.12",
17+
"react/promise": "^3.1",
18+
"reactivex/rxphp": "dev-template-types as 2.0.13",
1919
"wyrihaximus/constants": "^1.6.0",
2020
"wyrihaximus/metrics": "^1.0 || ^2",
21-
"wyrihaximus/react-awaitable-observable": "^1.1",
21+
"wyrihaximus/react-awaitable-observable": "dev-add-template-types as 1.2.0",
2222
"wyrihaximus/react-event-loop-rx-scheduler-hook-up": "^0.1.1"
2323
},
2424
"require-dev": {
2525
"wyrihaximus/async-test-utilities": "^7.2"
2626
},
27+
"repositories": [
28+
{
29+
"type": "vcs",
30+
"url": "https://github.com/WyriHaximus-secret-labs/rxphp"
31+
},
32+
{
33+
"type": "vcs",
34+
"url": "https://github.com/WyriHaximus/reactphp-awaitable-observable"
35+
}
36+
],
2737
"autoload": {
2838
"psr-4": {
2939
"ReactParallel\\EventLoop\\": "src/"

‎composer.lock‎

Lines changed: 163 additions & 104 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎etc/qa/phpstan.neon‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,10 @@ parameters:
66
- ReactParallel\EventLoop\CancelledException
77
- ReactParallel\EventLoop\CanceledFuture
88
- ReactParallel\EventLoop\KilledRuntime
9+
stubFiles:
10+
- ../../stubs/Event.stub
11+
- ../../stubs/Future.stub
12+
- ../../stubs/Channel.stub
913

1014
includes:
1115
- ../../vendor/wyrihaximus/async-test-utilities/rules.neon

‎src/EventLoopBridge.php‎

Lines changed: 65 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
use parallel\Channel;
88
use parallel\Events;
9+
use parallel\Events\Event;
910
use parallel\Future;
1011
use React\EventLoop\Loop;
1112
use React\EventLoop\TimerInterface;
@@ -35,21 +36,30 @@ final class EventLoopBridge
3536

3637
private Metrics|null $metrics = null;
3738

38-
/** @var Events<Events\Event> */
39+
/**
40+
* @template T
41+
* @var Events<Events\Event<T>>
42+
*/
3943
private Events $events;
4044

4145
private TimerInterface|null $timer = null;
4246

43-
/** @var array<int, Subject> */
47+
/**
48+
* @template T
49+
* @var array<int, Subject<T>>
50+
*/
4451
private array $channels = [];
4552

46-
/** @var array<int, Deferred> */
53+
/**
54+
* @template T
55+
* @var array<int, Deferred<T>>
56+
*/
4757
private array $futures = [];
4858

4959
/** @var array<float> */
5060
private array $scaleRange = self::DEFAULT_SCALE_RANGE;
5161
private int $scalePosition = self::DEFAULT_SCALE_POSITION;
52-
private int $scaleNoItemsCount = 0;
62+
private int $scaleNoItemsCount = ZERO;
5363

5464
public function __construct()
5565
{
@@ -65,9 +75,16 @@ public function withMetrics(Metrics $metrics): self
6575
return $self;
6676
}
6777

68-
/** @return iterable<mixed> */
78+
/**
79+
* @param Channel<T> $channel
80+
*
81+
* @return iterable<T>
82+
*
83+
* @template T
84+
*/
6985
public function observe(Channel $channel): iterable
7086
{
87+
/** @var Subject<T> $subject */
7188
$subject = new Subject();
7289
$this->channels[spl_object_id($channel)] = $subject;
7390
$this->events->addChannel($channel);
@@ -81,8 +98,16 @@ public function observe(Channel $channel): iterable
8198
return awaitObservable($subject);
8299
}
83100

101+
/**
102+
* @param Future<T> $futurea
103+
*
104+
* @return T
105+
*
106+
* @template T
107+
*/
84108
public function await(Future $future): mixed
85109
{
110+
/** @var Deferred<T> $deferred */
86111
$deferred = new Deferred();
87112
$this->futures[spl_object_id($future)] = $deferred;
88113
$this->events->addFuture(spl_object_hash($future), $future);
@@ -93,7 +118,6 @@ public function await(Future $future): mixed
93118

94119
$this->startTimer();
95120

96-
/** @phpstan-ignore-next-line */
97121
return await($deferred->promise());
98122
}
99123

@@ -199,6 +223,11 @@ private function runTimer(): void
199223
});
200224
}
201225

226+
/**
227+
* @param Event<T> $event
228+
*
229+
* @template T
230+
*/
202231
private function handleReadEvent(Events\Event $event): void
203232
{
204233
if ($event->object instanceof Future) {
@@ -212,6 +241,11 @@ private function handleReadEvent(Events\Event $event): void
212241
$this->handleChannelReadEvent($event);
213242
}
214243

244+
/**
245+
* @param Events\Event<T> $event
246+
*
247+
* @template T
248+
*/
215249
private function handleFutureReadEvent(Events\Event $event): void
216250
{
217251
$this->futures[spl_object_id($event->object)]->resolve($event->value);
@@ -226,6 +260,11 @@ private function handleFutureReadEvent(Events\Event $event): void
226260
$futures->gauge(new Label('state', 'resolve'))->incr();
227261
}
228262

263+
/**
264+
* @param Events\Event<T> $event
265+
*
266+
* @template T
267+
*/
229268
private function handleChannelReadEvent(Events\Event $event): void
230269
{
231270
$this->channels[spl_object_id($event->object)]->onNext($event->value);
@@ -238,6 +277,11 @@ private function handleChannelReadEvent(Events\Event $event): void
238277
$this->metrics->channelMessages()->counter(new Label('event', 'read'))->incr();
239278
}
240279

280+
/**
281+
* @param Events\Event<T> $event
282+
*
283+
* @template T
284+
*/
241285
private function handleCloseEvent(Events\Event $event): void
242286
{
243287
$this->channels[spl_object_id($event->object)]->onCompleted();
@@ -252,6 +296,11 @@ private function handleCloseEvent(Events\Event $event): void
252296
$channels->gauge(new Label('state', 'close'))->incr();
253297
}
254298

299+
/**
300+
* @param Events\Event<T> $event
301+
*
302+
* @template T
303+
*/
255304
private function handleCancelEvent(Events\Event $event): void
256305
{
257306
$this->futures[spl_object_id($event->object)]->reject(new CanceledFuture());
@@ -266,6 +315,11 @@ private function handleCancelEvent(Events\Event $event): void
266315
$futures->gauge(new Label('state', 'cancel'))->incr();
267316
}
268317

318+
/**
319+
* @param Events\Event<T> $event
320+
*
321+
* @template T
322+
*/
269323
private function handleKillEvent(Events\Event $event): void
270324
{
271325
$this->futures[spl_object_id($event->object)]->reject(new KilledRuntime());
@@ -280,6 +334,11 @@ private function handleKillEvent(Events\Event $event): void
280334
$futures->gauge(new Label('state', 'kill'))->incr();
281335
}
282336

337+
/**
338+
* @param Events\Event<T> $event
339+
*
340+
* @template T
341+
*/
283342
private function handleErrorEvent(Events\Event $event): void
284343
{
285344
if (! ($event->object instanceof Future)) {

‎stubs/Channel.stub‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
<?php
2+
3+
namespace parallel;
4+
5+
/**
6+
* @template T
7+
*/
8+
final class Channel
9+
{
10+
/**
11+
* Constant for Infinitely Buffered
12+
*/
13+
public const Infinite = -1;
14+
15+
/* Anonymous Constructor */
16+
17+
/**
18+
* Shall make an anonymous unbuffered channel
19+
* Shall make an anonymous buffered channel with the given capacity
20+
*
21+
* @param null|int $capacity May be Channel::Infinite or a positive integer
22+
*/
23+
public function __construct(?int $capacity = null) {}
24+
25+
/* Access */
26+
27+
/**
28+
* Shall make an unbuffered channel with the given name
29+
* Shall make a buffered channel with the given name and capacity
30+
*
31+
* @param string $name The name of the channel.
32+
* @param null|int $capacity May be Channel::Infinite or a positive integer
33+
*
34+
* @return Channel<T>
35+
*
36+
* @throws Channel\Error\Existence if channel already exists.
37+
*/
38+
public static function make(string $name, ?int $capacity = null): Channel {}
39+
40+
/**
41+
* Shall open the channel with the given name
42+
*
43+
* @param string $name
44+
* @return Channel<T>
45+
*
46+
* @throws Channel\Error\Existence if channel does not exist.
47+
*/
48+
public static function open(string $name): Channel {}
49+
50+
/* Sharing */
51+
52+
/**
53+
* Shall send the given value on this channel
54+
* @param T $value
55+
*
56+
* @throws Channel\Error\Closed if channel is closed.
57+
* @throws Channel\Error\IllegalValue if value is illegal.
58+
*/
59+
public function send($value): void {}
60+
61+
/**
62+
* Shall recv a value from this channel
63+
* @return T
64+
*
65+
* @throws Channel\Error\Closed if channel is closed.
66+
*/
67+
public function recv() {}
68+
69+
/* Closing */
70+
71+
/**
72+
* Shall close this channel
73+
* @throws Channel\Error\Closed if channel is closed.
74+
*/
75+
public function close(): void {}
76+
77+
/**
78+
* Returns name of channel
79+
* @return string
80+
*/
81+
public function __toString(): string {}
82+
}

‎stubs/Event.stub‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
<?php
2+
3+
namespace parallel\Events;
4+
5+
use parallel\Channel;
6+
use parallel\Future;
7+
8+
/**
9+
* @template T
10+
*/
11+
final class Event
12+
{
13+
/**
14+
* Shall be one of Event\Type constants
15+
* @var int
16+
*/
17+
public $type;
18+
19+
/**
20+
* Shall be the source of the event (target name)
21+
* @var string
22+
*/
23+
public $source;
24+
25+
/**
26+
* Shall be either Future or Channel
27+
* @var Future<T>|Channel<T>
28+
*/
29+
public $object;
30+
31+
/**
32+
* Shall be set for Read/Error events
33+
* @var T
34+
*/
35+
public $value;
36+
}

‎stubs/Future.stub‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
<?php
2+
3+
namespace parallel;
4+
5+
use Throwable;
6+
7+
/**
8+
* @template T
9+
*/
10+
final class Future
11+
{
12+
/* Resolution */
13+
14+
/**
15+
* Shall return (and if necessary wait for) return from task
16+
*
17+
* @return T
18+
*
19+
* @throws Future\Error if waiting failed (internal error).
20+
* @throws Future\Error\Killed if \parallel\Runtime executing task was killed.
21+
* @throws Future\Error\Cancelled if task was cancelled.
22+
* @throws Future\Error\Foreign if task raised an unrecognized uncaught exception.
23+
* @throws Throwable Shall rethrow \Throwable uncaught in task
24+
*/
25+
public function value() {}
26+
27+
/* State */
28+
29+
/**
30+
* Shall indicate if the task is completed
31+
* @return bool
32+
*/
33+
public function done(): bool {}
34+
35+
/**
36+
* Shall indicate if the task was cancelled
37+
* @return bool
38+
*/
39+
public function cancelled(): bool {}
40+
41+
/* Cancellation */
42+
43+
/**
44+
* Shall try to cancel the task
45+
* Note: If task is running, it will be interrupted.
46+
* Warning: Internal function calls in progress cannot be interrupted.
47+
*
48+
* @return bool
49+
*
50+
* @throws Future\Error\Killed if \parallel\Runtime executing task was killed.
51+
* @throws Future\Error\Cancelled if task was already cancelled.
52+
*/
53+
public function cancel(): bool {}
54+
}

‎tests/EventLoopBridgeTest.php‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,6 @@ public function read(): void
7676

7777
$promises[] = resolve($eventLoopBridge->await($future));
7878

79-
/** @phpstan-ignore-next-line */
8079
$rd = await(all($promises));
8180

8281
$range = [];

0 commit comments

Comments
 (0)