Cavemanon/piratepoet
8
Fork 0
Code Pull Requests Actions Releases 16 Activity

- Refactors Mailer class

Browse Source 
- Refactors SendingQueue worker class
- Adds Maier router with a send() method + ability to specify sending method
- Updates tests
- Introduces 'stopping' and 'starting' cron states
- Improves cron control mechanism
Closes #276
This commit is contained in:
Vlad
2016-01-05 10:34:57 -05:00
parent 112fe0cd6e
commit f1bf2bb097
30 changed files with 1202 additions and 1433 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
lib/Cron/Workers/SendingQueue.php
View File

@@ -1,11 +1,11 @@
<?php
namespace MailPoet\Cron\Workers;
use MailPoet\Mailer\Mailer;
use MailPoet\Models\Newsletter;
use MailPoet\Models\NewsletterStatistics;
use MailPoet\Models\Subscriber;
use MailPoet\Newsletter\Renderer\Renderer;
use MailPoet\Router\Mailer;
if(!defined('ABSPATH')) exit;
@@ -15,100 +15,110 @@ class SendingQueue {
}
function process() {
$queues =
\MailPoet\Models\SendingQueue::orderByDesc('priority')
->whereNull('deleted_at')
->whereNull('status')
->findResultSet();
foreach($queues as $queue) {
$newsletter = Newsletter::findOne($queue->newsletter_id);
foreach($this->getQueues() as $queue) {
$newsletter = Newsletter::findOne($queue->newsletter_id)
->asArray();
if(!$newsletter) {
continue;
};
$newsletter = $newsletter->asArray();
$mailer = new Mailer($httpRequest = false);
if(!empty($newsletter['sender_address']) &&
!empty($newsletter['sender_name'])
) {
$mailer->fromName = $newsletter['sender_name'];
$mailer->fromEmail = $newsletter['sender_address'];
$mailer->fromNameEmail = sprintf(
'%s <%s>',
$mailer->fromName,
$mailer->fromEmail
);
}
if(!empty($newsletter['reply_to_address']) &&
!empty($newsletter['reply_to_name'])
) {
$mailer->replyToName = $newsletter['reply_to_name'];
$mailer->replyToEmail = $newsletter['reply_to_address'];
$mailer->replyToNameEmail = sprintf(
'%s <%s>',
$mailer->replyToName,
$mailer->replyToEmail
);
}
$mailer->mailer = $mailer->buildMailer();
$renderer = new Renderer(json_decode($newsletter['body'], true));
$newsletter = array(
'subject' => $newsletter['subject'],
'id' => $newsletter['id'],
'body' => array(
'html' => $renderer->renderAll(),
'text' => ''
// TODO: add text body
)
);
$newsletter = $this->renderNewsletter($newsletter);
$mailer = $this->configureMailerForNewsletter($newsletter);
$subscribers = json_decode($queue->subscribers, true);
$subscribersToProcess = $subscribers['to_process'];
if(!isset($subscribers['failed'])) $subscribers['failed'] = array();
if(!isset($subscribers['processed'])) $subscribers['processed'] = array();
if(!isset($subscribers['failed'])) $subscribers['failed'] = array();
foreach(array_chunk($subscribersToProcess, 200) as $subscriberIds) {
$dbSubscribers = Subscriber::whereIn('id', $subscriberIds)
->findArray();
foreach($dbSubscribers as $i => $dbSubscriber) {
foreach($dbSubscribers as $dbSubscriber) {
$this->checkExecutionTimer();
// TODO: replace shortcodes in the newsletter
$result = $mailer->mailer->send(
$newsletter,
$mailer->transformSubscriber($dbSubscriber)
);
$newsletterStatistics = NewsletterStatistics::create();
$newsletterStatistics->subscriber_id = $dbSubscriber['id'];
$newsletterStatistics->newsletter_id = $newsletter['id'];
$newsletterStatistics->queue_id = $queue->id;
$newsletterStatistics->save();
$result = $this->sendNewsletter(
$mailer,
$this->processNewsletter($newsletter),
$dbSubscriber);
if($result) {
$this->updateStatistics($newsletter['id'], $dbSubscriber['id'], $queue->id);
$subscribers['processed'][] = $dbSubscriber['id'];
} else {
$subscribers['failed'][] = $dbSubscriber['id'];
}
$subscribers['to_process'] = array_values(
array_diff(
$subscribers['to_process'],
array_merge($subscribers['processed'], $subscribers['failed'])
)
);
$queue->count_processed =
count($subscribers['processed']) + count($subscribers['failed']);
$queue->count_to_process = count($subscribers['to_process']);
$queue->count_failed = count($subscribers['failed']);
$queue->count_total =
$queue->count_processed + $queue->count_to_process;
if(!$queue->count_to_process) {
$queue->processed_at = date('Y-m-d H:i:s');
$queue->status = 'completed';
}
$queue->subscribers = json_encode($subscribers);
$queue->save();
} else $subscribers['failed'][] = $dbSubscriber['id'];
$this->updateQueue($queue, $subscribers);
}
}
}
}
function processNewsletter($newsletter) {
// TODO: replace shortcodes, etc..
return $newsletter;
}
function sendNewsletter($mailer, $newsletter, $subscriber) {
return $mailer->mailerInstance->send(
$newsletter,
$mailer->transformSubscriber($subscriber)
);
}
function updateStatistics($newsletterId, $subscriberId, $queueId) {
$newsletterStatistics = NewsletterStatistics::create();
$newsletterStatistics->subscriber_id = $newsletterId;
$newsletterStatistics->newsletter_id = $subscriberId;
$newsletterStatistics->queue_id = $queueId;
$newsletterStatistics->save();
}
function updateQueue($queue, $subscribers) {
$subscribers['to_process'] = array_values(
array_diff(
$subscribers['to_process'],
array_merge($subscribers['processed'], $subscribers['failed'])
)
);
$queue->count_processed =
count($subscribers['processed']) + count($subscribers['failed']);
$queue->count_to_process = count($subscribers['to_process']);
$queue->count_failed = count($subscribers['failed']);
$queue->count_total =
$queue->count_processed + $queue->count_to_process;
if(!$queue->count_to_process) {
$queue->processed_at = date('Y-m-d H:i:s');
$queue->status = 'completed';
}
$queue->subscribers = json_encode($subscribers);
$queue->save();
}
function configureMailerForNewsletter($newsletter) {
if(!empty($newsletter['sender_address']) && !empty($newsletter['sender_name'])) {
$sender = array(
'name' => $newsletter['sender_name'],
'address' => $newsletter['sender_address']
);
} else $sender = false;
if(!empty($newsletter['reply_to_address']) && !empty($newsletter['reply_to_name'])) {
$replyTo = array(
'name' => $newsletter['reply_to_name'],
'address' => $newsletter['reply_to_address']
);
} else $replyTo = false;
$mailer = new Mailer($method = false, $sender, $replyTo);
return $mailer;
}
function checkExecutionTimer() {
$elapsedTime = microtime(true) - $this->timer;
if($elapsedTime >= 28) throw new \Exception('Maximum execution time reached.');
if($elapsedTime >= 30) throw new \Exception('Maximum execution time reached.');
}
function getQueues() {
return \MailPoet\Models\SendingQueue::orderByDesc('priority')
->whereNull('deleted_at')
->whereNull('status')
->findResultSet();
}
function renderNewsletter($newsletter) {
$renderer = new Renderer(json_decode($newsletter['body'], true));
$newsletter['body'] = $renderer->renderAll();
return $newsletter;
}
}