23489465477

Committed 24 Mar 2026 12:29PM UTC coverage: 93.321% (-1.9%) from 95.179%

Build # 23489465477

Build Type

push

github

Committed by

nekufa

Commit Message

ephermal consumer configuration fix

Coverage Stats

5 of 6 new or added lines in 3 files covered. (83.33%)

46 existing lines in 3 files now uncovered.

1537 of 1647 relevant lines covered (93.32%)

75.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.72

/src/Consumer/Configuration.php

<?php

declare(strict_types=1);

namespace Basis\Nats\Consumer;

use DateTimeInterface;

class Configuration
{
    private const OPT_START_TIME_FORMAT = 'Y-m-d\TH:i:s.uP'; # ISO8601 with microseconds

    private bool $ephemeral = false;
    private ?bool $flowControl = null;
    private ?bool $headersOnly = null;
    private ?int $ackWait = null;
    private ?int $idleHeartbeat = null;
    private ?int $maxAckPending = null;
    private ?int $maxDeliver = null;
    private ?int $maxWaiting = null;
    private ?int $startSequence = null;
    private ?DateTimeInterface $startTime = null;
    private ?string $deliverGroup = null;
    private ?string $deliverSubject = null;
    private ?string $description = null;
    private ?string $subjectFilter = null;
    private ?array $subjectFilters = null;
    private string $ackPolicy = AckPolicy::EXPLICIT;
    private string $deliverPolicy = DeliverPolicy::ALL;
    private string $replayPolicy = ReplayPolicy::INSTANT;
    private ?int $inactiveThreshold = null;

    public function __construct(
        private readonly string $stream,
        private ?string $name = null
    ) {
    }

    public function getAckPolicy(): string
    {
        return $this->ackPolicy;
    }

    public function getName(): ?string
    {
        return $this->name;
    }

    public function getStream(): string
    {
        return $this->stream;
    }

    public function getSubjectFilter(): ?string
    {
        return $this->subjectFilter;
    }

    public function getAckWait()
    {
        return $this->ackWait;
    }

    public function getDeliverGroup()
    {
        return $this->deliverGroup;
    }

    public function getDeliverPolicy()
    {
        return $this->deliverPolicy;
    }

    public function getDeliverSubject()
    {
        return $this->deliverSubject;
    }

    public function getDescription()
    {
        return $this->description;
    }

    public function getFlowControl()
    {
        return $this->flowControl;
    }

    public function getHeadersOnly()
    {
        return $this->headersOnly;
    }

    public function getIdleHeartbeat()
    {
        return $this->idleHeartbeat;
    }

    public function getMaxAckPending()
    {
        return $this->maxAckPending;
    }

    public function getMaxDeliver()
    {
        return $this->maxDeliver;
    }

    public function getMaxWaiting()
    {
        return $this->maxWaiting;
    }

    public function getStartSequence()
    {
        return $this->startSequence;
    }

    public function getStartTime()
    {
        return $this->startTime;
    }

    public function getReplayPolicy()
    {
        return $this->replayPolicy;
    }

    public function isEphemeral(): bool
    {
        return $this->ephemeral;
    }

    /**
     * @deprected
     */
    public function ephemeral(): self
    {
        return $this->setEphemeral(true);
    }

    public function setEphemeral(bool $ephemeral): self
    {
        $this->ephemeral = $ephemeral;
        return $this;
    }

    public function setAckPolicy(string $ackPolicy): self
    {
        $this->ackPolicy = AckPolicy::validate($ackPolicy);
        return $this;
    }

    public function setAckWait(int $ackWait): self
    {
        $this->ackWait = $ackWait;
        return $this;
    }

    public function setDeliverGroup(?string $deliverGroup): self
    {
        $this->deliverGroup = $deliverGroup;
        return $this;
    }

    public function setDeliverPolicy(string $deliverPolicy): self
    {
        $this->deliverPolicy = DeliverPolicy::validate($deliverPolicy);
        return $this;
    }

    public function setDeliverSubject(?string $deliverSubject): self
    {
        $this->deliverSubject = $deliverSubject;
        return $this;
    }

    public function setDescription(?string $description): self
    {
        $this->description = $description;
        return $this;
    }

    public function setFlowControl(?bool $flowControl): self
    {
        $this->flowControl = $flowControl;
        return $this;
    }

    public function setHeadersOnly(?bool $headersOnly): self
    {
        $this->headersOnly = $headersOnly;
        return $this;
    }

    public function setIdleHeartbeat(?int $idleHeartbeat): self
    {
        $this->idleHeartbeat = $idleHeartbeat;
        return $this;
    }

    public function setMaxAckPending(int $maxAckPending): self
    {
        $this->maxAckPending = $maxAckPending;
        return $this;
    }

    public function setMaxDeliver(int $maxDeliver): self
    {
        $this->maxDeliver = $maxDeliver;
        return $this;
    }

    public function setMaxWaiting(int $maxWaiting): self
    {
        $this->maxWaiting = $maxWaiting;
        return $this;
    }

    public function setName(string $name): self
    {
        if ($this->isEphemeral()) {
            $this->name = $name;
        }

        return $this;
    }

    public function setStartSequence(int $startSeq): self
    {
        $this->startSequence = $startSeq;
        return $this;
    }

    public function setStartTime(DateTimeInterface $startTime): self
    {
        $this->startTime = $startTime;
        return $this;
    }

    public function setReplayPolicy(string $replayPolicy): self
    {
        $this->replayPolicy = ReplayPolicy::validate($replayPolicy);
        return $this;
    }

    public function setSubjectFilter(string $subjectFilter): self
    {
        $this->subjectFilter = $subjectFilter;
        return $this;
    }

    public function setSubjectFilters(array $subjectFilters): self
    {
        $this->subjectFilters = $subjectFilters;
        return $this;
    }

    public function getSubjectFilters(): ?array
    {
        return $this->subjectFilters;
    }

    public function setInactiveThreshold(int $inactiveThresholdNanoSeconds): self
    {
        $this->inactiveThreshold = $inactiveThresholdNanoSeconds;
        return $this;
    }

    public function getInactiveThreshold(): ?int
    {
        return $this->inactiveThreshold;
    }

    public static function fromObject(object $object): static
    {
        $config = $object->config;
        // For ephemeral consumers, durable_name is not set
        // Use name from config if durable_name exists, otherwise null (ephemeral)
        $name = isset($config->durable_name) ? $config->durable_name : null;

        $instance = new static($object->stream_name, $name);

        // Set ephemeral if no durable name
        if ($name === null) {
            $instance->setEphemeral(true);
        }

        $instance->setAckPolicy($config->ack_policy);

        if (isset($config->ack_wait)) {
            $instance->setAckWait($config->ack_wait);
        }

        if (isset($config->deliver_group)) {
            $instance->setDeliverGroup($config->deliver_group);
        }

        $instance->setDeliverPolicy($config->deliver_policy);

        if (isset($config->deliver_subject)) {
            $instance->setDeliverSubject($config->deliver_subject);
        }

        if (isset($config->description)) {
            $instance->setDescription($config->description);
        }

        if (isset($config->flow_control)) {
            $instance->setFlowControl($config->flow_control);
        }

        if (isset($config->headers_only)) {
            $instance->setHeadersOnly($config->headers_only);
        }

        if (isset($config->idle_heartbeat)) {
            $instance->setIdleHeartbeat($config->idle_heartbeat);
        }

        if (isset($config->max_ack_pending)) {
            $instance->setMaxAckPending($config->max_ack_pending);
        }

        if (isset($config->max_deliver)) {
            $instance->setMaxDeliver($config->max_deliver);
        }

        if (isset($config->max_waiting)) {
            $instance->setMaxWaiting($config->max_waiting);
        }

        if (isset($config->replay_policy)) {
            $instance->setReplayPolicy($config->replay_policy);
        }

        if (isset($config->inactive_threshold)) {
            $instance->setInactiveThreshold($config->inactive_threshold);
        }

        // Handle start sequence / start time based on deliver policy
        if (isset($config->opt_start_seq)) {
            $instance->setStartSequence($config->opt_start_seq);
        }

        if (isset($config->opt_start_time)) {
            $startTime = \DateTime::createFromFormat(self::OPT_START_TIME_FORMAT, $config->opt_start_time);
            if ($startTime !== false) {
                $instance->setStartTime($startTime);
            }
        }

        // Handle subject filters (filter_subjects takes precedence over filter_subject)
        if (isset($config->filter_subjects)) {
            $instance->setSubjectFilters($config->filter_subjects);
        } elseif (isset($config->filter_subject)) {
            $instance->setSubjectFilter($config->filter_subject);
        }

        return $instance;
    }

    public function toArray(): array
    {
        $config = [
            'ack_policy' => $this->getAckPolicy(),
            'ack_wait' => $this->getAckWait(),
            'deliver_group' => $this->getDeliverGroup(),
            'deliver_policy' => $this->getDeliverPolicy(),
            'deliver_subject' => $this->getDeliverSubject(),
            'description' => $this->getDescription(),
            'durable_name' => $this->isEphemeral() ? null : $this->getName(),
            'flow_control' => $this->getFlowControl(),
            'headers_only' => $this->getHeadersOnly(),
            'idle_heartbeat' => $this->getIdleHeartbeat(),
            'max_ack_pending' => $this->getMaxAckPending(),
            'max_deliver' => $this->getMaxDeliver(),
            'max_waiting' => $this->getMaxWaiting(),
            'replay_policy' => $this->getReplayPolicy(),
            'inactive_threshold' => $this->getInactiveThreshold(),
        ];

        switch ($this->getDeliverPolicy()) {
            case DeliverPolicy::BY_START_SEQUENCE:
                $config['opt_start_seq'] = $this->getStartSequence();
                break;

            case DeliverPolicy::BY_START_TIME:
                $config['opt_start_time'] = $this->getStartTime()?->format(self::OPT_START_TIME_FORMAT);
                break;
        }

        if ($this->getSubjectFilters()) {
            $config['filter_subjects'] = $this->getSubjectFilters();
        } elseif ($this->getSubjectFilter()) {
            $config['filter_subject'] = $this->getSubjectFilter();
        }

        foreach ($config as $k => $v) {
            if ($v === null) {
                unset($config[$k]);
            }
        }

        return [
            'stream_name' => $this->getStream(),
            'config' => $config,
        ];
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Basis\Nats\Consumer;
6
7	use DateTimeInterface;
8
9	class Configuration
10	{
11	private const OPT_START_TIME_FORMAT = 'Y-m-d\TH:i:s.uP'; # ISO8601 with microseconds
12
13	private bool $ephemeral = false;
14	private ?bool $flowControl = null;
15	private ?bool $headersOnly = null;
16	private ?int $ackWait = null;
17	private ?int $idleHeartbeat = null;
18	private ?int $maxAckPending = null;
19	private ?int $maxDeliver = null;
20	private ?int $maxWaiting = null;
21	private ?int $startSequence = null;
22	private ?DateTimeInterface $startTime = null;
23	private ?string $deliverGroup = null;
24	private ?string $deliverSubject = null;
25	private ?string $description = null;
26	private ?string $subjectFilter = null;
27	private ?array $subjectFilters = null;
28	private string $ackPolicy = AckPolicy::EXPLICIT;
29	private string $deliverPolicy = DeliverPolicy::ALL;
30	private string $replayPolicy = ReplayPolicy::INSTANT;
31	private ?int $inactiveThreshold = null;
32
33	public function __construct(	92✔
34	private readonly string $stream,
35	private ?string $name = null
36	) {
37	}	92✔
38
39	public function getAckPolicy(): string	80✔
40	{
41	return $this->ackPolicy;	80✔
42	}
43
44	public function getName(): ?string	80✔
45	{
46	return $this->name;	80✔
47	}
48
49	public function getStream(): string	80✔
50	{
51	return $this->stream;	80✔
52	}
53
54	public function getSubjectFilter(): ?string	76✔
55	{
56	return $this->subjectFilter;	76✔
57	}
58
59	public function getAckWait()	80✔
60	{
61	return $this->ackWait;	80✔
62	}
63
64	public function getDeliverGroup()	80✔
65	{
66	return $this->deliverGroup;	80✔
67	}
68
69	public function getDeliverPolicy()	80✔
70	{
71	return $this->deliverPolicy;	80✔
72	}
73
74	public function getDeliverSubject()	80✔
75	{
76	return $this->deliverSubject;	80✔
77	}
78
79	public function getDescription()	80✔
80	{
81	return $this->description;	80✔
82	}
83
84	public function getFlowControl()	80✔
85	{
86	return $this->flowControl;	80✔
87	}
88
89	public function getHeadersOnly()	80✔
90	{
91	return $this->headersOnly;	80✔
92	}
93
94	public function getIdleHeartbeat()	80✔
95	{
96	return $this->idleHeartbeat;	80✔
97	}
98
99	public function getMaxAckPending()	80✔
100	{
101	return $this->maxAckPending;	80✔
102	}
103
104	public function getMaxDeliver()	80✔
105	{
106	return $this->maxDeliver;	80✔
107	}
108
109	public function getMaxWaiting()	80✔
110	{
111	return $this->maxWaiting;	80✔
112	}
113
114	public function getStartSequence()	4✔
115	{
116	return $this->startSequence;	4✔
117	}
118
119	public function getStartTime()	4✔
120	{
121	return $this->startTime;	4✔
122	}
123
124	public function getReplayPolicy()	80✔
125	{
126	return $this->replayPolicy;	80✔
127	}
128
129	public function isEphemeral(): bool	80✔
130	{
131	return $this->ephemeral;	80✔
132	}
133
134	/**
135	* @deprected
136	*/
UNCOV 137	public function ephemeral(): self	×
138	{
NEW 139	return $this->setEphemeral(true);	×
140	}
141
142	public function setEphemeral(bool $ephemeral): self	12✔
143	{
144	$this->ephemeral = $ephemeral;	12✔
145	return $this;	12✔
146	}
147
148	public function setAckPolicy(string $ackPolicy): self	36✔
149	{
150	$this->ackPolicy = AckPolicy::validate($ackPolicy);	36✔
151	return $this;	32✔
152	}
153
154	public function setAckWait(int $ackWait): self	16✔
155	{
156	$this->ackWait = $ackWait;	16✔
157	return $this;	16✔
158	}
159
UNCOV 160	public function setDeliverGroup(?string $deliverGroup): self	×
161	{
UNCOV 162	$this->deliverGroup = $deliverGroup;	×
UNCOV 163	return $this;	×
164	}
165
166	public function setDeliverPolicy(string $deliverPolicy): self	20✔
167	{
168	$this->deliverPolicy = DeliverPolicy::validate($deliverPolicy);	20✔
169	return $this;	16✔
170	}
171
172	public function setDeliverSubject(?string $deliverSubject): self	×
173	{
174	$this->deliverSubject = $deliverSubject;	×
175	return $this;	×
176	}
177
UNCOV 178	public function setDescription(?string $description): self	×
179	{
UNCOV 180	$this->description = $description;	×
UNCOV 181	return $this;	×
182	}
183
UNCOV 184	public function setFlowControl(?bool $flowControl): self	×
185	{
UNCOV 186	$this->flowControl = $flowControl;	×
UNCOV 187	return $this;	×
188	}
189
UNCOV 190	public function setHeadersOnly(?bool $headersOnly): self	×
191	{
UNCOV 192	$this->headersOnly = $headersOnly;	×
UNCOV 193	return $this;	×
194	}
195
UNCOV 196	public function setIdleHeartbeat(?int $idleHeartbeat): self	×
197	{
UNCOV 198	$this->idleHeartbeat = $idleHeartbeat;	×
UNCOV 199	return $this;	×
200	}
201
202	public function setMaxAckPending(int $maxAckPending): self	8✔
203	{
204	$this->maxAckPending = $maxAckPending;	8✔
205	return $this;	8✔
206	}
207
208	public function setMaxDeliver(int $maxDeliver): self	4✔
209	{
210	$this->maxDeliver = $maxDeliver;	4✔
211	return $this;	4✔
212	}
213
214	public function setMaxWaiting(int $maxWaiting): self	4✔
215	{
216	$this->maxWaiting = $maxWaiting;	4✔
217	return $this;	4✔
218	}
219
220	public function setName(string $name): self	12✔
221	{
222	if ($this->isEphemeral()) {	12✔
223	$this->name = $name;	12✔
224	}
225
226	return $this;	12✔
227	}
228
229	public function setStartSequence(int $startSeq): self	4✔
230	{
231	$this->startSequence = $startSeq;	4✔
232	return $this;	4✔
233	}
234
235	public function setStartTime(DateTimeInterface $startTime): self	4✔
236	{
237	$this->startTime = $startTime;	4✔
238	return $this;	4✔
239	}
240
241	public function setReplayPolicy(string $replayPolicy): self	28✔
242	{
243	$this->replayPolicy = ReplayPolicy::validate($replayPolicy);	28✔
244	return $this;	24✔
245	}
246
247	public function setSubjectFilter(string $subjectFilter): self	56✔
248	{
249	$this->subjectFilter = $subjectFilter;	56✔
250	return $this;	56✔
251	}
252
253	public function setSubjectFilters(array $subjectFilters): self	4✔
254	{
255	$this->subjectFilters = $subjectFilters;	4✔
256	return $this;	4✔
257	}
258
259	public function getSubjectFilters(): ?array	80✔
260	{
261	return $this->subjectFilters;	80✔
262	}
263
264	public function setInactiveThreshold(int $inactiveThresholdNanoSeconds): self	4✔
265	{
266	$this->inactiveThreshold = $inactiveThresholdNanoSeconds;	4✔
267	return $this;	4✔
268	}
269
270	public function getInactiveThreshold(): ?int	80✔
271	{
272	return $this->inactiveThreshold;	80✔
273	}
274
275	public static function fromObject(object $object): static	4✔
276	{
277	$config = $object->config;	4✔
278	// For ephemeral consumers, durable_name is not set
279	// Use name from config if durable_name exists, otherwise null (ephemeral)
280	$name = isset($config->durable_name) ? $config->durable_name : null;	4✔
281
282	$instance = new static($object->stream_name, $name);	4✔
283
284	// Set ephemeral if no durable name
285	if ($name === null) {	4✔
UNCOV 286	$instance->setEphemeral(true);	×
287	}
288
289	$instance->setAckPolicy($config->ack_policy);	4✔
290
291	if (isset($config->ack_wait)) {	4✔
292	$instance->setAckWait($config->ack_wait);	4✔
293	}
294
295	if (isset($config->deliver_group)) {	4✔
UNCOV 296	$instance->setDeliverGroup($config->deliver_group);	×
297	}
298
299	$instance->setDeliverPolicy($config->deliver_policy);	4✔
300
301	if (isset($config->deliver_subject)) {	4✔
UNCOV 302	$instance->setDeliverSubject($config->deliver_subject);	×
303	}
304
305	if (isset($config->description)) {	4✔
UNCOV 306	$instance->setDescription($config->description);	×
307	}
308
309	if (isset($config->flow_control)) {	4✔
UNCOV 310	$instance->setFlowControl($config->flow_control);	×
311	}
312
313	if (isset($config->headers_only)) {	4✔
UNCOV 314	$instance->setHeadersOnly($config->headers_only);	×
315	}
316
317	if (isset($config->idle_heartbeat)) {	4✔
UNCOV 318	$instance->setIdleHeartbeat($config->idle_heartbeat);	×
319	}
320
321	if (isset($config->max_ack_pending)) {	4✔
322	$instance->setMaxAckPending($config->max_ack_pending);	4✔
323	}
324
325	if (isset($config->max_deliver)) {	4✔
326	$instance->setMaxDeliver($config->max_deliver);	4✔
327	}
328
329	if (isset($config->max_waiting)) {	4✔
330	$instance->setMaxWaiting($config->max_waiting);	4✔
331	}
332
333	if (isset($config->replay_policy)) {	4✔
334	$instance->setReplayPolicy($config->replay_policy);	4✔
335	}
336
337	if (isset($config->inactive_threshold)) {	4✔
UNCOV 338	$instance->setInactiveThreshold($config->inactive_threshold);	×
339	}
340
341	// Handle start sequence / start time based on deliver policy
342	if (isset($config->opt_start_seq)) {	4✔
UNCOV 343	$instance->setStartSequence($config->opt_start_seq);	×
344	}
345
346	if (isset($config->opt_start_time)) {	4✔
UNCOV 347	$startTime = \DateTime::createFromFormat(self::OPT_START_TIME_FORMAT, $config->opt_start_time);	×
UNCOV 348	if ($startTime !== false) {	×
UNCOV 349	$instance->setStartTime($startTime);	×
350	}
351	}
352
353	// Handle subject filters (filter_subjects takes precedence over filter_subject)
354	if (isset($config->filter_subjects)) {	4✔
355	$instance->setSubjectFilters($config->filter_subjects);	4✔
UNCOV 356	} elseif (isset($config->filter_subject)) {	×
UNCOV 357	$instance->setSubjectFilter($config->filter_subject);	×
358	}
359
360	return $instance;	4✔
361	}
362
363	public function toArray(): array	80✔
364	{
365	$config = [	80✔
366	'ack_policy' => $this->getAckPolicy(),	80✔
367	'ack_wait' => $this->getAckWait(),	80✔
368	'deliver_group' => $this->getDeliverGroup(),	80✔
369	'deliver_policy' => $this->getDeliverPolicy(),	80✔
370	'deliver_subject' => $this->getDeliverSubject(),	80✔
371	'description' => $this->getDescription(),	80✔
372	'durable_name' => $this->isEphemeral() ? null : $this->getName(),	80✔
373	'flow_control' => $this->getFlowControl(),	80✔
374	'headers_only' => $this->getHeadersOnly(),	80✔
375	'idle_heartbeat' => $this->getIdleHeartbeat(),	80✔
376	'max_ack_pending' => $this->getMaxAckPending(),	80✔
377	'max_deliver' => $this->getMaxDeliver(),	80✔
378	'max_waiting' => $this->getMaxWaiting(),	80✔
379	'replay_policy' => $this->getReplayPolicy(),	80✔
380	'inactive_threshold' => $this->getInactiveThreshold(),	80✔
381	];	80✔
382
383	switch ($this->getDeliverPolicy()) {	80✔
384	case DeliverPolicy::BY_START_SEQUENCE:	80✔
385	$config['opt_start_seq'] = $this->getStartSequence();	4✔
386	break;	4✔
387
388	case DeliverPolicy::BY_START_TIME:	76✔
389	$config['opt_start_time'] = $this->getStartTime()?->format(self::OPT_START_TIME_FORMAT);	4✔
390	break;	4✔
391	}
392
393	if ($this->getSubjectFilters()) {	80✔
394	$config['filter_subjects'] = $this->getSubjectFilters();	4✔
395	} elseif ($this->getSubjectFilter()) {	76✔
396	$config['filter_subject'] = $this->getSubjectFilter();	52✔
397	}
398
399	foreach ($config as $k => $v) {	80✔
400	if ($v === null) {	80✔
401	unset($config[$k]);	80✔
402	}
403	}
404
405	return [	80✔
406	'stream_name' => $this->getStream(),	80✔
407	'config' => $config,	80✔
408	];	80✔
409	}
410	}

basis-company / nats.php / 23489465477

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous