Skip to content

Commit d4c5c2a

Browse files
GianfriAurGianfriAur
committed
[WIP] AutoPublisher
1 parent 419cf3a commit d4c5c2a

4 files changed

Lines changed: 476 additions & 11 deletions

File tree

‎src/Daemon/AutoPublisher.php‎

Lines changed: 248 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,248 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace PhpOpcua\SessionManager\Daemon;
6+
7+
use Closure;
8+
use PhpOpcua\Client\Exception\ConnectionException;
9+
use Psr\Log\LoggerInterface;
10+
use React\EventLoop\LoopInterface;
11+
use React\EventLoop\TimerInterface;
12+
use Throwable;
13+
14+
/**
15+
* Manages automatic publish cycles for sessions with active subscriptions.
16+
*
17+
* For each session that has at least one subscription, a self-rescheduling one-shot timer
18+
* calls {@see \PhpOpcua\Client\Client::publish()} periodically. The client's internal
19+
* {@see \PhpOpcua\Client\Client\ManagesSubscriptionsTrait::dispatchPublishEvents()} dispatches
20+
* PSR-14 events (DataChangeReceived, EventNotificationReceived, etc.) automatically.
21+
*
22+
* Acknowledgements are tracked internally and sent on each subsequent publish call.
23+
*/
24+
class AutoPublisher
25+
{
26+
private const MAX_CONSECUTIVE_ERRORS = 5;
27+
28+
/** @var array<string, TimerInterface|null> */
29+
private array $timers = [];
30+
31+
/** @var array<string, array{subscriptionId: int, sequenceNumber: int}[]> */
32+
private array $pendingAcks = [];
33+
34+
/** @var array<string, int> */
35+
private array $errorCounts = [];
36+
37+
/**
38+
* @param SessionStore $store The in-memory session registry.
39+
* @param LoopInterface $loop The ReactPHP event loop.
40+
* @param LoggerInterface $logger Logger for auto-publish lifecycle events.
41+
* @param Closure(Session): bool $recoveryCallback Called on ConnectionException to attempt session recovery.
42+
*/
43+
public function __construct(
44+
private readonly SessionStore $store,
45+
private readonly LoopInterface $loop,
46+
private readonly LoggerInterface $logger,
47+
private readonly Closure $recoveryCallback,
48+
)
49+
{
50+
}
51+
52+
/**
53+
* Start automatic publishing for a session.
54+
*
55+
* Idempotent — calling this on a session that is already active has no effect.
56+
*
57+
* @param string $sessionId
58+
* @return void
59+
*/
60+
public function startSession(string $sessionId): void
61+
{
62+
if (isset($this->timers[$sessionId])) {
63+
return;
64+
}
65+
66+
$this->logger->info('Auto-publish started for session {id}', ['id' => $sessionId]);
67+
$this->scheduleNext($sessionId, 0.0);
68+
}
69+
70+
/**
71+
* Stop automatic publishing for a session and clean up internal state.
72+
*
73+
* @param string $sessionId
74+
* @return void
75+
*/
76+
public function stopSession(string $sessionId): void
77+
{
78+
if (isset($this->timers[$sessionId])) {
79+
$this->loop->cancelTimer($this->timers[$sessionId]);
80+
}
81+
82+
unset($this->timers[$sessionId], $this->pendingAcks[$sessionId], $this->errorCounts[$sessionId]);
83+
84+
$this->logger->info('Auto-publish stopped for session {id}', ['id' => $sessionId]);
85+
}
86+
87+
/**
88+
* Check whether auto-publish is currently active for a session.
89+
*
90+
* @param string $sessionId
91+
* @return bool
92+
*/
93+
public function isActive(string $sessionId): bool
94+
{
95+
return isset($this->timers[$sessionId]);
96+
}
97+
98+
/**
99+
* Stop automatic publishing for all active sessions.
100+
*
101+
* @return void
102+
*/
103+
public function stopAll(): void
104+
{
105+
foreach (array_keys($this->timers) as $sessionId) {
106+
$this->stopSession($sessionId);
107+
}
108+
}
109+
110+
/**
111+
* Schedule the next publish cycle for a session after a delay.
112+
*
113+
* @param string $sessionId
114+
* @param float $delay Delay in seconds before the next publish cycle.
115+
* @return void
116+
*/
117+
private function scheduleNext(string $sessionId, float $delay): void
118+
{
119+
$this->timers[$sessionId] = $this->loop->addTimer($delay, function () use ($sessionId) {
120+
unset($this->timers[$sessionId]);
121+
$this->publishCycle($sessionId);
122+
});
123+
}
124+
125+
/**
126+
* Execute a single publish cycle for a session.
127+
*
128+
* Calls {@see \PhpOpcua\Client\Client::publish()} with accumulated acknowledgements,
129+
* processes the result, and schedules the next cycle. On connection failure, attempts
130+
* recovery via the configured callback. On repeated errors, stops auto-publish.
131+
*
132+
* @param string $sessionId
133+
* @return void
134+
*/
135+
private function publishCycle(string $sessionId): void
136+
{
137+
try {
138+
$session = $this->store->get($sessionId);
139+
} catch (Throwable) {
140+
return;
141+
}
142+
143+
if (!$session->hasSubscriptions()) {
144+
$this->stopSession($sessionId);
145+
return;
146+
}
147+
148+
$acks = $this->pendingAcks[$sessionId] ?? [];
149+
$this->pendingAcks[$sessionId] = [];
150+
151+
try {
152+
$result = $session->client->publish($acks);
153+
154+
$session->touch();
155+
$this->pendingAcks[$sessionId][] = [
156+
'subscriptionId' => $result->subscriptionId,
157+
'sequenceNumber' => $result->sequenceNumber,
158+
];
159+
$this->errorCounts[$sessionId] = 0;
160+
161+
$this->scheduleNextBasedOnResult($sessionId, $session, $result->moreNotifications);
162+
} catch (ConnectionException $e) {
163+
$this->handleConnectionError($sessionId, $session, $e);
164+
} catch (Throwable $e) {
165+
$this->handleGenericError($sessionId, $e);
166+
}
167+
}
168+
169+
/**
170+
* Schedule the next publish based on whether more notifications are available.
171+
*
172+
* @param string $sessionId
173+
* @param Session $session
174+
* @param bool $moreNotifications
175+
* @return void
176+
*/
177+
private function scheduleNextBasedOnResult(string $sessionId, Session $session, bool $moreNotifications): void
178+
{
179+
if ($moreNotifications) {
180+
$this->scheduleNext($sessionId, 0.01);
181+
return;
182+
}
183+
184+
$this->scheduleNext($sessionId, $session->getMinPublishingInterval() * 0.75);
185+
}
186+
187+
/**
188+
* Handle a connection error during publish by attempting session recovery.
189+
*
190+
* @param string $sessionId
191+
* @param Session $session
192+
* @param ConnectionException $e
193+
* @return void
194+
*/
195+
private function handleConnectionError(string $sessionId, Session $session, ConnectionException $e): void
196+
{
197+
$this->logger->warning('Auto-publish connection error for session {id}: {message}', [
198+
'id' => $sessionId,
199+
'message' => $e->getMessage(),
200+
]);
201+
202+
$this->pendingAcks[$sessionId] = [];
203+
204+
try {
205+
$recovered = ($this->recoveryCallback)($session);
206+
} catch (Throwable) {
207+
$recovered = false;
208+
}
209+
210+
if ($recovered && $session->hasSubscriptions()) {
211+
$this->scheduleNext($sessionId, 1.0);
212+
return;
213+
}
214+
215+
$this->logger->error('Auto-publish recovery failed for session {id}, stopping', ['id' => $sessionId]);
216+
$this->stopSession($sessionId);
217+
}
218+
219+
/**
220+
* Handle a non-connection error during publish with exponential backoff.
221+
*
222+
* @param string $sessionId
223+
* @param Throwable $e
224+
* @return void
225+
*/
226+
private function handleGenericError(string $sessionId, Throwable $e): void
227+
{
228+
$this->errorCounts[$sessionId] = ($this->errorCounts[$sessionId] ?? 0) + 1;
229+
230+
$this->logger->warning('Auto-publish error for session {id} ({count}/{max}): {message}', [
231+
'id' => $sessionId,
232+
'count' => $this->errorCounts[$sessionId],
233+
'max' => self::MAX_CONSECUTIVE_ERRORS,
234+
'message' => $e->getMessage(),
235+
]);
236+
237+
if ($this->errorCounts[$sessionId] >= self::MAX_CONSECUTIVE_ERRORS) {
238+
$this->logger->error('Auto-publish stopped for session {id} after {max} consecutive errors', [
239+
'id' => $sessionId,
240+
'max' => self::MAX_CONSECUTIVE_ERRORS,
241+
]);
242+
$this->stopSession($sessionId);
243+
return;
244+
}
245+
246+
$this->scheduleNext($sessionId, 5.0);
247+
}
248+
}

0 commit comments

Comments
 (0)