9555551525

Committed 17 Jun 2024 10:16PM UTC coverage: 70.262% (-1.1%) from 71.395%

Build # 9555551525

Build Type

push

github

Committed by

prolic

Commit Message

update coveralls repo token

Run Details

3466 of 4933 relevant lines covered (70.26%)

67.7 hits per line

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

35.64

/src/Internal/SubscriptionsManager.php

<?php

/**
 * This file is part of `prooph/event-store-client`.
 * (c) 2018-2024 Alexander Miertsch <kontakt@codeliner.ws>
 * (c) 2018-2024 Sascha-Oliver Prolic <saschaprolic@googlemail.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

declare(strict_types=1);

namespace Prooph\EventStoreClient\Internal;

use Prooph\EventStore\Exception\ConnectionClosed;
use Prooph\EventStore\Exception\OperationTimedOut;
use Prooph\EventStore\Exception\RetriesLimitReached;
use Prooph\EventStore\SubscriptionDropReason;
use Prooph\EventStore\Util\DateTime;
use Prooph\EventStore\Util\Guid;
use Prooph\EventStoreClient\ConnectionSettings;
use Prooph\EventStoreClient\Transport\Tcp\TcpPackageConnection;
use SplQueue;

/** @internal  */
class SubscriptionsManager
{
    /** @var array<string, SubscriptionItem> */
    private array $activeSubscriptions = [];

    private SplQueue $waitingSubscriptions;

    /** @var list<SubscriptionItem> */
    private array $retryPendingSubscriptions = [];

    public function __construct(private string $connectionName, private ConnectionSettings $settings)
    {
        $this->waitingSubscriptions = new SplQueue();
    }

    public function getActiveSubscription(string $correlationId): ?SubscriptionItem
    {
        return $this->activeSubscriptions[$correlationId] ?? null;
    }

    public function cleanUp(): void
    {
        $connectionClosedException = ConnectionClosed::withName($this->connectionName);

        foreach ($this->activeSubscriptions as $subscriptionItem) {
            $subscriptionItem->operation()->dropSubscription(
                SubscriptionDropReason::ConnectionClosed,
                $connectionClosedException
            );
        }

        while (! $this->waitingSubscriptions->isEmpty()) {
            $subscriptionItem = $this->waitingSubscriptions->dequeue();
            \assert($subscriptionItem instanceof SubscriptionItem);
            $subscriptionItem->operation()->dropSubscription(
                SubscriptionDropReason::ConnectionClosed,
                $connectionClosedException
            );
        }

        foreach ($this->retryPendingSubscriptions as $subscriptionItem) {
            $subscriptionItem->operation()->dropSubscription(
                SubscriptionDropReason::ConnectionClosed,
                $connectionClosedException
            );
        }

        $this->activeSubscriptions = [];
        $this->retryPendingSubscriptions = [];
    }

    public function purgeSubscribedAndDroppedSubscriptions(string $connectionId): void
    {
        $subscriptionsToRemove = new SplQueue();

        foreach ($this->activeSubscriptions as $subscriptionItem) {
            if ($subscriptionItem->connectionId() !== $connectionId) {
                continue;
            }

            $subscriptionItem->operation()->connectionClosed();
            $subscriptionsToRemove->enqueue($subscriptionItem);
        }

        while (! $subscriptionsToRemove->isEmpty()) {
            $subscriptionItem = $subscriptionsToRemove->dequeue();
            \assert($subscriptionItem instanceof SubscriptionItem);
            unset($this->activeSubscriptions[$subscriptionItem->correlationId()]);
        }
    }

    public function checkTimeoutsAndRetry(TcpPackageConnection $connection): void
    {
        $retrySubscriptions = new SplQueue();
        $removeSubscriptions = new SplQueue();

        foreach ($this->activeSubscriptions as $subscription) {
            if ($subscription->isSubscribed()) {
                continue;
            }

            if ($subscription->connectionId() !== $connection->connectionId()) {
                $this->retryPendingSubscriptions[] = $subscription;
            } elseif ($subscription->timeout() > 0
                && (float) DateTime::utcNow()->format('U.u') - (float) $subscription->lastUpdated()->format('U.u') > $this->settings->operationTimeout()
            ) {
                $err = \sprintf(
                    'EventStoreNodeConnection \'%s\': subscription never got confirmation from server',
                    $connection->connectionId()
                );

                $this->settings->log()->error($err);

                if ($this->settings->failOnNoServerResponse()) {
                    $subscription->operation()->dropSubscription(
                        SubscriptionDropReason::SubscribingError,
                        new OperationTimedOut($err)
                    );
                    $removeSubscriptions->enqueue($subscription);
                } else {
                    $retrySubscriptions->enqueue($subscription);
                }
            }
        }

        while (! $retrySubscriptions->isEmpty()) {
            $this->scheduleSubscriptionRetry($retrySubscriptions->dequeue());
        }

        while (! $removeSubscriptions->isEmpty()) {
            $this->removeSubscription($removeSubscriptions->dequeue());
        }

        if (\count($this->retryPendingSubscriptions) > 0) {
            foreach ($this->retryPendingSubscriptions as $subscription) {
                $subscription->incRetryCount();
                $this->startSubscription($subscription, $connection);
            }

            $this->retryPendingSubscriptions = [];
        }

        while (! $this->waitingSubscriptions->isEmpty()) {
            $this->startSubscription($this->waitingSubscriptions->dequeue(), $connection);
        }
    }

    public function removeSubscription(SubscriptionItem $subscription): bool
    {
        $result = isset($this->activeSubscriptions[$subscription->correlationId()]);
        $this->logDebug('RemoveSubscription %s, result %s', (string) $subscription, $result ? 'yes' : 'no');
        unset($this->activeSubscriptions[$subscription->correlationId()]);

        return $result;
    }

    public function scheduleSubscriptionRetry(SubscriptionItem $subscription): void
    {
        if (! $this->removeSubscription($subscription)) {
            $this->logDebug('RemoveSubscription failed when trying to retry %s', (string) $subscription);

            return;
        }

        if ($subscription->maxRetries() >= 0 && $subscription->retryCount() >= $subscription->maxRetries()) {
            $this->logDebug('RETRIES LIMIT REACHED when trying to retry %s', (string) $subscription);
            $subscription->operation()->dropSubscription(
                SubscriptionDropReason::SubscribingError,
                RetriesLimitReached::with($subscription->retryCount())
            );

            return;
        }

        $this->logDebug('retrying subscription %s', (string) $subscription);
        $this->retryPendingSubscriptions[] = $subscription;
    }

    public function enqueueSubscription(SubscriptionItem $subscriptionItem): void
    {
        $this->waitingSubscriptions->enqueue($subscriptionItem);
    }

    public function startSubscription(SubscriptionItem $subscription, TcpPackageConnection $connection): void
    {
        if ($subscription->isSubscribed()) {
            $this->logDebug('StartSubscription REMOVING due to already subscribed %s', (string) $subscription);
            $this->removeSubscription($subscription);

            return;
        }

        $correlationId = Guid::generateAsHex();
        $subscription->setCorrelationId($correlationId);
        $subscription->setConnectionId($connection->connectionId());
        $subscription->setLastUpdated(DateTime::utcNow());

        $this->activeSubscriptions[$correlationId] = $subscription;

        if (! $subscription->operation()->subscribe($correlationId, $connection)) {
            $this->logDebug('StartSubscription REMOVING AS COULD NOT SUBSCRIBE %s', (string) $subscription);
            $this->removeSubscription($subscription);
        }
        $this->logDebug('StartSubscription SUBSCRIBING %s', (string) $subscription);
    }

    private function logDebug(string $message, string ...$parameters): void
    {
        if ($this->settings->verboseLogging()) {
            $message = empty($parameters)
                ? $message
                : \sprintf($message, ...$parameters);

            $this->settings->log()->debug(\sprintf(
                'EventStoreNodeConnection \'%s\': %s',
                $this->connectionName,
                $message
            ));
        }
    }
}

1	<?php
2
3	/**
4	* This file is part of `prooph/event-store-client`.
5	* (c) 2018-2024 Alexander Miertsch <kontakt@codeliner.ws>
6	* (c) 2018-2024 Sascha-Oliver Prolic <saschaprolic@googlemail.com>
7	*
8	* For the full copyright and license information, please view the LICENSE
9	* file that was distributed with this source code.
10	*/
11
12	declare(strict_types=1);
13
14	namespace Prooph\EventStoreClient\Internal;
15
16	use Prooph\EventStore\Exception\ConnectionClosed;
17	use Prooph\EventStore\Exception\OperationTimedOut;
18	use Prooph\EventStore\Exception\RetriesLimitReached;
19	use Prooph\EventStore\SubscriptionDropReason;
20	use Prooph\EventStore\Util\DateTime;
21	use Prooph\EventStore\Util\Guid;
22	use Prooph\EventStoreClient\ConnectionSettings;
23	use Prooph\EventStoreClient\Transport\Tcp\TcpPackageConnection;
24	use SplQueue;
25
26	/** @internal */
27	class SubscriptionsManager
28	{
29	/** @var array<string, SubscriptionItem> */
30	private array $activeSubscriptions = [];
31
32	private SplQueue $waitingSubscriptions;
33
34	/** @var list<SubscriptionItem> */
35	private array $retryPendingSubscriptions = [];
36
37	public function __construct(private string $connectionName, private ConnectionSettings $settings)
38	{
39	$this->waitingSubscriptions = new SplQueue();	420✔
40	}
41
42	public function getActiveSubscription(string $correlationId): ?SubscriptionItem
43	{
44	return $this->activeSubscriptions[$correlationId] ?? null;	87✔
45	}
46
47	public function cleanUp(): void
48	{
49	$connectionClosedException = ConnectionClosed::withName($this->connectionName);	408✔
50
51	foreach ($this->activeSubscriptions as $subscriptionItem) {	408✔
52	$subscriptionItem->operation()->dropSubscription(	72✔
53	SubscriptionDropReason::ConnectionClosed,	72✔
54	$connectionClosedException	72✔
55	);	72✔
56	}
57
58	while (! $this->waitingSubscriptions->isEmpty()) {	408✔
59	$subscriptionItem = $this->waitingSubscriptions->dequeue();	×
60	\assert($subscriptionItem instanceof SubscriptionItem);
61	$subscriptionItem->operation()->dropSubscription(	×
62	SubscriptionDropReason::ConnectionClosed,	×
63	$connectionClosedException	×
64	);	×
65	}
66
67	foreach ($this->retryPendingSubscriptions as $subscriptionItem) {	408✔
68	$subscriptionItem->operation()->dropSubscription(	×
69	SubscriptionDropReason::ConnectionClosed,	×
70	$connectionClosedException	×
71	);	×
72	}
73
74	$this->activeSubscriptions = [];	408✔
75	$this->retryPendingSubscriptions = [];	408✔
76	}
77
78	public function purgeSubscribedAndDroppedSubscriptions(string $connectionId): void
79	{
80	$subscriptionsToRemove = new SplQueue();	×
81
82	foreach ($this->activeSubscriptions as $subscriptionItem) {	×
83	if ($subscriptionItem->connectionId() !== $connectionId) {	×
84	continue;	×
85	}
86
87	$subscriptionItem->operation()->connectionClosed();	×
88	$subscriptionsToRemove->enqueue($subscriptionItem);	×
89	}
90
91	while (! $subscriptionsToRemove->isEmpty()) {	×
92	$subscriptionItem = $subscriptionsToRemove->dequeue();	×
93	\assert($subscriptionItem instanceof SubscriptionItem);
94	unset($this->activeSubscriptions[$subscriptionItem->correlationId()]);	×
95	}
96	}
97
98	public function checkTimeoutsAndRetry(TcpPackageConnection $connection): void
99	{
100	$retrySubscriptions = new SplQueue();	399✔
101	$removeSubscriptions = new SplQueue();	399✔
102
103	foreach ($this->activeSubscriptions as $subscription) {	399✔
104	if ($subscription->isSubscribed()) {	8✔
105	continue;	8✔
106	}
107
108	if ($subscription->connectionId() !== $connection->connectionId()) {	×
109	$this->retryPendingSubscriptions[] = $subscription;	×
110	} elseif ($subscription->timeout() > 0	×
111	&& (float) DateTime::utcNow()->format('U.u') - (float) $subscription->lastUpdated()->format('U.u') > $this->settings->operationTimeout()	×
112	) {
113	$err = \sprintf(	×
114	'EventStoreNodeConnection \'%s\': subscription never got confirmation from server',	×
115	$connection->connectionId()	×
116	);	×
117
118	$this->settings->log()->error($err);	×
119
120	if ($this->settings->failOnNoServerResponse()) {	×
121	$subscription->operation()->dropSubscription(	×
122	SubscriptionDropReason::SubscribingError,	×
123	new OperationTimedOut($err)	×
124	);	×
125	$removeSubscriptions->enqueue($subscription);	×
126	} else {
127	$retrySubscriptions->enqueue($subscription);	×
128	}
129	}
130	}
131
132	while (! $retrySubscriptions->isEmpty()) {	399✔
133	$this->scheduleSubscriptionRetry($retrySubscriptions->dequeue());	×
134	}
135
136	while (! $removeSubscriptions->isEmpty()) {	399✔
137	$this->removeSubscription($removeSubscriptions->dequeue());	×
138	}
139
140	if (\count($this->retryPendingSubscriptions) > 0) {	399✔
141	foreach ($this->retryPendingSubscriptions as $subscription) {	×
142	$subscription->incRetryCount();	×
143	$this->startSubscription($subscription, $connection);	×
144	}
145
146	$this->retryPendingSubscriptions = [];	×
147	}
148
149	while (! $this->waitingSubscriptions->isEmpty()) {	399✔
150	$this->startSubscription($this->waitingSubscriptions->dequeue(), $connection);	23✔
151	}
152	}
153
154	public function removeSubscription(SubscriptionItem $subscription): bool
155	{
156	$result = isset($this->activeSubscriptions[$subscription->correlationId()]);	19✔
157	$this->logDebug('RemoveSubscription %s, result %s', (string) $subscription, $result ? 'yes' : 'no');	19✔
158	unset($this->activeSubscriptions[$subscription->correlationId()]);	19✔
159
160	return $result;	19✔
161	}
162
163	public function scheduleSubscriptionRetry(SubscriptionItem $subscription): void
164	{
165	if (! $this->removeSubscription($subscription)) {	×
166	$this->logDebug('RemoveSubscription failed when trying to retry %s', (string) $subscription);	×
167
168	return;	×
169	}
170
171	if ($subscription->maxRetries() >= 0 && $subscription->retryCount() >= $subscription->maxRetries()) {	×
172	$this->logDebug('RETRIES LIMIT REACHED when trying to retry %s', (string) $subscription);	×
173	$subscription->operation()->dropSubscription(	×
174	SubscriptionDropReason::SubscribingError,	×
175	RetriesLimitReached::with($subscription->retryCount())	×
176	);	×
177
178	return;	×
179	}
180
181	$this->logDebug('retrying subscription %s', (string) $subscription);	×
182	$this->retryPendingSubscriptions[] = $subscription;	×
183	}
184
185	public function enqueueSubscription(SubscriptionItem $subscriptionItem): void
186	{
187	$this->waitingSubscriptions->enqueue($subscriptionItem);	23✔
188	}
189
190	public function startSubscription(SubscriptionItem $subscription, TcpPackageConnection $connection): void
191	{
192	if ($subscription->isSubscribed()) {	87✔
193	$this->logDebug('StartSubscription REMOVING due to already subscribed %s', (string) $subscription);	×
194	$this->removeSubscription($subscription);	×
195
196	return;	×
197	}
198
199	$correlationId = Guid::generateAsHex();	87✔
200	$subscription->setCorrelationId($correlationId);	87✔
201	$subscription->setConnectionId($connection->connectionId());	87✔
202	$subscription->setLastUpdated(DateTime::utcNow());	87✔
203
204	$this->activeSubscriptions[$correlationId] = $subscription;	87✔
205
206	if (! $subscription->operation()->subscribe($correlationId, $connection)) {	87✔
207	$this->logDebug('StartSubscription REMOVING AS COULD NOT SUBSCRIBE %s', (string) $subscription);	×
208	$this->removeSubscription($subscription);	×
209	}
210	$this->logDebug('StartSubscription SUBSCRIBING %s', (string) $subscription);	87✔
211	}
212
213	private function logDebug(string $message, string ...$parameters): void
214	{
215	if ($this->settings->verboseLogging()) {	87✔
216	$message = empty($parameters)	×
217	? $message	×
218	: \sprintf($message, ...$parameters);	×
219
220	$this->settings->log()->debug(\sprintf(	×
221	'EventStoreNodeConnection \'%s\': %s',	×
222	$this->connectionName,	×
223	$message	×
224	));	×
225	}
226	}
227	}

prooph / event-store-client / 9555551525

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