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

- Joins bulk and individual processing into one method

Browse Source 
- Refactors code as per code review comments
This commit is contained in:
Vlad
2016-06-20 23:12:32 -04:00
parent e807aad814
commit f32d6bb331
10 changed files with 155 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
lib/Cron/Daemon.php
View File

@ -1,8 +1,8 @@
<?php
namespace MailPoet\Cron;
use MailPoet\Cron\Workers\Scheduler;
use MailPoet\Cron\Workers\SendingQueue;
use MailPoet\Cron\Workers\Scheduler as SchedulerWorker;
use MailPoet\Cron\Workers\SendingQueue\SendingQueue as SendingQueueWorker;
require_once(ABSPATH . 'wp-includes/pluggable.php');
@ -41,22 +41,22 @@ class Daemon {
}
$this->abortIfStopped($daemon);
try {
$scheduler = new Scheduler();
$scheduler->process($this->timer);
$queue = new SendingQueue();
$queue->process($this->timer);
$scheduler = new SchedulerWorker($this->timer);
$scheduler->process();
$queue = new SendingQueueWorker($this->timer);
$queue->process();
} catch(\Exception $e) {
// continue processing, no need to catch errors
// continue processing, no need to handle errors
}
$elapsed_time = microtime(true) - $this->timer;
if($elapsed_time < CronHelper::DAEMON_EXECUTION_LIMIT) {
sleep(CronHelper::DAEMON_EXECUTION_LIMIT - $elapsed_time);
}
// after each execution, re-read daemon data in case it was deleted or
// after each execution, re-read daemon data in case its status was changed
// its status has changed
$daemon = CronHelper::getDaemon();
if(!$daemon || $daemon['token'] !== $this->data['token']) {
exit;
self::terminate();
}
$daemon['counter']++;
$this->abortIfStopped($daemon);
@ -69,11 +69,13 @@ class Daemon {
}
function abortIfStopped($daemon) {
if($daemon['status'] === self::STATUS_STOPPED) exit;
if($daemon['status'] === self::STATUS_STOPPED) {
self::terminate();
}
if($daemon['status'] === self::STATUS_STOPPING) {
$daemon['status'] = self::STATUS_STOPPED;
CronHelper::saveDaemon($daemon);
exit;
self::terminate();
}
}
@ -83,6 +85,10 @@ class Daemon {
function callSelf() {
CronHelper::accessDaemon($this->token, self::REQUEST_TIMEOUT);
self::terminate();
}
function terminate() {
exit;
}
}

1
lib/Cron/Supervisor.php
View File

@ -26,7 +26,6 @@ class Supervisor {
$daemon['status'] === Daemon::STATUS_STOPPED
) {
return $this->formatDaemonStatusMessage($daemon['status']);
}
$elapsed_time = time() - (int)$daemon['updated_at'];
// if it's been less than 40 seconds since last execution and we're not

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

@ -4,51 +4,48 @@ namespace MailPoet\Cron\Workers\SendingQueue;
use MailPoet\Cron\CronHelper;
use MailPoet\Cron\Workers\SendingQueue\Tasks\Mailer as MailerTask;
use MailPoet\Cron\Workers\SendingQueue\Tasks\Newsletter as NewsletterTask;
use MailPoet\Cron\Workers\SendingQueue\Tasks\Statistics as StatisticsTask;
use MailPoet\Cron\Workers\SendingQueue\Tasks\Subscribers as SubscribersTask;
use MailPoet\Models\Newsletter as NewsletterModel;
use MailPoet\Models\SendingQueue as SendingQueueModel;
use MailPoet\Models\Subscriber;
use MailPoet\Models\StatisticsNewsletters as StatisticsNewslettersModel;
use MailPoet\Models\Subscriber as SubscriberModel;
use MailPoet\Util\Helpers;
if(!defined('ABSPATH')) exit;
class SendingQueue {
public $mta_config;
public $mailer_task;
public $newsletter_task;
private $timer;
const BATCH_SIZE = 50;
function __construct($timer = false) {
$this->mta_config = MailerTask::getMailerConfig();
$this->mailer_task = new MailerTask();
$this->newsletter_task = new NewsletterTask();
$this->timer = ($timer) ? $timer : microtime(true);
CronHelper::checkExecutionTimer($this->timer);
}
function process() {
return;
$mta_log = MailerTask::getMailerLog();
MailerTask::checkSendingLimit($this->mta_config, $mta_log);
$this->mailer_task->checkSendingLimit();
foreach($this->getQueues() as $queue) {
// get and pre-process newsletter (render, replace shortcodes/links, etc.)
$newsletter = NewsletterTask::getAndPreProcess($queue->asArray());
$newsletter = $this->newsletter_task->getAndPreProcess($queue->asArray());
if(!$newsletter) {
$queue->delete();
continue;
}
if(is_null($queue->newsletter_rendered_body)) {
$queue->newsletter_rendered_body = json_encode($newsletter['rendered_body']);
//$queue->save();
$queue->save();
}
// get subscribers
$queue->subscribers = SubscribersTask::get($queue->asArray());
$queue->subscribers = SubscribersTask::get($queue->subscribers);
// configure mailer with newsletter data (from/reply-to)
$mailer = MailerTask::configureMailer($newsletter);
// determine if processing is done in bulk or individually
$processing_method = MailerTask::getProcessingMethod($this->mta_config);
$mailer = $this->mailer_task->configureMailer($newsletter);
foreach(array_chunk($queue->subscribers['to_process'], self::BATCH_SIZE)
as $subscribers_to_process_ids
) {
$found_subscribers = Subscriber::whereIn('id', $subscribers_to_process_ids)
$found_subscribers = SubscriberModel::whereIn('id', $subscribers_to_process_ids)
->findArray();
$found_subscribers_ids = Helpers::arrayColumn($found_subscribers, 'id');
// if some subscribers weren't found, remove them from the processing list
@ -63,30 +60,26 @@ class SendingQueue {
$this->updateQueue($queue);
continue;
}
$queue->subscribers = call_user_func_array(
array(
$this,
$processing_method
),
array(
$queue = $this->processQueue(
$queue,
$mailer,
$mta_log,
$newsletter,
$found_subscribers,
$queue
)
$found_subscribers
);
}
}
}
// TODO: merge processBulkSubscribers with processIndividualSubscriber
function processBulkSubscribers($mailer, $mta_log, $newsletter, $subscribers, $queue) {
$subscribers_ids = Helpers::arrayColumn($subscribers, 'id');
function processQueue($queue, $mailer, $newsletter, $subscribers) {
// determine if processing is done in bulk or individually
$processing_method = $this->mailer_task->getProcessingMethod();
$prepared_newsletters = array();
$prepared_subscribers = array();
$statistics = array();
foreach($subscribers as $subscriber) {
// render shortcodes and replace subscriber data in tracked links
$prepared_newsletters[] =
NewsletterTask::prepareNewsletterForSending(
$this->newsletter_task->prepareNewsletterForSending(
$newsletter,
$subscriber,
$queue->asArray()
@ -95,58 +88,53 @@ class SendingQueue {
$queue->newsletter_rendered_subject = $prepared_newsletters[0]['subject'];
}
// format subscriber name/address according to mailer settings
$prepared_subscribers[] = MailerTask::prepareSubscriberForSending(
$prepared_subscribers[] = $this->mailer_task->prepareSubscriberForSending(
$mailer,
$subscriber
);
$prepared_subscribers_ids[] = $subscriber['id'];
// keep track of values for statistics purposes
$statistics[] = array(
'newsletter_id' => $newsletter['id'],
'subscriber_id' => $subscriber['id'],
'queue_id' => $queue->id
);
if($processing_method === 'individual') {
$queue = $this->sendNewsletters(
$queue,
$mailer,
$prepared_subscribers_ids,
$prepared_newsletters[0],
$prepared_subscribers[0],
$statistics
);
$prepared_newsletters = array();
$prepared_subscribers = array();
$statistics = array();
}
// send
$send_result = MailerTask::send($mailer, $prepared_newsletters, $prepared_subscribers);
if(!$send_result) {
// update failed/to process list
$queue->subscribers = SubscribersTask::updateFailedList(
$subscribers_ids,
$queue->subscribers
);
} else {
// update processed/to process list
$queue->subscribers = SubscribersTask::updateProcessedList(
$subscribers_ids,
$queue->subscribers
);
// log statistics
StatisticsTask::processAndLogBulkNewsletterStatistics(
$subscribers_ids,
$newsletter['id'],
$queue->id
);
// keep track of sent items
$mta_log = MailerTask::updateMailerLog($mta_log);
}
$this->updateQueue($queue);
MailerTask::checkSendingLimit($this->mta_config, $mta_log);
CronHelper::checkExecutionTimer($this->timer);
if($processing_method === 'bulk') {
$queue = $this->sendNewsletters(
$queue,
$mailer,
$prepared_subscribers_ids,
$prepared_newsletters,
$prepared_subscribers,
$statistics
);
}
return $queue;
}
function processIndividualSubscriber($mailer, $mta_log, $newsletter, $subscribers, $queue) {
$subscribers_ids = Helpers::arrayColumn($subscribers, 'id');
foreach($subscribers as $subscriber) {
// render shortcodes and replace subscriber data in tracked links
$prepared_newsletter =
NewsletterTask::prepareNewsletterForSending(
$newsletter,
$subscriber,
$queue->asArray()
);
if(!$queue->newsletter_rendered_subject) {
$queue->newsletter_rendered_subject = $prepared_newsletter['subject'];
}
// format subscriber name/address according to mailer settings
$prepared_subscriber = MailerTask::prepareSubscriberForSending(
function sendNewsletters(
$queue, $mailer, $subscribers_ids, $newsletters, $subscribers, $statistics
) {
// send newsletter
$send_result = $this->mailer_task->send(
$mailer,
$subscriber
$newsletters,
$subscribers
);
$send_result = MailerTask::send($mailer, $prepared_newsletter, $prepared_subscriber);
if(!$send_result) {
// update failed/to process list
$queue->subscribers = SubscribersTask::updateFailedList(
@ -160,20 +148,17 @@ class SendingQueue {
$queue->subscribers
);
// log statistics
StatisticsTask::logStatistics(
array(
$newsletter['id'],
$subscriber['id'],
$queue->id
)
);
StatisticsNewslettersModel::createMultiple($statistics);
// keep track of sent items
$mta_log = MailerTask::updateMailerLog($mta_log);
$this->mailer_task->updateMailerLog();
$subscribers_to_process_count = count($queue->subscribers['to_process']);
}
$queue = $this->updateQueue($queue);
MailerTask::checkSendingLimit($this->mta_config, $mta_log);
CronHelper::checkExecutionTimer($this->timer);
if ($subscribers_to_process_count) {
$this->mailer_task->checkSendingLimit();
}
CronHelper::checkExecutionTimer($this->timer);
return $queue;
}
function getQueues() {

53
lib/Cron/Workers/SendingQueue/Tasks/Mailer.php
View File

@ -7,7 +7,15 @@ use MailPoet\Models\Setting;
if(!defined('ABSPATH')) exit;
class Mailer {
static function configureMailer(array $newsletter) {
public $mta_config;
public $mta_log;
function __construct() {
$this->mta_config = $this->getMailerConfig();
$this->mta_log = $this->getMailerLog();
}
function configureMailer(array $newsletter) {
$sender['address'] = (!empty($newsletter['sender_address'])) ?
$newsletter['sender_address'] :
false;
@ -30,7 +38,7 @@ class Mailer {
return $mailer;
}
static function getMailerConfig() {
function getMailerConfig() {
$mta_config = Setting::getValue('mta');
if(!$mta_config) {
throw new \Exception(__('Mailer is not configured.'));
@ -38,13 +46,7 @@ class Mailer {
return $mta_config;
}
static function updateMailerLog($mta_log) {
$mta_log['sent']++;
Setting::setValue('mta_log', $mta_log);
return $mta_log;
}
static function getMailerLog() {
function getMailerLog() {
$mta_log = Setting::getValue('mta_log');
if(!$mta_log) {
$mta_log = array(
@ -56,40 +58,45 @@ class Mailer {
return $mta_log;
}
static function getProcessingMethod($mta_config) {
return ($mta_config['method'] === 'MailPoet') ?
'processBulkSubscribers' :
'processIndividualSubscriber';
function updateMailerLog() {
$this->mta_log['sent']++;
Setting::setValue('mta_log', $this->mta_log);
}
static function prepareSubscriberForSending($mailer, $subscriber) {
function getProcessingMethod() {
return ($this->mta_config['method'] === 'MailPoet') ?
'bulk' :
'individual';
}
function prepareSubscriberForSending($mailer, $subscriber) {
return ($mailer instanceof \MailPoet\Mailer\Mailer) ?
$mailer->transformSubscriber($subscriber) :
false;
}
static function send($mailer, $newsletter, $subscriber) {
function send($mailer, $newsletter, $subscriber) {
return ($mailer instanceof \MailPoet\Mailer\Mailer) ?
$mailer->mailer_instance->send($newsletter, $subscriber) :
false;
}
static function checkSendingLimit($mta_config, $mta_log) {
$frequency_interval = (int) $mta_config['frequency']['interval'] * 60;
$frequency_limit = (int) $mta_config['frequency']['emails'];
$elapsed_time = time() - (int) $mta_log['started'];
if($mta_log['sent'] === $frequency_limit &&
function checkSendingLimit() {
if($this->mta_config['method'] === 'MailPoet') return;
$frequency_interval = (int) $this->mta_config['frequency']['interval'] * 60;
$frequency_limit = (int) $this->mta_config['frequency']['emails'];
$elapsed_time = time() - (int) $this->mta_log['started'];
if($this->mta_log['sent'] === $frequency_limit &&
$elapsed_time <= $frequency_interval
) {
throw new \Exception(__('Sending frequency limit has been reached.'));
}
if($elapsed_time > $frequency_interval) {
$mta_log = array(
$this->mta_log = array(
'sent' => 0,
'started' => time()
);
Setting::setValue('mta_log', $mta_log);
Setting::setValue('mta_log', $this->mta_log);
}
return;
}
}

32
lib/Cron/Workers/SendingQueue/Tasks/Newsletter.php
View File

@ -14,13 +14,21 @@ use MailPoet\Util\Helpers;
if(!defined('ABSPATH')) exit;
class Newsletter {
static function get($newsletter_id) {
public $tracking_enabled;
public $tracking_image_inserted;
function __construct() {
$this->tracking_enabled = (boolean) Setting::getValue('tracking.enabled');
$this->tracking_image_inserted = false;
}
function get($newsletter_id) {
$newsletter = NewsletterModel::findOne($newsletter_id);
return ($newsletter) ? $newsletter->asArray() : false;
}
static function getAndPreProcess(array $queue) {
$newsletter = self::get($queue['newsletter_id']);
function getAndPreProcess(array $queue) {
$newsletter = $this->get($queue['newsletter_id']);
if(!$newsletter) {
return false;
}
@ -31,16 +39,18 @@ class Newsletter {
return $newsletter;
}
// if tracking is enabled, do additional processing
if((boolean) Setting::getValue('tracking.enabled')) {
// add tracking image
OpenTracking::addTrackingImage();
if($this->tracking_enabled) {
// hook once to the newsletter post-processing filter and add tracking image
if(!$this->tracking_image_inserted) {
$this->tracking_image_inserted = OpenTracking::addTrackingImage();
}
// render newsletter
$newsletter = self::render($newsletter);
$newsletter = $this->render($newsletter);
// hash and save all links
$newsletter = LinksTask::process($newsletter, $queue);
} else {
// render newsletter
$newsletter = self::render($newsletter);
$newsletter = $this->render($newsletter);
}
// check if this is a post notification and if it contains posts
$newsletter_contains_posts = strpos($newsletter['rendered_body']['html'], 'data-post-id');
@ -52,13 +62,13 @@ class Newsletter {
return $newsletter;
}
static function render($newsletter) {
function render($newsletter) {
$renderer = new Renderer($newsletter);
$newsletter['rendered_body'] = $renderer->render();
return $newsletter;
}
static function prepareNewsletterForSending(
function prepareNewsletterForSending(
array $newsletter, array $subscriber, array $queue
) {
// shortcodes and links will be replaced in the subject, html and text body
@ -76,7 +86,7 @@ class Newsletter {
$subscriber,
$queue
);
if((boolean) Setting::getValue('tracking.enabled')) {
if($this->tracking_enabled) {
$prepared_newsletter = NewsletterLinks::replaceSubscriberData(
$newsletter['id'],
$subscriber['id'],

28
lib/Cron/Workers/SendingQueue/Tasks/Statistics.php
View File

@ -1,28 +0,0 @@
<?php
namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
use MailPoet\Models\StatisticsNewsletters;
use MailPoet\Util\Helpers;
if(!defined('ABSPATH')) exit;
class Statistics {
static function processAndLogBulkNewsletterStatistics(
array $processed_subscribers_ids, $newsletter_id, $queue_id
) {
$newsletter_statistics = array();
foreach($processed_subscribers_ids as $subscriber_id) {
$newsletter_statistics[] = array(
$newsletter_id,
$subscriber_id,
$queue_id
);
}
$newsletter_statistics = Helpers::flattenArray($newsletter_statistics);
return self::logStatistics($newsletter_statistics);
}
static function logStatistics($newsletter_statistics) {
return StatisticsNewsletters::createMultiple($newsletter_statistics);
}
}

12
lib/Cron/Workers/SendingQueue/Tasks/Subscribers.php
View File

@ -4,8 +4,8 @@ namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
if(!defined('ABSPATH')) exit;
class Subscribers {
static function get(array $queue) {
$subscribers = unserialize($queue['subscribers']);
static function get($subscribers) {
$subscribers = unserialize($subscribers);
if(empty($subscribers['processed'])) {
$subscribers['processed'] = array();
}
@ -41,8 +41,8 @@ class Subscribers {
$failed_subscribers
);
$queue_subscribers['to_process'] = array_diff(
$failed_subscribers,
$queue_subscribers['to_process']
$queue_subscribers['to_process'],
$failed_subscribers
);
return $queue_subscribers;
}
@ -55,8 +55,8 @@ class Subscribers {
$processed_subscribers
);
$queue_subscribers['to_process'] = array_diff(
$processed_subscribers,
$queue_subscribers['to_process']
$queue_subscribers['to_process'],
$processed_subscribers
);
return $queue_subscribers;
}

2
lib/Newsletter/Links/Links.php
View File

@ -49,7 +49,7 @@ class Links {
);
}
}
return Helpers::arrayUnique($extracted_links);
return array_unique($extracted_links, SORT_REGULAR);
}
static function process($content) {

9
lib/Newsletter/Renderer/PostProcess/OpenTracking.php
View File

@ -2,10 +2,9 @@
namespace MailPoet\Newsletter\Renderer\PostProcess;
use MailPoet\Newsletter\Links\Links;
use MailPoet\Newsletter\Renderer\Renderer;
class OpenTracking {
const OPEN_TRACKING_FILTER_NAME = 'mailpoet_rendering_post_process';
static function process($template) {
$DOM = new \pQuery();
$DOM = $DOM->parseStr($template);
@ -20,11 +19,9 @@ class OpenTracking {
}
static function addTrackingImage() {
if(array_key_exists(self::OPEN_TRACKING_FILTER_NAME, $GLOBALS['wp_filter'])) {
return;
};
add_filter('mailpoet_rendering_post_process', function ($template) {
add_filter(Renderer::POST_PROCESS_FILTER, function ($template) {
return OpenTracking::process($template);
});
return true;
}
}

3
lib/Newsletter/Renderer/Renderer.php
View File

@ -10,6 +10,7 @@ class Renderer {
public $CSS_inliner;
public $newsletter;
const NEWSLETTER_TEMPLATE = 'Template.html';
const POST_PROCESS_FILTER = 'mailpoet_rendering_post_process';
function __construct(array $newsletter) {
$this->newsletter = $newsletter;
@ -102,7 +103,7 @@ class Renderer {
str_replace('&', '&amp;', $template->html())
);
$template = apply_filters(
'mailpoet_rendering_post_process',
self::POST_PROCESS_FILTER,
$DOM->__toString()
);
return $template;