Skip to content

Commit 97a14a8

Browse files
GianfriAurGianfriAur
committed
[TEST] AutoPublisher
1 parent d4c5c2a commit 97a14a8

3 files changed

Lines changed: 649 additions & 0 deletions

File tree

‎tests/Unit/AutoPublisherTest.php‎

Lines changed: 323 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,323 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
use PhpOpcua\Client\Client;
6+
use PhpOpcua\Client\Exception\ConnectionException;
7+
use PhpOpcua\Client\Types\PublishResult;
8+
use PhpOpcua\SessionManager\Daemon\AutoPublisher;
9+
use PhpOpcua\SessionManager\Daemon\Session;
10+
use PhpOpcua\SessionManager\Daemon\SessionStore;
11+
use Psr\Log\NullLogger;
12+
use React\EventLoop\LoopInterface;
13+
use React\EventLoop\TimerInterface;
14+
15+
function drainAutoPublishCallbacks(array &$callbacks, int $maxIterations = 50): void
16+
{
17+
$i = 0;
18+
while (!empty($callbacks) && $i < $maxIterations) {
19+
$cb = array_shift($callbacks);
20+
($cb['callback'])();
21+
$i++;
22+
}
23+
}
24+
25+
describe('AutoPublisher', function () {
26+
27+
beforeEach(function () {
28+
$this->store = new SessionStore();
29+
$this->callbacks = [];
30+
$this->timer = $this->createStub(TimerInterface::class);
31+
$callbacks = &$this->callbacks;
32+
$timer = $this->timer;
33+
34+
$this->loop = $this->createMock(LoopInterface::class);
35+
$this->loop->method('addTimer')->willReturnCallback(function ($delay, $callback) use (&$callbacks, $timer) {
36+
$callbacks[] = ['delay' => $delay, 'callback' => $callback];
37+
return $timer;
38+
});
39+
40+
$this->recoveryResult = true;
41+
$this->recoveryCalled = false;
42+
$recoveryResult = &$this->recoveryResult;
43+
$recoveryCalled = &$this->recoveryCalled;
44+
45+
$this->publisher = new AutoPublisher(
46+
$this->store,
47+
$this->loop,
48+
new NullLogger(),
49+
function (Session $s) use (&$recoveryResult, &$recoveryCalled) {
50+
$recoveryCalled = true;
51+
return $recoveryResult;
52+
},
53+
);
54+
});
55+
56+
it('is not active for unknown sessions', function () {
57+
expect($this->publisher->isActive('nonexistent'))->toBeFalse();
58+
});
59+
60+
it('starts auto-publish and schedules a timer', function () {
61+
$this->publisher->startSession('s1');
62+
63+
expect($this->publisher->isActive('s1'))->toBeTrue();
64+
expect($this->callbacks)->toHaveCount(1);
65+
expect($this->callbacks[0]['delay'])->toBe(0.0);
66+
});
67+
68+
it('is idempotent on repeated startSession calls', function () {
69+
$this->publisher->startSession('s1');
70+
$this->publisher->startSession('s1');
71+
72+
expect($this->callbacks)->toHaveCount(1);
73+
});
74+
75+
it('stops auto-publish and cancels the timer', function () {
76+
$this->loop->expects($this->once())->method('cancelTimer')->with($this->timer);
77+
78+
$this->publisher->startSession('s1');
79+
$this->publisher->stopSession('s1');
80+
81+
expect($this->publisher->isActive('s1'))->toBeFalse();
82+
});
83+
84+
it('stopSession is safe for inactive sessions', function () {
85+
$this->publisher->stopSession('nonexistent');
86+
87+
expect($this->publisher->isActive('nonexistent'))->toBeFalse();
88+
});
89+
90+
it('stops all active sessions', function () {
91+
$this->publisher->startSession('s1');
92+
$this->publisher->startSession('s2');
93+
94+
$this->publisher->stopAll();
95+
96+
expect($this->publisher->isActive('s1'))->toBeFalse();
97+
expect($this->publisher->isActive('s2'))->toBeFalse();
98+
});
99+
100+
it('calls publish and schedules next cycle on success', function () {
101+
$client = $this->createStub(Client::class);
102+
$client->method('publish')->willReturn(new PublishResult(1, 10, false, [], []));
103+
104+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
105+
$session->addSubscription(1, 500.0);
106+
$this->store->create($session);
107+
108+
$this->publisher->startSession('s1');
109+
drainAutoPublishCallbacks($this->callbacks, 1);
110+
111+
expect($this->publisher->isActive('s1'))->toBeTrue();
112+
expect($this->callbacks)->toHaveCount(1);
113+
expect($this->callbacks[0]['delay'])->toBe(0.5 * 0.75);
114+
});
115+
116+
it('drains quickly when moreNotifications is true', function () {
117+
$callCount = 0;
118+
$client = $this->createStub(Client::class);
119+
$client->method('publish')->willReturnCallback(function () use (&$callCount) {
120+
$callCount++;
121+
return new PublishResult(1, $callCount, $callCount < 3, [], []);
122+
});
123+
124+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
125+
$session->addSubscription(1, 500.0);
126+
$this->store->create($session);
127+
128+
$this->publisher->startSession('s1');
129+
drainAutoPublishCallbacks($this->callbacks, 5);
130+
131+
expect($callCount)->toBeGreaterThanOrEqual(3);
132+
});
133+
134+
it('uses short delay for moreNotifications drain', function () {
135+
$client = $this->createStub(Client::class);
136+
$client->method('publish')->willReturn(new PublishResult(1, 1, true, [], []));
137+
138+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
139+
$session->addSubscription(1, 500.0);
140+
$this->store->create($session);
141+
142+
$this->publisher->startSession('s1');
143+
drainAutoPublishCallbacks($this->callbacks, 1);
144+
145+
expect($this->callbacks[0]['delay'])->toBe(0.01);
146+
});
147+
148+
it('stops when session has no subscriptions', function () {
149+
$client = $this->createStub(Client::class);
150+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
151+
$this->store->create($session);
152+
153+
$this->publisher->startSession('s1');
154+
drainAutoPublishCallbacks($this->callbacks, 1);
155+
156+
expect($this->publisher->isActive('s1'))->toBeFalse();
157+
});
158+
159+
it('stops when session is removed from store', function () {
160+
$client = $this->createStub(Client::class);
161+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
162+
$session->addSubscription(1, 500.0);
163+
$this->store->create($session);
164+
165+
$this->publisher->startSession('s1');
166+
$this->store->remove('s1');
167+
drainAutoPublishCallbacks($this->callbacks, 1);
168+
169+
expect($this->publisher->isActive('s1'))->toBeFalse();
170+
});
171+
172+
it('attempts recovery on ConnectionException', function () {
173+
$client = $this->createStub(Client::class);
174+
$client->method('publish')->willThrowException(new ConnectionException('Connection lost'));
175+
176+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
177+
$session->addSubscription(1, 500.0);
178+
$this->store->create($session);
179+
180+
$this->publisher->startSession('s1');
181+
drainAutoPublishCallbacks($this->callbacks, 1);
182+
183+
expect($this->recoveryCalled)->toBeTrue();
184+
expect($this->publisher->isActive('s1'))->toBeTrue();
185+
expect($this->callbacks[0]['delay'])->toBe(1.0);
186+
});
187+
188+
it('stops after failed recovery', function () {
189+
$this->recoveryResult = false;
190+
191+
$client = $this->createStub(Client::class);
192+
$client->method('publish')->willThrowException(new ConnectionException('Connection lost'));
193+
194+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
195+
$session->addSubscription(1, 500.0);
196+
$this->store->create($session);
197+
198+
$this->publisher->startSession('s1');
199+
drainAutoPublishCallbacks($this->callbacks, 1);
200+
201+
expect($this->recoveryCalled)->toBeTrue();
202+
expect($this->publisher->isActive('s1'))->toBeFalse();
203+
});
204+
205+
it('stops after max consecutive generic errors', function () {
206+
$callCount = 0;
207+
$client = $this->createStub(Client::class);
208+
$client->method('publish')->willReturnCallback(function () use (&$callCount) {
209+
$callCount++;
210+
throw new RuntimeException('Generic error');
211+
});
212+
213+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
214+
$session->addSubscription(1, 500.0);
215+
$this->store->create($session);
216+
217+
$this->publisher->startSession('s1');
218+
drainAutoPublishCallbacks($this->callbacks, 10);
219+
220+
expect($callCount)->toBe(5);
221+
expect($this->publisher->isActive('s1'))->toBeFalse();
222+
});
223+
224+
it('uses backoff delay on generic errors', function () {
225+
$client = $this->createStub(Client::class);
226+
$client->method('publish')->willThrowException(new RuntimeException('Error'));
227+
228+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
229+
$session->addSubscription(1, 500.0);
230+
$this->store->create($session);
231+
232+
$this->publisher->startSession('s1');
233+
drainAutoPublishCallbacks($this->callbacks, 1);
234+
235+
expect($this->callbacks[0]['delay'])->toBe(5.0);
236+
});
237+
238+
it('resets error count on successful publish', function () {
239+
$callCount = 0;
240+
$client = $this->createStub(Client::class);
241+
$client->method('publish')->willReturnCallback(function () use (&$callCount) {
242+
$callCount++;
243+
if ($callCount <= 3) {
244+
throw new RuntimeException('Transient error');
245+
}
246+
return new PublishResult(1, $callCount, false, [], []);
247+
});
248+
249+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
250+
$session->addSubscription(1, 500.0);
251+
$this->store->create($session);
252+
253+
$this->publisher->startSession('s1');
254+
drainAutoPublishCallbacks($this->callbacks, 10);
255+
256+
expect($callCount)->toBeGreaterThan(3);
257+
expect($this->publisher->isActive('s1'))->toBeTrue();
258+
});
259+
260+
it('passes acknowledgements from previous publish', function () {
261+
$receivedAcks = [];
262+
$callCount = 0;
263+
$client = $this->createStub(Client::class);
264+
$client->method('publish')->willReturnCallback(function (array $acks) use (&$receivedAcks, &$callCount) {
265+
$callCount++;
266+
$receivedAcks[] = $acks;
267+
if ($callCount >= 3) {
268+
throw new RuntimeException('Stop');
269+
}
270+
return new PublishResult(1, $callCount * 10, false, [], []);
271+
});
272+
273+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
274+
$session->addSubscription(1, 100.0);
275+
$this->store->create($session);
276+
277+
$this->publisher->startSession('s1');
278+
drainAutoPublishCallbacks($this->callbacks, 5);
279+
280+
expect($receivedAcks[0])->toBe([]);
281+
expect($receivedAcks[1])->toBe([['subscriptionId' => 1, 'sequenceNumber' => 10]]);
282+
expect($receivedAcks[2])->toBe([['subscriptionId' => 1, 'sequenceNumber' => 20]]);
283+
});
284+
285+
it('touches the session on successful publish', function () {
286+
$client = $this->createStub(Client::class);
287+
$client->method('publish')->willReturn(new PublishResult(1, 10, false, [], []));
288+
289+
$oldTime = microtime(true) - 100;
290+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], $oldTime);
291+
$session->addSubscription(1, 500.0);
292+
$this->store->create($session);
293+
294+
$this->publisher->startSession('s1');
295+
drainAutoPublishCallbacks($this->callbacks, 1);
296+
297+
expect($session->lastUsed)->toBeGreaterThan($oldTime);
298+
});
299+
300+
it('clears pending acks on connection error', function () {
301+
$callCount = 0;
302+
$receivedAcks = [];
303+
$client = $this->createStub(Client::class);
304+
$client->method('publish')->willReturnCallback(function (array $acks) use (&$callCount, &$receivedAcks) {
305+
$callCount++;
306+
$receivedAcks[] = $acks;
307+
if ($callCount === 2) {
308+
throw new ConnectionException('Lost');
309+
}
310+
return new PublishResult(1, $callCount * 10, false, [], []);
311+
});
312+
313+
$session = new Session('s1', $client, 'opc.tcp://localhost:4840', [], microtime(true));
314+
$session->addSubscription(1, 500.0);
315+
$this->store->create($session);
316+
317+
$this->publisher->startSession('s1');
318+
drainAutoPublishCallbacks($this->callbacks, 3);
319+
320+
expect($receivedAcks[2])->toBe([]);
321+
});
322+
323+
});

0 commit comments

Comments
 (0)