Skip to content

Commit 94155dc

Browse files
tatevikg1tatevikg1
committed
SendRateLimiter
1 parent 351f73a commit 94155dc

File tree

6 files changed

+217
-87
lines changed

6 files changed

+217
-87
lines changed

‎config/services/services.yml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@ services:
3535
autowire: true
3636
autoconfigure: true
3737
public: true
38+
39+
PhpList\Core\Domain\Messaging\Service\SendRateLimiter:
40+
autowire: true
41+
autoconfigure: true
3842
arguments:
3943
$mailqueueBatchSize: '%messaging.mail_queue_batch_size%'
4044
$mailqueueBatchPeriod: '%messaging.mail_queue_period%'

‎src/Domain/Messaging/Service/Processor/CampaignProcessor.php‎

Lines changed: 6 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
use Doctrine\ORM\EntityManagerInterface;
88
use PhpList\Core\Domain\Messaging\Model\Message;
99
use PhpList\Core\Domain\Messaging\Service\MessageProcessingPreparator;
10-
use PhpList\Core\Domain\Common\IspRestrictionsProvider;
10+
use PhpList\Core\Domain\Messaging\Service\SendRateLimiter;
1111
use PhpList\Core\Domain\Subscription\Service\Provider\SubscriberProvider;
1212
use Psr\Log\LoggerInterface;
1313
use Symfony\Component\Console\Output\OutputInterface;
@@ -22,77 +22,33 @@ class CampaignProcessor
2222
private SubscriberProvider $subscriberProvider;
2323
private MessageProcessingPreparator $messagePreparator;
2424
private LoggerInterface $logger;
25-
private ?IspRestrictionsProvider $ispRestrictionsProvider;
26-
private ?int $mailqueueBatchSize;
27-
private ?int $mailqueueBatchPeriod;
28-
private ?int $mailqueueThrottle;
25+
private SendRateLimiter $rateLimiter;
2926

3027
public function __construct(
3128
MailerInterface $mailer,
3229
EntityManagerInterface $entityManager,
3330
SubscriberProvider $subscriberProvider,
3431
MessageProcessingPreparator $messagePreparator,
3532
LoggerInterface $logger,
36-
?IspRestrictionsProvider $ispRestrictionsProvider = null,
37-
?int $mailqueueBatchSize = null,
38-
?int $mailqueueBatchPeriod = null,
39-
?int $mailqueueThrottle = null,
33+
SendRateLimiter $rateLimiter,
4034
) {
4135
$this->mailer = $mailer;
4236
$this->entityManager = $entityManager;
4337
$this->subscriberProvider = $subscriberProvider;
4438
$this->messagePreparator = $messagePreparator;
4539
$this->logger = $logger;
46-
$this->ispRestrictionsProvider = $ispRestrictionsProvider;
47-
$this->mailqueueBatchSize = $mailqueueBatchSize;
48-
$this->mailqueueBatchPeriod = $mailqueueBatchPeriod;
49-
$this->mailqueueThrottle = $mailqueueThrottle;
40+
$this->rateLimiter = $rateLimiter;
5041
}
5142

5243
public function process(Message $campaign, ?OutputInterface $output = null): void
5344
{
5445
$this->updateMessageStatus($campaign, Message\MessageStatus::Prepared);
55-
$ispRestrictions = $this->ispRestrictionsProvider->load();
5646
$subscribers = $this->subscriberProvider->getSubscribersForMessage($campaign);
5747

58-
$cfgBatch = ($this->mailqueueBatchSize ?? 0);
59-
$ispMax = isset($ispRestrictions->maxBatch) ? (int)$ispRestrictions->maxBatch : null;
60-
61-
$cfgPeriod = ($this->mailqueueBatchPeriod ?? 0);
62-
$ispMinPeriod = ($ispRestrictions->minBatchPeriod ?? 0);
63-
64-
$cfgThrottle = ($this->mailqueueThrottle ?? 0);
65-
$ispMinThrottle = (int)($ispRestrictions->minThrottle ?? 0);
66-
67-
if ($cfgBatch <= 0) {
68-
$batchSize = $ispMax !== null ? max(0, $ispMax) : 0;
69-
} else {
70-
$batchSize = $ispMax !== null ? min($cfgBatch, max(1, $ispMax)) : $cfgBatch;
71-
}
72-
73-
$batchPeriod = max(0, $cfgPeriod, $ispMinPeriod);
74-
75-
$throttleSec = max(0, $cfgThrottle, $ispMinThrottle);
76-
77-
$sentInBatch = 0;
78-
$batchStart = microtime(true);
79-
8048
$this->updateMessageStatus($campaign, Message\MessageStatus::InProcess);
8149

8250
foreach ($subscribers as $subscriber) {
83-
if ($batchSize > 0 && $batchPeriod > 0 && $sentInBatch >= $batchSize) {
84-
$elapsed = microtime(true) - $batchStart;
85-
$remaining = (int)ceil($batchPeriod - $elapsed);
86-
if ($remaining > 0) {
87-
$output?->writeln(sprintf(
88-
'Batch limit reached, sleeping %ds to respect MAILQUEUE_BATCH_PERIOD',
89-
$remaining
90-
));
91-
sleep($remaining);
92-
}
93-
$batchStart = microtime(true);
94-
$sentInBatch = 0;
95-
}
51+
$this->rateLimiter->awaitTurn($output);
9652

9753
if (!filter_var($subscriber->getEmail(), FILTER_VALIDATE_EMAIL)) {
9854
continue;
@@ -107,18 +63,14 @@ public function process(Message $campaign, ?OutputInterface $output = null): voi
10763

10864
try {
10965
$this->mailer->send($email);
110-
$sentInBatch++;
66+
$this->rateLimiter->afterSend();
11167
} catch (Throwable $e) {
11268
$this->logger->error($e->getMessage(), [
11369
'subscriber_id' => $subscriber->getId(),
11470
'campaign_id' => $campaign->getId(),
11571
]);
11672
$output?->writeln('Failed to send to: ' . $subscriber->getEmail());
11773
}
118-
119-
if ($throttleSec > 0) {
120-
sleep($throttleSec);
121-
}
12274
}
12375

12476
$this->updateMessageStatus($campaign, Message\MessageStatus::Sent);

‎src/Domain/Messaging/Service/SendRateLimiter.php‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace PhpList\Core\Domain\Messaging\Service;
6+
7+
use PhpList\Core\Domain\Common\IspRestrictionsProvider;
8+
use Symfony\Component\Console\Output\OutputInterface;
9+
10+
/**
11+
* Encapsulates batching and throttling logic for sending emails respecting
12+
* configuration and ISP restrictions.
13+
*/
14+
class SendRateLimiter
15+
{
16+
private int $batchSize;
17+
private int $batchPeriod;
18+
private int $throttleSec;
19+
private int $sentInBatch = 0;
20+
private float $batchStart = 0.0;
21+
22+
public function __construct(
23+
private readonly IspRestrictionsProvider $ispRestrictionsProvider,
24+
private readonly ?int $mailqueueBatchSize = null,
25+
private readonly ?int $mailqueueBatchPeriod = null,
26+
private readonly ?int $mailqueueThrottle = null,
27+
) {
28+
$this->initializeLimits();
29+
}
30+
31+
private function initializeLimits(): void
32+
{
33+
$isp = $this->ispRestrictionsProvider->load();
34+
35+
$cfgBatch = $this->mailqueueBatchSize ?? 0;
36+
$ispMax = isset($isp->maxBatch) ? (int)$isp->maxBatch : null;
37+
38+
$cfgPeriod = $this->mailqueueBatchPeriod ?? 0;
39+
$ispMinPeriod = $isp->minBatchPeriod ?? 0;
40+
41+
$cfgThrottle = $this->mailqueueThrottle ?? 0;
42+
$ispMinThrottle = (int)($isp->minThrottle ?? 0);
43+
44+
if ($cfgBatch <= 0) {
45+
$this->batchSize = $ispMax !== null ? max(0, $ispMax) : 0;
46+
} else {
47+
$this->batchSize = $ispMax !== null ? min($cfgBatch, max(1, $ispMax)) : $cfgBatch;
48+
}
49+
$this->batchPeriod = max(0, $cfgPeriod, $ispMinPeriod);
50+
$this->throttleSec = max(0, $cfgThrottle, $ispMinThrottle);
51+
52+
$this->sentInBatch = 0;
53+
$this->batchStart = microtime(true);
54+
}
55+
56+
/**
57+
* Call before attempting to send another message. It will sleep if needed to
58+
* respect batch limits. Returns true when it's okay to proceed.
59+
*/
60+
public function awaitTurn(?OutputInterface $output = null): bool
61+
{
62+
if ($this->batchSize > 0 && $this->batchPeriod > 0 && $this->sentInBatch >= $this->batchSize) {
63+
$elapsed = microtime(true) - $this->batchStart;
64+
$remaining = (int)ceil($this->batchPeriod - $elapsed);
65+
if ($remaining > 0) {
66+
$output?->writeln(sprintf(
67+
'Batch limit reached, sleeping %ds to respect MAILQUEUE_BATCH_PERIOD',
68+
$remaining
69+
));
70+
sleep($remaining);
71+
}
72+
$this->batchStart = microtime(true);
73+
$this->sentInBatch = 0;
74+
}
75+
return true;
76+
}
77+
78+
/**
79+
* Call after a successful sending to update counters and apply per-message throttle.
80+
*/
81+
public function afterSend(): void
82+
{
83+
$this->sentInBatch++;
84+
if ($this->throttleSec > 0) {
85+
sleep($this->throttleSec);
86+
}
87+
}
88+
}

‎tests/Unit/Domain/Messaging/Command/ProcessQueueCommandTest.php‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -82,8 +82,8 @@ public function testExecuteWithNoCampaigns(): void
8282
->method('ensureCampaignsHaveUuid');
8383

8484
$this->messageRepository->expects($this->once())
85-
->method('findBy')
86-
->with(['status' => 'submitted'])
85+
->method('getByStatusAndEmbargo')
86+
->with($this->anything(), $this->anything())
8787
->willReturn([]);
8888

8989
$this->campaignProcessor->expects($this->never())
@@ -112,8 +112,8 @@ public function testExecuteWithCampaigns(): void
112112
$campaign = $this->createMock(Message::class);
113113

114114
$this->messageRepository->expects($this->once())
115-
->method('findBy')
116-
->with(['status' => 'submitted'])
115+
->method('getByStatusAndEmbargo')
116+
->with($this->anything(), $this->anything())
117117
->willReturn([$campaign]);
118118

119119
$this->campaignProcessor->expects($this->once())
@@ -145,8 +145,8 @@ public function testExecuteWithMultipleCampaigns(): void
145145
$campaign2 = $this->createMock(Message::class);
146146

147147
$this->messageRepository->expects($this->once())
148-
->method('findBy')
149-
->with(['status' => 'submitted'])
148+
->method('getByStatusAndEmbargo')
149+
->with($this->anything(), $this->anything())
150150
->willReturn([$campaign1, $campaign2]);
151151

152152
$this->campaignProcessor->expects($this->exactly(2))
@@ -179,8 +179,8 @@ public function testExecuteWithProcessorException(): void
179179
$campaign = $this->createMock(Message::class);
180180

181181
$this->messageRepository->expects($this->once())
182-
->method('findBy')
183-
->with(['status' => 'submitted'])
182+
->method('getByStatusAndEmbargo')
183+
->with($this->anything(), $this->anything())
184184
->willReturn([$campaign]);
185185

186186
$this->campaignProcessor->expects($this->once())

‎tests/Unit/Domain/Messaging/Service/Processor/CampaignProcessorTest.php‎

Lines changed: 25 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
use PhpList\Core\Domain\Messaging\Model\Message\MessageMetadata;
1212
use PhpList\Core\Domain\Messaging\Service\MessageProcessingPreparator;
1313
use PhpList\Core\Domain\Messaging\Service\Processor\CampaignProcessor;
14+
use PhpList\Core\Domain\Messaging\Service\SendRateLimiter;
1415
use PhpList\Core\Domain\Subscription\Model\Subscriber;
1516
use PhpList\Core\Domain\Subscription\Service\Provider\SubscriberProvider;
1617
use PHPUnit\Framework\MockObject\MockObject;
@@ -29,6 +30,7 @@ class CampaignProcessorTest extends TestCase
2930
private LoggerInterface&MockObject $logger;
3031
private OutputInterface&MockObject $output;
3132
private CampaignProcessor $campaignProcessor;
33+
private SendRateLimiter&MockObject $rateLimiter;
3234

3335
protected function setUp(): void
3436
{
@@ -38,13 +40,17 @@ protected function setUp(): void
3840
$this->messagePreparator = $this->createMock(MessageProcessingPreparator::class);
3941
$this->logger = $this->createMock(LoggerInterface::class);
4042
$this->output = $this->createMock(OutputInterface::class);
43+
$this->rateLimiter = $this->createMock(SendRateLimiter::class);
44+
$this->rateLimiter->method('awaitTurn');
45+
$this->rateLimiter->method('afterSend');
4146

4247
$this->campaignProcessor = new CampaignProcessor(
4348
$this->mailer,
4449
$this->entityManager,
4550
$this->subscriberProvider,
4651
$this->messagePreparator,
47-
$this->logger
52+
$this->logger,
53+
$this->rateLimiter
4854
);
4955
}
5056

@@ -59,11 +65,10 @@ public function testProcessWithNoSubscribers(): void
5965
->with($campaign)
6066
->willReturn([]);
6167

62-
$metadata->expects($this->once())
63-
->method('setStatus')
64-
->with(Message\MessageStatus::Sent);
68+
$metadata->expects($this->atLeastOnce())
69+
->method('setStatus');
6570

66-
$this->entityManager->expects($this->once())
71+
$this->entityManager->expects($this->atLeastOnce())
6772
->method('flush');
6873

6974
$this->mailer->expects($this->never())
@@ -87,11 +92,10 @@ public function testProcessWithInvalidSubscriberEmail(): void
8792
->with($campaign)
8893
->willReturn([$subscriber]);
8994

90-
$metadata->expects($this->once())
91-
->method('setStatus')
92-
->with(Message\MessageStatus::Sent);
95+
$metadata->expects($this->atLeastOnce())
96+
->method('setStatus');
9397

94-
$this->entityManager->expects($this->once())
98+
$this->entityManager->expects($this->atLeastOnce())
9599
->method('flush');
96100

97101
$this->messagePreparator->expects($this->never())
@@ -134,11 +138,10 @@ public function testProcessWithValidSubscriberEmail(): void
134138
return true;
135139
}));
136140

137-
$metadata->expects($this->once())
138-
->method('setStatus')
139-
->with(Message\MessageStatus::Sent);
141+
$metadata->expects($this->atLeastOnce())
142+
->method('setStatus');
140143

141-
$this->entityManager->expects($this->once())
144+
$this->entityManager->expects($this->atLeastOnce())
142145
->method('flush');
143146

144147
$this->campaignProcessor->process($campaign, $this->output);
@@ -181,11 +184,10 @@ public function testProcessWithMailerException(): void
181184
->method('writeln')
182185
->with('Failed to send to: [email protected]');
183186

184-
$metadata->expects($this->once())
185-
->method('setStatus')
186-
->with(Message\MessageStatus::Sent);
187+
$metadata->expects($this->atLeastOnce())
188+
->method('setStatus');
187189

188-
$this->entityManager->expects($this->once())
190+
$this->entityManager->expects($this->atLeastOnce())
189191
->method('flush');
190192

191193
$this->campaignProcessor->process($campaign, $this->output);
@@ -221,11 +223,10 @@ public function testProcessWithMultipleSubscribers(): void
221223
$this->mailer->expects($this->exactly(2))
222224
->method('send');
223225

224-
$metadata->expects($this->once())
225-
->method('setStatus')
226-
->with(Message\MessageStatus::Sent);
226+
$metadata->expects($this->atLeastOnce())
227+
->method('setStatus');
227228

228-
$this->entityManager->expects($this->once())
229+
$this->entityManager->expects($this->atLeastOnce())
229230
->method('flush');
230231

231232
$this->campaignProcessor->process($campaign, $this->output);
@@ -264,11 +265,10 @@ public function testProcessWithNullOutput(): void
264265
'campaign_id' => 123,
265266
]);
266267

267-
$metadata->expects($this->once())
268-
->method('setStatus')
269-
->with(Message\MessageStatus::Sent);
268+
$metadata->expects($this->atLeastOnce())
269+
->method('setStatus');
270270

271-
$this->entityManager->expects($this->once())
271+
$this->entityManager->expects($this->atLeastOnce())
272272
->method('flush');
273273

274274
$this->campaignProcessor->process($campaign, null);

0 commit comments

Comments
 (0)