Repository navigation
Expand file tree
/
Copy pathJetStreamContextTest.php
More file actions
10380 lines (8852 loc) · 478 KB
/
Copy pathJetStreamContextTest.php
File metadata and controls
10380 lines (8852 loc) · 478 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
<?php
declare(strict_types=1);
namespace IDCT\NATS\Tests\Unit;
use DateTimeImmutable;
use DateTimeZone;
use IDCT\NATS\Connection\Enum\ConnectionState;
use IDCT\NATS\Connection\NatsOptions;
use IDCT\NATS\Core\NatsClient;
use IDCT\NATS\Core\NatsHeaders;
use IDCT\NATS\Core\NatsMessage;
use IDCT\NATS\Exception\JetStreamException;
use IDCT\NATS\Exception\NatsException;
use IDCT\NATS\Exception\TimeoutException;
use IDCT\NATS\Exception\UnsupportedFeatureException;
use IDCT\NATS\JetStream\Configuration\ConsumerConfiguration;
use IDCT\NATS\JetStream\Configuration\StreamConfiguration;
use IDCT\NATS\JetStream\Configuration\StreamSource;
use IDCT\NATS\JetStream\Consumers\PullConsumerIterator;
use IDCT\NATS\JetStream\HeartbeatWatchdogState;
use IDCT\NATS\JetStream\JetStreamContext;
use IDCT\NATS\JetStream\KeyValue\KeyValueBucket;
use IDCT\NATS\JetStream\Models\StreamInfo;
use IDCT\NATS\JetStream\Schedule;
use IDCT\NATS\Tests\Support\FakeTransport;
use PHPUnit\Framework\TestCase;
use Revolt\EventLoop;
use Revolt\EventLoop\CallbackType;
use function Amp\delay;
final class JetStreamContextTest extends TestCase
{
private const INFO_2_12 = 'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n";
/**
* A subscription's idle-heartbeat watchdog (#113) is a live EventLoop timer that outlives a test
* whose client is never disconnect()ed. Cancel every callback left registered so the shared loop
* (Revolt keeps one per process) starts each test clean and a leaked watchdog cannot fire against
* a lingering client from an earlier test.
*/
protected function tearDown(): void
{
foreach (EventLoop::getIdentifiers() as $id) {
EventLoop::cancel($id);
}
}
/**
* Counts EventLoop repeat timers currently registered (the watchdog is one). Used by the
* teardown-leak regression to probe watchdog arm/cancel without reaching into private state.
*/
private static function countRepeatTimers(): int
{
$count = 0;
foreach (EventLoop::getIdentifiers() as $id) {
if (EventLoop::getType($id) === CallbackType::Repeat) {
$count++;
}
}
return $count;
}
private function jsOkResponse(string $json): string
{
return sprintf("MSG _INBOX.any 1 %d\r\n%s\r\n", strlen($json), $json);
}
/**
* Post-#118 the request inbox is MUXED: request()/requestWithHeaders() no longer subscribe a fresh
* inbox per call; instead ONE long-lived wildcard "<base>.*" (sid 1) serves every reply, and each
* request publishes reply-to "<base>.<token>" and routes by that token. A pre-seeded
* "MSG _INBOX.x <sid>" therefore no longer reaches the waiting request (its subject is not the
* request's random "<base>.<token>", so dispatchMuxReply drops it and the request times out).
*
* This installs a dynamic responder that mirrors a real server: it learns the mux base from the
* wildcard SUB, and for each request PUB/HPUB (a publish whose reply-to lives under that base)
* pops the next reply frame and re-emits it on the CAPTURED reply-to with the mux sid (1). Frames
* are delivered FIFO in the order requests are written, so a method that issues several requests
* (e.g. create->update, create->info->delete) still gets its replies in sequence.
*
* Only request replies belong here; subscription deliveries (a push consumer's deliver subject,
* a pull fetch inbox) are NOT mux replies - keep those pre-seeded, or enqueue them on their SUB via
* FakeTransport::$enqueueOnWriteContaining so they arrive after the deliver subscription exists.
*
* @param list<string> $frames Pre-built MSG/HMSG reply frames; only their subject+sid are rewritten.
*/
private function muxReplies(FakeTransport $transport, array $frames): void
{
$queue = $frames;
$base = null;
$transport->onWrite = static function (string $bytes) use (&$queue, &$base): array {
$head = strtok($bytes, "\r\n");
if ($head === false) {
return [];
}
if (str_starts_with($head, 'SUB ')) {
// Learn the mux base from the wildcard registration "SUB <base>.* <sid>".
$subject = explode(' ', $head)[1] ?? '';
if (str_starts_with($subject, '_INBOX.') && str_ends_with($subject, '.*')) {
$base = substr($subject, 0, -1); // strip the trailing "*" -> "_INBOX.<hex>."
}
return [];
}
if ($base === null || (!str_starts_with($head, 'PUB ') && !str_starts_with($head, 'HPUB '))) {
return [];
}
// PUB <subject> <replyTo> <len> | HPUB <subject> <replyTo> <hdrLen> <totLen>
$replyTo = explode(' ', $head)[2] ?? '';
if (!str_starts_with($replyTo, $base) || $queue === []) {
// A plain publish, or a subscription's own inbox (fetch/direct-get) - not a mux request.
return [];
}
return [self::rewriteReplyFrame(array_shift($queue), $replyTo)];
};
}
/**
* Rewrites a pre-built MSG/HMSG reply frame's subject and sid to the request's captured mux reply-to
* and the mux sid (1), preserving the payload/header bytes and declared lengths verbatim so the
* reply's content is delivered unchanged - only its addressing is corrected for the mux inbox.
*/
private static function rewriteReplyFrame(string $frame, string $replyTo): string
{
$pos = strpos($frame, "\r\n");
self::assertNotFalse($pos, 'reply frame must have a head line');
$tokens = explode(' ', substr($frame, 0, $pos));
$tokens[1] = $replyTo; // subject
$tokens[2] = '1'; // mux sid
return implode(' ', $tokens) . substr($frame, $pos);
}
/**
* Extracts the mux reply-to (the 3rd token) from a written PUB/HPUB request frame, or '' when the
* frame is not a request publish. Lets a dynamic onWrite responder echo a reply on the request's
* actual reply-to (the mux "<base>.<token>") instead of a guessed inbox subject (#118).
*/
private static function requestReplyTo(string $bytes): string
{
$head = strtok($bytes, "\r\n");
if ($head === false || (!str_starts_with($head, 'PUB ') && !str_starts_with($head, 'HPUB '))) {
return '';
}
return explode(' ', $head)[2] ?? '';
}
/**
* Builds a mux reply frame (MSG on the given reply-to, mux sid 1) carrying $payload - the wire form
* the server sends back on the shared request inbox (#118).
*/
private static function muxMsg(string $replyTo, string $payload): string
{
return sprintf("MSG %s 1 %d\r\n%s\r\n", $replyTo, strlen($payload), $payload);
}
/**
* Installs a mux-aware server for ordered/push consumer recreate tests. Pre-#118 each request owned
* a per-request SUB, so a CONSUMER.CREATE/DELETE reply routed by a fixed sid and the rotated deliver
* inbox landed at a sid shifted by those per-request subs (2,4,7,...). Post-#118 all requests share
* the mux inbox (no per-request SUB), so: (a) CREATE/DELETE replies must be echoed on the request's
* captured reply-to, and (b) the rotated deliver sid collapses to the next free sid (2,3,4,...). This
* responder removes both hazards by construction: request replies go on the captured reply-to, and
* each deliver epoch's frames are emitted on the deliver SUB using the sid the server ACTUALLY
* assigned (captured live), so tests never hard-code a sid.
*
* @param list<callable(string):list<string>> $onCreate FIFO, one per CONSUMER.CREATE PUB; gets the reply-to.
* @param list<callable(string):list<string>> $onDelete FIFO, one per CONSUMER.DELETE PUB; gets the reply-to.
* @param list<callable(int):list<string>> $deliverEpochs FIFO, one per deliver SUB; gets the captured sid.
*/
private function orderedConsumerServer(
FakeTransport $transport,
array $onCreate,
array $onDelete = [],
array $deliverEpochs = [],
): void {
// The callback arrays are captured BY VALUE (so PHPStan keeps their callable return types); FIFO
// consumption is tracked by the by-reference cursors below.
$createIdx = 0;
$deleteIdx = 0;
$deliverIdx = 0;
$transport->onWrite = static function (string $bytes) use (
$onCreate,
$onDelete,
$deliverEpochs,
&$createIdx,
&$deleteIdx,
&$deliverIdx,
): array {
$head = strtok($bytes, "\r\n");
if ($head === false) {
return [];
}
if (str_starts_with($head, 'SUB ')) {
$parts = explode(' ', $head);
$subject = $parts[1] ?? '';
// The mux wildcard SUB ends with ".*"; every other SUB here is a (rotated) deliver inbox.
$epoch = $deliverEpochs[$deliverIdx] ?? null;
if ($epoch !== null && !str_ends_with($subject, '.*')) {
$deliverIdx++;
return $epoch((int) ($parts[2] ?? 0));
}
return [];
}
$replyTo = self::requestReplyTo($bytes);
if ($replyTo === '') {
return [];
}
$create = $onCreate[$createIdx] ?? null;
if ($create !== null && str_contains($bytes, '$JS.API.CONSUMER.CREATE.')) {
$createIdx++;
return $create($replyTo);
}
$delete = $onDelete[$deleteIdx] ?? null;
if ($delete !== null && str_contains($bytes, '$JS.API.CONSUMER.DELETE.')) {
$deleteIdx++;
return $delete($replyTo);
}
return [];
};
}
/**
* Verifies accountInfo() returns parsed account metrics.
*/
public function testAccountInfo(): void
{
$accountPayload = '{"memory":11,"storage":22,"streams":3,"consumers":4}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.any 1 %d\r\n%s\r\n", strlen($accountPayload), $accountPayload),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$account = $client->jetStream()->accountInfo()->await();
self::assertSame(11, $account->memory);
self::assertSame(22, $account->storage);
self::assertStringStartsWith('PUB $JS.API.INFO _INBOX.', $transport->writes[3]);
}
/**
* Verifies addStream() builds the CREATE payload from a typed StreamConfiguration (#53).
*/
public function testAddStreamFromBuilder(): void
{
$reply = '{"config":{"name":"ORDERS","subjects":["orders.*"]}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($reply), $reply),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$config = \IDCT\NATS\JetStream\Configuration\StreamConfiguration::create('ORDERS')
->subjects('orders.*', 'orders.archive')
->retention(\IDCT\NATS\JetStream\Enum\RetentionPolicy::WorkQueue)
->storage(\IDCT\NATS\JetStream\Enum\StorageBackend::Memory)
->maxBytes(4096)
->maxAge(60)
->replicas(3);
$info = $client->jetStream()->addStream($config)->await();
self::assertSame('ORDERS', $info->name);
$create = $transport->writes[3];
self::assertStringContainsString('$JS.API.STREAM.CREATE.ORDERS', $create);
self::assertStringContainsString('"subjects":["orders.*","orders.archive"]', $create);
self::assertStringContainsString('"retention":"workqueue"', $create);
self::assertStringContainsString('"storage":"memory"', $create);
self::assertStringContainsString('"max_bytes":4096', $create);
self::assertStringContainsString('"max_age":60000000000', $create);
self::assertStringContainsString('"num_replicas":3', $create);
}
/**
* Verifies addConsumer() builds the CREATE payload from a typed ConsumerConfiguration (#54).
*/
public function testAddConsumerFromBuilder(): void
{
$reply = '{"stream_name":"ORDERS","name":"worker","config":{"durable_name":"worker"}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($reply), $reply),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$config = \IDCT\NATS\JetStream\Configuration\ConsumerConfiguration::create()
->durable('worker')
->ackPolicy(\IDCT\NATS\JetStream\Enum\AckPolicy::Explicit)
->maxDeliver(5)
->ackWait(1000)
->backoff([1000, 2000]);
$info = $client->jetStream()->addConsumer('ORDERS', $config)->await();
self::assertSame('worker', $info->name);
$create = $transport->writes[3];
self::assertStringContainsString('$JS.API.CONSUMER.CREATE.ORDERS.worker', $create);
self::assertStringContainsString('"durable_name":"worker"', $create);
self::assertStringContainsString('"ack_policy":"explicit"', $create);
self::assertStringContainsString('"max_deliver":5', $create);
self::assertStringContainsString('"ack_wait":1000000000', $create);
self::assertStringContainsString('"backoff":[1000000000,2000000000]', $create);
}
/**
* Verifies keyValueBucketNames() lists KV_-prefixed streams with the prefix stripped (#60).
*/
public function testKeyValueBucketNames(): void
{
$reply = '{"streams":["KV_cfg","OBJ_assets","ORDERS","KV_sessions"],"total":4}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($reply), $reply),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
self::assertSame(['cfg', 'sessions'], $client->jetStream()->keyValueBucketNames()->await());
}
/**
* Verifies objectStoreBucketNames() lists OBJ_-prefixed streams with the prefix stripped (#60).
*/
public function testObjectStoreBucketNames(): void
{
$reply = '{"streams":["KV_cfg","OBJ_assets","ORDERS","OBJ_media"],"total":4}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($reply), $reply),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
self::assertSame(['assets', 'media'], $client->jetStream()->objectStoreBucketNames()->await());
}
/**
* Verifies streamNames() returns names from STREAM.NAMES (#35).
*/
public function testStreamNames(): void
{
$reply = '{"streams":["ORDERS","EVENTS"],"total":2,"offset":0,"limit":256}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($reply), $reply),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$names = $client->jetStream()->streamNames()->await();
self::assertSame(['ORDERS', 'EVENTS'], $names);
self::assertStringContainsString('$JS.API.STREAM.NAMES', $transport->writes[3]);
}
/**
* Verifies consumerNames() returns names from CONSUMER.NAMES (#35).
*/
public function testConsumerNames(): void
{
$reply = '{"consumers":["worker-1","worker-2"],"total":2}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($reply), $reply),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$names = $client->jetStream()->consumerNames('ORDERS')->await();
self::assertSame(['worker-1', 'worker-2'], $names);
self::assertStringContainsString('$JS.API.CONSUMER.NAMES.ORDERS', $transport->writes[3]);
}
/**
* Verifies getLastMessageForSubject() uses last_by_subj and parses the stored message (#36).
*/
public function testGetLastMessageForSubject(): void
{
$reply = '{"message":{"subject":"orders.new","seq":7,"data":"' . base64_encode('hello') . '"}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($reply), $reply),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$message = $client->jetStream()->getLastMessageForSubject('ORDERS', 'orders.new')->await();
self::assertSame('orders.new', $message->subject);
self::assertSame('hello', $message->payload);
self::assertStringContainsString('$JS.API.STREAM.MSG.GET.ORDERS', $transport->writes[3]);
self::assertStringContainsString('"last_by_subj":"orders.new"', $transport->writes[3]);
}
/**
* Verifies getLastMessageForSubject() rejects wildcard subjects (#36).
*/
public function testGetLastMessageForSubjectRejectsWildcard(): void
{
$client = new NatsClient(new NatsOptions());
$this->expectException(JetStreamException::class);
$this->expectExceptionMessage('non-wildcard');
$client->jetStream()->getLastMessageForSubject('ORDERS', 'orders.*')->await();
}
/**
* Verifies createOrUpdateStream() falls back to UPDATE when the stream already exists (#44).
*/
public function testCreateOrUpdateStreamFallsBackToUpdate(): void
{
$createErr = '{"error":{"code":400,"err_code":10058,"description":"stream name already in use"}}';
$updateOk = '{"config":{"name":"ORDERS","subjects":["orders.*","orders.archive"]}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
// Muxed replies, FIFO: CREATE -> already-in-use, then UPDATE -> ok (#118).
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createErr), $createErr),
sprintf("MSG _INBOX.b 2 %d\r\n%s\r\n", strlen($updateOk), $updateOk),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$info = $client->jetStream()->createOrUpdateStream('ORDERS', ['orders.*', 'orders.archive'])->await();
self::assertSame('ORDERS', $info->name);
// writes: [0]=CONNECT [1]=PING [2]=mux SUB [3]=CREATE PUB [4]=UPDATE PUB (no per-request SUB/UNSUB).
self::assertStringContainsString('$JS.API.STREAM.CREATE.ORDERS', $transport->writes[3]);
self::assertStringContainsString('$JS.API.STREAM.UPDATE.ORDERS', $transport->writes[4]);
}
/**
* Verifies a JetStream API error envelope's err_code is exposed via getErrCode() alongside the
* HTTP-like code kept in getCode() (#154).
*/
public function testApiErrorEnvelopeExposesErrCode(): void
{
$err = '{"error":{"code":404,"err_code":10059,"description":"stream not found"}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($err), $err),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
try {
$client->jetStream()->getStream('MISSING')->await();
self::fail('Expected a JetStreamException');
} catch (JetStreamException $e) {
self::assertSame(404, $e->getCode());
self::assertSame(10059, $e->getErrCode());
}
}
/**
* Verifies a publish-expectation error ack exposes err_code 10071 via getErrCode(), and that an
* envelope without err_code yields null (old servers) (#154).
*/
public function testPublishExpectationMismatchExposesErrCode(): void
{
$withErrCode = '{"error":{"code":400,"err_code":10071,"description":"wrong last sequence: 5"}}';
$withoutErrCode = '{"error":{"code":400,"description":"wrong last sequence: 5"}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($withErrCode), $withErrCode),
sprintf("MSG _INBOX.b 2 %d\r\n%s\r\n", strlen($withoutErrCode), $withoutErrCode),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
try {
$client->jetStream()->publish('orders.created', '{"id":1}', expectedLastSequence: 99)->await();
self::fail('Expected a JetStreamException');
} catch (JetStreamException $e) {
self::assertSame(400, $e->getCode());
self::assertSame(10071, $e->getErrCode());
}
try {
$client->jetStream()->publish('orders.created', '{"id":1}', expectedLastSequence: 99)->await();
self::fail('Expected a JetStreamException');
} catch (JetStreamException $e) {
self::assertSame(400, $e->getCode());
self::assertNull($e->getErrCode());
}
}
/**
* Verifies createOrUpdateStream() discriminates "stream name already in use" by err_code 10058,
* not by description wording: a reworded description still falls back to UPDATE (#154).
*/
public function testCreateOrUpdateStreamFallsBackToUpdateByErrCode(): void
{
$createErr = '{"error":{"code":400,"err_code":10058,"description":"cannot add stream: that name is taken"}}';
$updateOk = '{"config":{"name":"ORDERS","subjects":["orders.*"]}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createErr), $createErr),
sprintf("MSG _INBOX.b 2 %d\r\n%s\r\n", strlen($updateOk), $updateOk),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$info = $client->jetStream()->createOrUpdateStream('ORDERS', ['orders.*'])->await();
self::assertSame('ORDERS', $info->name);
self::assertStringContainsString('$JS.API.STREAM.UPDATE.ORDERS', $transport->writes[4]);
}
/**
* Verifies createOrUpdateStream() trusts a present err_code over a misleading description: an
* error whose wording mentions "already in use" but whose err_code is not 10058 is re-thrown
* instead of triggering the UPDATE fallback (#154).
*/
public function testCreateOrUpdateStreamRethrowsWhenErrCodeIsNotStreamNameInUse(): void
{
$createErr = '{"error":{"code":400,"err_code":10065,"description":"subjects [orders.*] already in use by an existing stream"}}';
$updateOk = '{"config":{"name":"ORDERS","subjects":["orders.*"]}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createErr), $createErr),
// An UPDATE reply is queued so a wrong fallback would succeed silently instead of hanging.
sprintf("MSG _INBOX.b 2 %d\r\n%s\r\n", strlen($updateOk), $updateOk),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$this->expectException(JetStreamException::class);
$this->expectExceptionMessage('already in use by an existing stream');
$client->jetStream()->createOrUpdateStream('ORDERS', ['orders.*'])->await();
}
/**
* Verifies createOrUpdateStream() still falls back to UPDATE via the description substring when
* the envelope carries no err_code (old servers) (#154).
*/
public function testCreateOrUpdateStreamFallsBackToUpdateWithoutErrCode(): void
{
$createErr = '{"error":{"code":400,"description":"stream name already in use"}}';
$updateOk = '{"config":{"name":"ORDERS","subjects":["orders.*"]}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createErr), $createErr),
sprintf("MSG _INBOX.b 2 %d\r\n%s\r\n", strlen($updateOk), $updateOk),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$info = $client->jetStream()->createOrUpdateStream('ORDERS', ['orders.*'])->await();
self::assertSame('ORDERS', $info->name);
self::assertStringContainsString('$JS.API.STREAM.UPDATE.ORDERS', $transport->writes[4]);
}
/**
* Verifies create/get/delete stream operations map expected payload fields.
*/
public function testStreamCrud(): void
{
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
// Muxed replies, FIFO: CREATE, INFO, DELETE (#118).
$this->muxReplies($transport, [
"MSG _INBOX.a 1 52\r\n{\"config\":{\"name\":\"ORDERS\",\"subjects\":[\"orders.*\"]}}\r\n",
"MSG _INBOX.b 2 52\r\n{\"config\":{\"name\":\"ORDERS\",\"subjects\":[\"orders.*\"]}}\r\n",
"MSG _INBOX.c 3 16\r\n{\"success\":true}\r\n",
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$js = $client->jetStream();
$created = $js->createStream('ORDERS', ['orders.*'])->await();
$fetched = $js->getStream('ORDERS')->await();
$deleted = $js->deleteStream('ORDERS')->await();
self::assertSame('ORDERS', $created->name);
self::assertSame(['orders.*'], $created->subjects);
self::assertSame('ORDERS', $fetched->name);
self::assertTrue($deleted);
// writes: [0]=CONNECT [1]=PING [2]=mux SUB [3]=CREATE [4]=INFO [5]=DELETE (one mux SUB, no UNSUB).
self::assertStringContainsString('$JS.API.STREAM.CREATE.ORDERS', $transport->writes[3]);
self::assertStringContainsString('$JS.API.STREAM.INFO.ORDERS', $transport->writes[4]);
self::assertStringContainsString('$JS.API.STREAM.DELETE.ORDERS', $transport->writes[5]);
}
/**
* Verifies JetStream API error payloads are converted to JetStreamException.
*/
public function testJetStreamApiErrorMapping(): void
{
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
"MSG _INBOX.a 1 48\r\n{\"error\":{\"code\":404,\"description\":\"not found\"}}\r\n",
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$this->expectException(JetStreamException::class);
$this->expectExceptionMessage('not found');
$client->jetStream()->getStream('MISSING')->await();
}
/**
* Verifies the client returns the same JetStream context instance on repeated access.
*/
public function testJetStreamContextIsCached(): void
{
$client = new NatsClient(new NatsOptions(), new FakeTransport());
$a = $client->jetStream();
$b = $client->jetStream();
self::assertInstanceOf(JetStreamContext::class, $a);
self::assertSame($a, $b);
}
/**
* Verifies objectStore() constructs an equivalent bucket wrapper per call instead of memoizing:
* the per-name cache had no eviction, so a long-lived context accumulated one wrapper per bucket
* name touched, forever (#133). Fresh instances must stay behaviorally interchangeable (equal
* state, same backing stream).
*/
public function testObjectStoreConstructsEquivalentBucketPerCall(): void
{
$client = new NatsClient(new NatsOptions(), new FakeTransport());
$a = $client->jetStream()->objectStore('assets');
$b = $client->jetStream()->objectStore('assets');
$c = $client->jetStream()->objectStore('other');
// No memoization: each call builds a new wrapper (nothing is retained on the context)...
self::assertNotSame($a, $b);
// ...and the wrappers are all-readonly value objects, so repeated calls are interchangeable.
self::assertEquals($a, $b);
self::assertSame('OBJ_assets', $a->streamName());
self::assertSame('OBJ_assets', $b->streamName());
self::assertSame('OBJ_other', $c->streamName());
}
/**
* Verifies keyValue() constructs an equivalent bucket wrapper per call instead of memoizing -
* same no-eviction rationale as the object store accessor (#133).
*/
public function testKeyValueConstructsEquivalentBucketPerCall(): void
{
$client = new NatsClient(new NatsOptions(), new FakeTransport());
$a = $client->jetStream()->keyValue('profiles');
$b = $client->jetStream()->keyValue('profiles');
$c = $client->jetStream()->keyValue('sessions');
self::assertInstanceOf(KeyValueBucket::class, $a);
// No memoization: each call builds a new wrapper (nothing is retained on the context)...
self::assertNotSame($a, $b);
// ...and the wrappers are all-readonly value objects, so repeated calls are interchangeable.
self::assertEquals($a, $b);
self::assertSame('KV_profiles', $a->streamName());
self::assertSame('KV_profiles', $b->streamName());
self::assertSame('KV_sessions', $c->streamName());
}
/**
* Verifies pullConsumer helper returns an iterator wrapper.
*/
public function testPullConsumerReturnsIterator(): void
{
$client = new NatsClient(new NatsOptions(), new FakeTransport());
$iterator = $client->jetStream()->pullConsumer('ORDERS', 'PROC');
self::assertInstanceOf(PullConsumerIterator::class, $iterator);
}
/**
* Verifies consumer create/get/delete operations map expected payload fields.
*/
public function testConsumerCrud(): void
{
$createPayload = '{"stream_name":"ORDERS","name":"PROC","config":{"durable_name":"PROC"}}';
$infoPayload = '{"stream_name":"ORDERS","name":"PROC","config":{"durable_name":"PROC"}}';
$deletePayload = '{"success":true}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
// Muxed replies, FIFO: CREATE, INFO, DELETE (#118).
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createPayload), $createPayload),
sprintf("MSG _INBOX.b 2 %d\r\n%s\r\n", strlen($infoPayload), $infoPayload),
sprintf("MSG _INBOX.c 3 %d\r\n%s\r\n", strlen($deletePayload), $deletePayload),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$js = $client->jetStream();
$created = $js->createConsumer('ORDERS', 'PROC', 'orders.*')->await();
$fetched = $js->getConsumer('ORDERS', 'PROC')->await();
$deleted = $js->deleteConsumer('ORDERS', 'PROC')->await();
self::assertSame('ORDERS', $created->streamName);
self::assertSame('PROC', $created->name);
self::assertSame('PROC', $fetched->name);
self::assertTrue($deleted);
// writes: [0]=CONNECT [1]=PING [2]=mux SUB [3]=CREATE [4]=INFO [5]=DELETE (one mux SUB, no UNSUB).
self::assertStringContainsString('$JS.API.CONSUMER.CREATE.ORDERS.PROC', $transport->writes[3]);
self::assertStringContainsString('$JS.API.CONSUMER.INFO.ORDERS.PROC', $transport->writes[4]);
self::assertStringContainsString('$JS.API.CONSUMER.DELETE.ORDERS.PROC', $transport->writes[5]);
}
/**
* Verifies createConsumer sends filter_subjects (and omits the singular filter_subject) when an
* array of filters is supplied via options (issue #10).
*/
public function testCreateConsumerWithFilterSubjects(): void
{
$createPayload = '{"stream_name":"ORDERS","name":"PROC","config":{"durable_name":"PROC"}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createPayload), $createPayload),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$client->jetStream()->createConsumer('ORDERS', 'PROC', null, ['filter_subjects' => ['orders.eu.>', 'orders.us.>']])->await();
self::assertStringContainsString('"filter_subjects":["orders.eu.>","orders.us.>"]', $transport->writes[3]);
self::assertStringNotContainsString('"filter_subject"', $transport->writes[3]);
}
/**
* Verifies combining a single filter subject with filter_subjects is rejected before dispatch.
*/
public function testCreateConsumerRejectsBothFilterForms(): void
{
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$this->expectException(JetStreamException::class);
$this->expectExceptionMessage('Use either a single filter subject or filter_subjects, not both');
try {
$client->jetStream()->createConsumer('ORDERS', 'PROC', 'orders.*', ['filter_subjects' => ['orders.eu.>']])->await();
} finally {
self::assertCount(2, $transport->writes);
}
}
/**
* Verifies a filter_subjects array containing an empty subject is rejected before dispatch.
*/
public function testCreateConsumerRejectsEmptyFilterSubjectEntry(): void
{
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$this->expectException(JetStreamException::class);
$this->expectExceptionMessage('filter_subjects must contain only non-empty subject strings');
try {
$client->jetStream()->createConsumer('ORDERS', 'PROC', null, ['filter_subjects' => ['orders.eu.>', '']])->await();
} finally {
self::assertCount(2, $transport->writes);
}
}
/**
* Verifies the mutual-exclusion guard also fires when the singular filter_subject is smuggled in
* via the options bag alongside filter_subjects (issue #10).
*/
public function testCreateConsumerRejectsFilterSubjectInOptionsConflict(): void
{
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$this->expectException(JetStreamException::class);
$this->expectExceptionMessage('Use either a single filter subject or filter_subjects, not both');
try {
$client->jetStream()->createConsumer('ORDERS', 'PROC', null, [
'filter_subject' => 'orders.eu.>',
'filter_subjects' => ['orders.us.>'],
])->await();
} finally {
self::assertCount(2, $transport->writes);
}
}
/**
* Verifies an empty filter subject is rejected uniformly on the ephemeral path too (issue #10).
*/
public function testCreateEphemeralConsumerRejectsEmptyFilterSubject(): void
{
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$this->expectException(JetStreamException::class);
$this->expectExceptionMessage('Consumer filter subject must not be empty');
try {
$client->jetStream()->createEphemeralConsumer('ORDERS', '')->await();
} finally {
self::assertCount(2, $transport->writes);
}
}
/**
* Verifies filter_subjects flows through the push-consumer create path too (issue #10).
*/
public function testCreatePushConsumerWithFilterSubjects(): void
{
$createPayload = '{"stream_name":"ORDERS","name":"PROC","config":{"durable_name":"PROC"}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createPayload), $createPayload),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$client->jetStream()->createPushConsumer('ORDERS', 'PROC', '_INBOX.deliver', null, [
'filter_subjects' => ['orders.eu.>', 'orders.us.>'],
])->await();
self::assertStringContainsString('"filter_subjects":["orders.eu.>","orders.us.>"]', $transport->writes[3]);
self::assertStringNotContainsString('"filter_subject"', $transport->writes[3]);
}
/**
* Verifies createConsumer validates and forwards priority-group config (issue #7).
*/
public function testCreateConsumerWithPriorityGroups(): void
{
$createPayload = '{"stream_name":"ORDERS","name":"PROC","config":{"durable_name":"PROC"}}';
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$this->muxReplies($transport, [
sprintf("MSG _INBOX.a 1 %d\r\n%s\r\n", strlen($createPayload), $createPayload),
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();
$client->jetStream()->createConsumer('ORDERS', 'PROC', null, [
'priority_groups' => ['g1'],
'priority_policy' => 'pinned_client',
])->await();
self::assertStringContainsString('"priority_groups":["g1"]', $transport->writes[3]);
self::assertStringContainsString('"priority_policy":"pinned_client"', $transport->writes[3]);
}
/**
* Verifies an invalid priority policy is rejected before dispatch.
*/
public function testCreateConsumerRejectsInvalidPriorityPolicy(): void
{
$transport = new FakeTransport([
'INFO {"server_id":"S1","server_name":"n1","version":"2.12.0","jetstream":true,"max_payload":1048576,"headers":true}' . "\r\n",
"PONG\r\n",
]);
$client = new NatsClient(new NatsOptions(requestTimeoutMs: 1000), $transport);
$client->connect()->await();