Skip to content

Commit aac64de

Browse files
kafkianskykafkiansky
committed
chore: add tests for server streaming rpc
1 parent 480ad2e commit aac64de

10 files changed

Lines changed: 372 additions & 2 deletions

‎composer.json‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
"thesis/protobuf-known-types": "^0.1.2"
3737
},
3838
"require-dev": {
39+
"ext-bcmath": "*",
3940
"phpunit/phpunit": "^12.4",
4041
"symfony/var-dumper": "^8.0"
4142
},
@@ -49,8 +50,16 @@
4950
},
5051
"autoload-dev": {
5152
"psr-4": {
52-
"Thesis\\Grpc\\": "tests/"
53-
}
53+
"Echos\\Api\\V1\\": "tests/genproto/Echos/Api/V1/",
54+
"File\\Api\\V1\\": "tests/genproto/File/Api/V1/",
55+
"Thesis\\Grpc\\": "tests/",
56+
"Topic\\Api\\V1\\": "tests/genproto/Topic/Api/V1/"
57+
},
58+
"files": [
59+
"tests/genproto/Echos/Api/V1/autoload.metadata.php",
60+
"tests/genproto/File/Api/V1/autoload.metadata.php",
61+
"tests/genproto/Topic/Api/V1/autoload.metadata.php"
62+
]
5463
},
5564
"config": {
5665
"lock": false,

‎tests/ServerStreamTest.php‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Thesis\Grpc;
6+
7+
use Amp\Cancellation;
8+
use BcMath\Number;
9+
use Google\Protobuf\Timestamp;
10+
use Google\Rpc\Code;
11+
use PHPUnit\Framework\Attributes\CoversClass;
12+
use PHPUnit\Framework\TestCase;
13+
use Thesis\Grpc\Server\ServerStreamHandler;
14+
use Topic\Api\V1\Event;
15+
use Topic\Api\V1\SubscribeRequest;
16+
use Topic\Api\V1\TopicServiceClient;
17+
use Topic\Api\V1\TopicServiceServer;
18+
use Topic\Api\V1\TopicServiceServerRegistry;
19+
20+
/**
21+
* @api
22+
*/
23+
#[CoversClass(Server::class)]
24+
#[CoversClass(Client::class)]
25+
#[CoversClass(ServerStreamHandler::class)]
26+
final class ServerStreamTest extends TestCase
27+
{
28+
private Server $server;
29+
30+
protected function setUp(): void
31+
{
32+
parent::setUp();
33+
34+
$this->server = new Server\Builder()
35+
->withServices(new TopicServiceServerRegistry(new TopicServer()))
36+
->build();
37+
38+
$this->server->start();
39+
}
40+
41+
protected function tearDown(): void
42+
{
43+
parent::tearDown();
44+
45+
$this->server->stop();
46+
}
47+
48+
public function testEventReceived(): void
49+
{
50+
$client = new TopicServiceClient(new Client\Builder()->build());
51+
52+
$stream = $client->subscribe(new SubscribeRequest('payments'));
53+
54+
$paymentEvents = [];
55+
foreach ($stream as $event) {
56+
$paymentEvents[] = $event;
57+
}
58+
59+
self::assertCount(2, $paymentEvents);
60+
self::assertEquals(
61+
[
62+
new Event('payment_finished', '{"id": 1}', new Timestamp(new Number(1_771_782_096))),
63+
new Event('payment_rejected', '{"id": 2}', new Timestamp(new Number(1_771_782_097))),
64+
],
65+
$paymentEvents,
66+
);
67+
68+
$stream = $client->subscribe(new SubscribeRequest('subscriptions'));
69+
70+
$subscriptionEvents = [];
71+
foreach ($stream as $event) {
72+
$subscriptionEvents[] = $event;
73+
}
74+
75+
self::assertCount(1, $subscriptionEvents);
76+
self::assertEquals(
77+
[
78+
new Event('subscription_terminated', '{"id": 1}', new Timestamp(new Number(1_771_782_099))),
79+
],
80+
$subscriptionEvents,
81+
);
82+
83+
$this->expectExceptionObject(new InvokeError(Code::FAILED_PRECONDITION, 'Unknown topic "users"'));
84+
$stream = $client->subscribe(new SubscribeRequest('users'));
85+
iterator_to_array($stream);
86+
}
87+
}
88+
89+
final readonly class TopicServer implements TopicServiceServer
90+
{
91+
#[\Override]
92+
public function subscribe(
93+
SubscribeRequest $request,
94+
Server\ServerStreamChannel $stream,
95+
Metadata $md,
96+
Cancellation $cancellation,
97+
): void {
98+
$events = [
99+
'payments' => [
100+
new Event('payment_finished', '{"id": 1}', new Timestamp(new Number(1_771_782_096))),
101+
new Event('payment_rejected', '{"id": 2}', new Timestamp(new Number(1_771_782_097))),
102+
],
103+
'subscriptions' => [
104+
new Event('subscription_terminated', '{"id": 1}', new Timestamp(new Number(1_771_782_099))),
105+
],
106+
];
107+
108+
foreach ($events[$request->topic] ?? throw new InvokeError(Code::FAILED_PRECONDITION, 'Unknown topic "' . $request->topic . '"') as $it) {
109+
$stream->send($it);
110+
}
111+
112+
$stream->close();
113+
}
114+
}

‎tests/genproto/Topic/Api/V1/Event.php‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
<?php
2+
3+
/**
4+
* Code generated by thesis/protoc-plugin. DO NOT EDIT.
5+
* Versions:
6+
* thesis/protoc-plugin — v0.1.7
7+
* protoc — v6.33.5
8+
* Source: tests/protos/topic_v1.proto
9+
*/
10+
11+
declare(strict_types=1);
12+
13+
namespace Topic\Api\V1;
14+
15+
use Thesis\Protobuf\Reflection;
16+
17+
/**
18+
* @api
19+
*/
20+
final readonly class Event
21+
{
22+
public function __construct(
23+
#[Reflection\Field(1, Reflection\StringT::T)]
24+
public string $type = '',
25+
#[Reflection\Field(2, Reflection\StringT::T)]
26+
public string $payload = '',
27+
#[Reflection\Field(3, new Reflection\ObjectT(\Google\Protobuf\Timestamp::class))]
28+
public ?\Google\Protobuf\Timestamp $ts = null,
29+
) {}
30+
}

‎tests/genproto/Topic/Api/V1/SubscribeRequest.php‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
<?php
2+
3+
/**
4+
* Code generated by thesis/protoc-plugin. DO NOT EDIT.
5+
* Versions:
6+
* thesis/protoc-plugin — v0.1.7
7+
* protoc — v6.33.5
8+
* Source: tests/protos/topic_v1.proto
9+
*/
10+
11+
declare(strict_types=1);
12+
13+
namespace Topic\Api\V1;
14+
15+
use Thesis\Protobuf\Reflection;
16+
17+
/**
18+
* @api
19+
*/
20+
final readonly class SubscribeRequest
21+
{
22+
public function __construct(
23+
#[Reflection\Field(1, Reflection\StringT::T)]
24+
public string $topic = '',
25+
) {}
26+
}

‎tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
<?php
2+
3+
/**
4+
* Code generated by thesis/protoc-plugin. DO NOT EDIT.
5+
* Versions:
6+
* thesis/protoc-plugin — v0.1.7
7+
* protoc — v6.33.5
8+
* Source: tests/protos/topic_v1.proto
9+
*/
10+
11+
declare(strict_types=1);
12+
13+
namespace Topic\Api\V1;
14+
15+
use Override;
16+
use Thesis\Protobuf\Pool;
17+
18+
/**
19+
* @api
20+
*/
21+
final readonly class TestsProtosTopicV1DescriptorRegistry implements Pool\Registrar
22+
{
23+
private const string DESCRIPTOR_BUFFER = 'Cht0ZXN0cy9wcm90b3MvdG9waWNfdjEucHJvdG8SDHRvcGljLmFwaS52MRofZ29vZ2xlL3Byb3RvYnVmL3RpbWVzdGFtcC5wcm90byIoChBTdWJzY3JpYmVSZXF1ZXN0EhQKBXRvcGljGAEgASgJUgV0b3BpYyJhCgVFdmVudBISCgR0eXBlGAEgASgJUgR0eXBlEhgKB3BheWxvYWQYAiABKAlSB3BheWxvYWQSKgoCdHMYAyABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wUgJ0czJSCgxUb3BpY1NlcnZpY2USQgoJU3Vic2NyaWJlEh4udG9waWMuYXBpLnYxLlN1YnNjcmliZVJlcXVlc3QaEy50b3BpYy5hcGkudjEuRXZlbnQwAUqQAwoGEgQAABIBCggKAQwSAwAAEgoJCgIDABIDAgApCggKAQISAwQAFQoKCgIEABIEBgAIAQoKCgMEAAESAwYIGAoLCgQEAAIAEgMHBBUKDAoFBAACAAUSAwcECgoMCgUEAAIAARIDBwsQCgwKBQQAAgADEgMHExQKCgoCBAESBAoADgEKCgoDBAEBEgMKCA0KCwoEBAECABIDCwQUCgwKBQQBAgAFEgMLBAoKDAoFBAECAAESAwsLDwoMCgUEAQIAAxIDCxITCgsKBAQBAgESAwwEFwoMCgUEAQIBBRIDDAQKCgwKBQQBAgEBEgMMCxIKDAoFBAECAQMSAwwVFgoLCgQEAQICEgMNBCUKDAoFBAECAgYSAw0EHQoMCgUEAQICARIDDR4gCgwKBQQBAgIDEgMNIyQKCgoCBgASBBAAEgEKCgoDBgABEgMQCBQKCwoEBgACABIDEQQ7CgwKBQYAAgABEgMRCBEKDAoFBgACAAISAxESIgoMCgUGAAIABhIDES0zCgwKBQYAAgADEgMRNDliBnByb3RvMw==';
24+
25+
#[Override]
26+
public function register(Pool\Registry $pool): void
27+
{
28+
$pool->add(Pool\Descriptor::base64(self::DESCRIPTOR_BUFFER), [
29+
'topic.api.v1.SubscribeRequest' => new Pool\MessageMetadata(\Topic\Api\V1\SubscribeRequest::class),
30+
'topic.api.v1.Event' => new Pool\MessageMetadata(\Topic\Api\V1\Event::class),
31+
]);
32+
}
33+
}

‎tests/genproto/Topic/Api/V1/TopicServiceClient.php‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
<?php
2+
3+
/**
4+
* Code generated by thesis/protoc-plugin. DO NOT EDIT.
5+
* Versions:
6+
* thesis/protoc-plugin — v0.1.7
7+
* protoc — v6.33.5
8+
* Source: tests/protos/topic_v1.proto
9+
*/
10+
11+
declare(strict_types=1);
12+
13+
namespace Topic\Api\V1;
14+
15+
use Amp\Cancellation;
16+
use Amp\NullCancellation;
17+
use Thesis\Grpc\Client;
18+
use Thesis\Grpc\Metadata;
19+
20+
/**
21+
* @api
22+
*/
23+
final readonly class TopicServiceClient
24+
{
25+
public function __construct(
26+
private Client $client,
27+
) {}
28+
29+
/**
30+
* @return Client\ServerStreamChannel<\Topic\Api\V1\SubscribeRequest, \Topic\Api\V1\Event>
31+
*/
32+
public function subscribe(
33+
\Topic\Api\V1\SubscribeRequest $request,
34+
Metadata $md = new Metadata(),
35+
Cancellation $cancellation = new NullCancellation(),
36+
): Client\ServerStreamChannel {
37+
/** @var Client\Invoke<\Topic\Api\V1\SubscribeRequest, \Topic\Api\V1\Event> $invoke */
38+
$invoke = new Client\Invoke(
39+
method: '/topic.api.v1.TopicService/Subscribe',
40+
type: \Topic\Api\V1\Event::class,
41+
);
42+
43+
$stream = $this->client->createStream(
44+
invoke: $invoke,
45+
md: $md,
46+
cancellation: $cancellation,
47+
);
48+
49+
$stream->send($request);
50+
$stream->close();
51+
52+
return new Client\ServerStreamChannel($stream);
53+
}
54+
}

‎tests/genproto/Topic/Api/V1/TopicServiceServer.php‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
<?php
2+
3+
/**
4+
* Code generated by thesis/protoc-plugin. DO NOT EDIT.
5+
* Versions:
6+
* thesis/protoc-plugin — v0.1.7
7+
* protoc — v6.33.5
8+
* Source: tests/protos/topic_v1.proto
9+
*/
10+
11+
declare(strict_types=1);
12+
13+
namespace Topic\Api\V1;
14+
15+
use Amp\Cancellation;
16+
use Thesis\Grpc\Metadata;
17+
use Thesis\Grpc\Server;
18+
19+
/**
20+
* @api
21+
*/
22+
interface TopicServiceServer
23+
{
24+
/**
25+
* @param Server\ServerStreamChannel<\Topic\Api\V1\SubscribeRequest, \Topic\Api\V1\Event> $stream
26+
*/
27+
public function subscribe(
28+
\Topic\Api\V1\SubscribeRequest $request,
29+
Server\ServerStreamChannel $stream,
30+
Metadata $md,
31+
Cancellation $cancellation,
32+
): void;
33+
}

‎tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
<?php
2+
3+
/**
4+
* Code generated by thesis/protoc-plugin. DO NOT EDIT.
5+
* Versions:
6+
* thesis/protoc-plugin — v0.1.7
7+
* protoc — v6.33.5
8+
* Source: tests/protos/topic_v1.proto
9+
*/
10+
11+
declare(strict_types=1);
12+
13+
namespace Topic\Api\V1;
14+
15+
use Override;
16+
use Thesis\Grpc\Server;
17+
18+
/**
19+
* @api
20+
*/
21+
final readonly class TopicServiceServerRegistry implements Server\ServiceRegistry
22+
{
23+
public function __construct(
24+
private \Topic\Api\V1\TopicServiceServer $server,
25+
) {}
26+
27+
#[Override]
28+
public function services(): iterable
29+
{
30+
yield new Server\Service('topic.api.v1.TopicService', [
31+
new Server\Rpc(
32+
new Server\Handle('Subscribe', \Topic\Api\V1\SubscribeRequest::class),
33+
new Server\ServerStreamHandler($this->server->subscribe(...)),
34+
Server\RpcType::ServerStream,
35+
),
36+
]);
37+
}
38+
}

‎tests/genproto/Topic/Api/V1/autoload.metadata.php‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
<?php
2+
3+
/**
4+
* Code generated by thesis/protoc-plugin. DO NOT EDIT.
5+
* Versions:
6+
* thesis/protoc-plugin — v0.1.7
7+
* protoc — v6.33.5
8+
*/
9+
10+
declare(strict_types=1);
11+
12+
\Thesis\Protobuf\Pool\Registry::get()->register(
13+
new \Thesis\Protobuf\Pool\OnceRegistrar(new \Topic\Api\V1\TestsProtosTopicV1DescriptorRegistry()),
14+
);

‎tests/protos/topic_v1.proto‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
syntax = "proto3";
2+
3+
import "google/protobuf/timestamp.proto";
4+
5+
package topic.api.v1;
6+
7+
message SubscribeRequest {
8+
string topic = 1;
9+
}
10+
11+
message Event {
12+
string type = 1;
13+
string payload = 2;
14+
google.protobuf.Timestamp ts = 3;
15+
}
16+
17+
service TopicService {
18+
rpc Subscribe(SubscribeRequest) returns (stream Event);
19+
}

0 commit comments

Comments
 (0)