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

Merge pull request #527 from mailpoet/sending_queue_refactor

Browse Source 
Sending queue refactor
This commit is contained in:
Tautvidas Sipavičius
2016-06-23 18:16:21 +03:00
committed by GitHub
parent 9652f75028 d414313749
commit 7ee83dad06
22 changed files with 639 additions and 448 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

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

@@ -1,407 +0,0 @@
<?php
namespace MailPoet\Cron\Workers;
use MailPoet\Cron\CronHelper;
use MailPoet\Mailer\Mailer;
use MailPoet\Models\Newsletter;
use MailPoet\Models\SendingQueue as SendingQueueModel;
use MailPoet\Models\NewsletterPost;
use MailPoet\Models\Setting;
use MailPoet\Models\StatisticsNewsletters;
use MailPoet\Models\Subscriber;
use MailPoet\Newsletter\Links\Links;
use MailPoet\Newsletter\Renderer\PostProcess\OpenTracking;
use MailPoet\Newsletter\Renderer\Renderer;
use MailPoet\Newsletter\Shortcodes\Shortcodes;
use MailPoet\Util\Helpers;
if(!defined('ABSPATH')) exit;
class SendingQueue {
public $mta_config;
public $mta_log;
public $processing_method;
private $timer;
const BATCH_SIZE = 50;
const DIVIDER = '***MailPoet***';
function __construct($timer = false) {
$this->mta_config = $this->getMailerConfig();
$this->mta_log = $this->getMailerLog();
$this->processing_method = ($this->mta_config['method'] === 'MailPoet') ?
'processBulkSubscribers' :
'processIndividualSubscriber';
$this->timer = ($timer) ? $timer : microtime(true);
CronHelper::checkExecutionTimer($this->timer);
}
function process() {
foreach($this->getQueues() as $queue) {
$newsletter = Newsletter::findOne($queue->newsletter_id);
if(!$newsletter) {
$queue->delete();
continue;
}
$newsletter = $newsletter->asArray();
$newsletter['body'] = $this->getOrRenderNewsletterBody($queue, $newsletter);
if($newsletter['type'] === 'notification' &&
strpos($newsletter['body']['html'], 'data-post-id') === false
){
$queue->delete();
continue;
}
$queue->subscribers = (object) unserialize($queue->subscribers);
if(!isset($queue->subscribers->processed)) {
$queue->subscribers->processed = array();
}
if(!isset($queue->subscribers->failed)) {
$queue->subscribers->failed = array();
}
$mailer = $this->configureMailer($newsletter);
foreach(array_chunk($queue->subscribers->to_process, self::BATCH_SIZE) as
$subscribers_ids) {
$subscribers = Subscriber::whereIn('id', $subscribers_ids)
->findArray();
if(count($subscribers_ids) !== count($subscribers)) {
$queue->subscribers->to_process = $this->recalculateSubscriberCount(
Helpers::arrayColumn($subscribers, 'id'),
$subscribers_ids,
$queue->subscribers->to_process
);
}
if(!count($queue->subscribers->to_process)) {
$this->updateQueue($queue);
continue;
}
$queue->subscribers = call_user_func_array(
array(
$this,
$this->processing_method
),
array(
$mailer,
$newsletter,
$subscribers,
$queue
)
);
}
}
}
function getOrRenderNewsletterBody($queue, $newsletter) {
// check if newsletter has been rendered, in which case return its contents
// or render and save for future reuse
if($queue->newsletter_rendered_body === null) {
if((boolean) Setting::getValue('tracking.enabled')) {
// insert tracking code
add_filter('mailpoet_rendering_post_process', function($template) {
return OpenTracking::process($template);
});
// render newsletter
$rendered_newsletter = $this->renderNewsletter($newsletter);
// process link shortcodes, extract and save links in the database
$processed_newsletter = $this->processLinksAndShortcodes(
$this->joinObject($rendered_newsletter),
$newsletter['id'],
$queue->id
);
list($newsletter['body']['html'], $newsletter['body']['text']) =
$this->splitObject($processed_newsletter);
}
else {
// render newsletter
$newsletter['body'] = $this->renderNewsletter($newsletter);
}
$this->extractAndSaveNewsletterPosts(
$newsletter['id'],
$newsletter['body']['html']
);
$queue->newsletter_rendered_body = json_encode($newsletter['body']);
$queue->save();
} else {
$newsletter['body'] = json_decode($queue->newsletter_rendered_body);
}
return (array) $newsletter['body'];
}
function processBulkSubscribers($mailer, $newsletter, $subscribers, $queue) {
foreach($subscribers as $subscriber) {
$processed_newsletters[] =
$this->processNewsletterBeforeSending($newsletter, $subscriber, $queue);
if(!$queue->newsletter_rendered_subject) {
$queue->newsletter_rendered_subject = $processed_newsletters[0]['subject'];
}
$transformed_subscribers[] =
$mailer->transformSubscriber($subscriber);
}
$result = $this->sendNewsletter(
$mailer,
$processed_newsletters,
$transformed_subscribers
);
$subscribers_ids = Helpers::arrayColumn($subscribers, 'id');
if(!$result) {
$queue->subscribers->failed = array_merge(
$queue->subscribers->failed,
$subscribers_ids
);
} else {
$newsletter_statistics =
array_map(function($data) use ($newsletter, $subscribers_ids, $queue) {
return array(
$newsletter['id'],
$subscribers_ids[$data],
$queue->id
);
}, range(0, count($transformed_subscribers) - 1));
$newsletter_statistics = Helpers::flattenArray($newsletter_statistics);
$this->updateMailerLog();
$this->updateNewsletterStatistics($newsletter_statistics);
$queue->subscribers->processed = array_merge(
$queue->subscribers->processed,
$subscribers_ids
);
}
$this->updateQueue($queue);
$this->checkSendingLimit();
CronHelper::checkExecutionTimer($this->timer);
return $queue->subscribers;
}
function processIndividualSubscriber($mailer, $newsletter, $subscribers, $queue) {
foreach($subscribers as $subscriber) {
$this->checkSendingLimit();
$processed_newsletter = $this->processNewsletterBeforeSending($newsletter, $subscriber, $queue);
if(!$queue->newsletter_rendered_subject) {
$queue->newsletter_rendered_subject = $processed_newsletter['subject'];
}
$transformed_subscriber = $mailer->transformSubscriber($subscriber);
$result = $this->sendNewsletter(
$mailer,
$processed_newsletter,
$transformed_subscriber
);
if(!$result) {
$queue->subscribers->failed[] = $subscriber['id'];
} else {
$queue->subscribers->processed[] = $subscriber['id'];
$newsletter_statistics = array(
$newsletter['id'],
$subscriber['id'],
$queue->id
);
$this->updateMailerLog();
$this->updateNewsletterStatistics($newsletter_statistics);
}
$this->updateQueue($queue);
CronHelper::checkExecutionTimer($this->timer);
}
return $queue->subscribers;
}
function updateNewsletterStatistics($data) {
return StatisticsNewsletters::createMultiple($data);
}
function renderNewsletter($newsletter) {
$renderer = new Renderer($newsletter);
return $renderer->render();
}
function processLinksAndShortcodes($content, $newsletter_id, $queue_id) {
// process only link shortcodes
$shortcodes = new Shortcodes($newsletter = false, $subscriber = false, $queue_id);
$content = $shortcodes->replace(
$content,
$categories = array('link')
);
// extract and save links and link shortcodes
list($content, $processed_links) =
Links::process(
$content,
$links = false,
$process_link_shortcodes = true,
$queue_id
);
Links::save($processed_links, $newsletter_id, $queue_id);
return $content;
}
function processNewsletterBeforeSending($newsletter, $subscriber = false, $queue) {
$data_for_shortcodes = array(
$newsletter['subject'],
$newsletter['body']['html'],
$newsletter['body']['text']
);
$processed_newsletter = $this->replaceShortcodes(
$newsletter,
$subscriber,
$queue,
$this->joinObject($data_for_shortcodes)
);
if((boolean) Setting::getValue('tracking.enabled')) {
$processed_newsletter = Links::replaceSubscriberData(
$newsletter['id'],
$subscriber['id'],
$queue->id,
$processed_newsletter
);
}
list($newsletter['subject'],
$newsletter['body']['html'],
$newsletter['body']['text']
) = $this->splitObject($processed_newsletter);
return $newsletter;
}
function replaceShortcodes($newsletter, $subscriber, $queue, $body) {
$shortcodes = new Shortcodes(
$newsletter,
$subscriber,
$queue
);
return $shortcodes->replace($body);
}
function sendNewsletter($mailer, $newsletter, $subscriber) {
return $mailer->mailer_instance->send(
$newsletter,
$subscriber
);
}
function configureMailer($newsletter) {
$sender['address'] = (!empty($newsletter['sender_address'])) ?
$newsletter['sender_address'] :
false;
$sender['name'] = (!empty($newsletter['sender_name'])) ?
$newsletter['sender_name'] :
false;
$reply_to['address'] = (!empty($newsletter['reply_to_address'])) ?
$newsletter['reply_to_address'] :
false;
$reply_to['name'] = (!empty($newsletter['reply_to_name'])) ?
$newsletter['reply_to_name'] :
false;
if(!$sender['address']) {
$sender = false;
}
if(!$reply_to['address']) {
$reply_to = false;
}
$mailer = new Mailer($method = false, $sender, $reply_to);
return $mailer;
}
function getQueues() {
return SendingQueueModel::orderByDesc('priority')
->whereNull('deleted_at')
->whereNull('status')
->findResultSet();
}
function updateQueue($queue) {
$queue = clone($queue);
$queue->subscribers->to_process = array_diff(
$queue->subscribers->to_process,
array_merge(
$queue->subscribers->processed,
$queue->subscribers->failed
)
);
$queue->subscribers->to_process = array_values(
$queue->subscribers->to_process
);
$queue->count_processed =
count($queue->subscribers->processed) + count($queue->subscribers->failed);
$queue->count_to_process = count($queue->subscribers->to_process);
$queue->count_failed = count($queue->subscribers->failed);
$queue->count_total =
$queue->count_processed + $queue->count_to_process;
if(!$queue->count_to_process) {
$queue->processed_at = current_time('mysql');
$queue->status = SendingQueueModel::STATUS_COMPLETED;
// set newsletter status to sent
$newsletter = Newsletter::findOne($queue->newsletter_id);
// if it's a standard newsletter, update its status
if($newsletter->type === Newsletter::TYPE_STANDARD) {
$newsletter->setStatus(Newsletter::STATUS_SENT);
} else if($newsletter->type === Newsletter::TYPE_NOTIFICATION) {
// TODO: Check with Vlad
}
}
$queue->subscribers = serialize((array) $queue->subscribers);
$queue->save();
}
function updateMailerLog() {
$this->mta_log['sent']++;
return Setting::setValue('mta_log', $this->mta_log);
}
function getMailerConfig() {
$mta_config = Setting::getValue('mta');
if(!$mta_config) {
throw new \Exception(__('Mailer is not configured.'));
}
return $mta_config;
}
function getMailerLog() {
$mta_log = Setting::getValue('mta_log');
if(!$mta_log) {
$mta_log = array(
'sent' => 0,
'started' => time()
);
Setting::setValue('mta_log', $mta_log);
}
return $mta_log;
}
function checkSendingLimit() {
$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) {
$this->mta_log = array(
'sent' => 0,
'started' => time()
);
Setting::setValue('mta_log', $this->mta_log);
}
return;
}
function recalculateSubscriberCount(
$found_subscriber, $existing_subscribers, $subscribers_to_process) {
$subscibers_to_exclude = array_diff($existing_subscribers, $found_subscriber);
return array_diff($subscribers_to_process, $subscibers_to_exclude);
}
function extractAndSaveNewsletterPosts($newletter_id, $content) {
preg_match_all('/data-post-id="(\d+)"/ism', $content, $posts);
$posts = $posts[1];
foreach($posts as $post) {
$newletter_post = NewsletterPost::create();
$newletter_post->newsletter_id = $newletter_id;
$newletter_post->post_id = $post;
$newletter_post->save();
}
}
private function joinObject($object = array()) {
return implode(self::DIVIDER, $object);
}
private function splitObject($object = array()) {
return explode(self::DIVIDER, $object);
}
}

188
lib/Cron/Workers/SendingQueue/SendingQueue.php Normal file
View File

@@ -0,0 +1,188 @@
<?php
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\Subscribers as SubscribersTask;
use MailPoet\Models\Newsletter as NewsletterModel;
use MailPoet\Models\SendingQueue as SendingQueueModel;
use MailPoet\Models\StatisticsNewsletters as StatisticsNewslettersModel;
use MailPoet\Models\Subscriber as SubscriberModel;
use MailPoet\Util\Helpers;
if(!defined('ABSPATH')) exit;
class SendingQueue {
public $mailer_task;
public $newsletter_task;
private $timer;
const BATCH_SIZE = 50;
function __construct($timer = false) {
$this->mailer_task = new MailerTask();
$this->newsletter_task = new NewsletterTask();
$this->timer = ($timer) ? $timer : microtime(true);
}
function process() {
$this->mailer_task->checkSendingLimit();
foreach($this->getQueues() as $queue) {
// get and pre-process newsletter (render, replace shortcodes/links, etc.)
$newsletter = $this->newsletter_task->getAndPreProcess($queue->asArray());
if(!$newsletter) {
$queue->delete();
continue;
}
// configure mailer
$this->mailer_task->configureMailer($newsletter);
if(is_null($queue->newsletter_rendered_body)) {
$queue->newsletter_rendered_body = json_encode($newsletter['rendered_body']);
$queue->save();
}
// get subscribers
$queue->subscribers = SubscribersTask::get($queue->subscribers);
foreach(array_chunk($queue->subscribers['to_process'], self::BATCH_SIZE)
as $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
if(count($found_subscribers_ids) !== count($subscribers_to_process_ids)) {
$queue->subscribers = SubscribersTask::updateToProcessList(
$found_subscribers_ids,
$subscribers_to_process_ids,
$queue->subscribers
);
}
if(!count($queue->subscribers['to_process'])) {
$this->updateQueue($queue);
continue;
}
$queue = $this->processQueue(
$queue,
$newsletter,
$found_subscribers
);
}
}
}
function processQueue($queue, $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[] =
$this->newsletter_task->prepareNewsletterForSending(
$newsletter,
$subscriber,
$queue->asArray()
);
if(!$queue->newsletter_rendered_subject) {
$queue->newsletter_rendered_subject = $prepared_newsletters[0]['subject'];
}
// format subscriber name/address according to mailer settings
$prepared_subscribers[] = $this->mailer_task->prepareSubscriberForSending(
$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,
$prepared_subscribers_ids,
$prepared_newsletters[0],
$prepared_subscribers[0],
$statistics
);
$prepared_newsletters = array();
$prepared_subscribers = array();
$statistics = array();
}
}
if($processing_method === 'bulk') {
$queue = $this->sendNewsletters(
$queue,
$prepared_subscribers_ids,
$prepared_newsletters,
$prepared_subscribers,
$statistics
);
}
return $queue;
}
function sendNewsletters(
$queue, $prepared_subscribers_ids, $prepared_newsletters,
$prepared_subscribers, $statistics
) {
// send newsletter
$send_result = $this->mailer_task->send(
$prepared_newsletters,
$prepared_subscribers
);
if(!$send_result) {
// update failed/to process list
$queue->subscribers = SubscribersTask::updateFailedList(
$prepared_subscribers_ids,
$queue->subscribers
);
} else {
// update processed/to process list
$queue->subscribers = SubscribersTask::updateProcessedList(
$prepared_subscribers_ids,
$queue->subscribers
);
// log statistics
StatisticsNewslettersModel::createMultiple($statistics);
// keep track of sent items
$this->mailer_task->updateMailerLog();
$subscribers_to_process_count = count($queue->subscribers['to_process']);
}
$queue = $this->updateQueue($queue);
if($subscribers_to_process_count) {
$this->mailer_task->checkSendingLimit();
}
CronHelper::checkExecutionTimer($this->timer);
return $queue;
}
function getQueues() {
return SendingQueueModel::orderByDesc('priority')
->whereNull('deleted_at')
->whereNull('status')
->findResultSet();
}
function updateQueue($queue) {
$queue->count_processed =
count($queue->subscribers['processed']) + count($queue->subscribers['failed']);
$queue->count_to_process = count($queue->subscribers['to_process']);
$queue->count_failed = count($queue->subscribers['failed']);
$queue->count_total =
$queue->count_processed + $queue->count_to_process;
if(!$queue->count_to_process) {
$queue->processed_at = current_time('mysql');
$queue->status = SendingQueueModel::STATUS_COMPLETED;
// set newsletter status to sent
$newsletter = NewsletterModel::findOne($queue->newsletter_id);
// if it's a standard newsletter, update its status
if($newsletter->type === NewsletterModel::TYPE_STANDARD) {
$newsletter->setStatus(NewsletterModel::STATUS_SENT);
}
}
$queue->subscribers = serialize((array) $queue->subscribers);
$queue->save();
return $queue;
}
}

33
lib/Cron/Workers/SendingQueue/Tasks/Links.php Normal file
View File

@@ -0,0 +1,33 @@
<?php
namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
use MailPoet\Newsletter\Links\Links as NewsletterLinks;
use MailPoet\Util\Helpers;
if(!defined('ABSPATH')) exit;
class Links {
static function process(array $newsletter, array $queue) {
list($newsletter, $links) = self::hashAndReplaceLinks($newsletter, $queue);
self::saveLinks($links, $newsletter, $queue);
return $newsletter;
}
static function hashAndReplaceLinks(array $newsletter, array $queue) {
// join HTML and TEXT rendered body into a text string
$content = Helpers::joinObject($newsletter['rendered_body']);
list($content, $links) = NewsletterLinks::process($content);
// split the processed body with hashed links back to HTML and TEXT
list($newsletter['rendered_body']['html'], $newsletter['rendered_body']['text'])
= Helpers::splitObject($content);
return array(
$newsletter,
$links
);
}
static function saveLinks($links, $newsletter, $queue) {
return NewsletterLinks::save($links, $newsletter['id'], $queue['id']);
}
}

103
lib/Cron/Workers/SendingQueue/Tasks/Mailer.php Normal file
View File

@@ -0,0 +1,103 @@
<?php
namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
use MailPoet\Mailer\Mailer as MailerFactory;
use MailPoet\Models\Setting;
if(!defined('ABSPATH')) exit;
class Mailer {
public $mta_config;
public $mta_log;
public $mailer;
function __construct() {
$this->mta_config = $this->getMailerConfig();
$this->mta_log = $this->getMailerLog();
$this->mailer = $this->configureMailer();
}
function configureMailer(array $newsletter = null) {
$sender['address'] = (!empty($newsletter['sender_address'])) ?
$newsletter['sender_address'] :
false;
$sender['name'] = (!empty($newsletter['sender_name'])) ?
$newsletter['sender_name'] :
false;
$reply_to['address'] = (!empty($newsletter['reply_to_address'])) ?
$newsletter['reply_to_address'] :
false;
$reply_to['name'] = (!empty($newsletter['reply_to_name'])) ?
$newsletter['reply_to_name'] :
false;
if(!$sender['address']) {
$sender = false;
}
if(!$reply_to['address']) {
$reply_to = false;
}
$this->mailer = new MailerFactory($method = false, $sender, $reply_to);
return $this->mailer;
}
function getMailerConfig() {
$mta_config = Setting::getValue('mta');
if(!$mta_config) {
throw new \Exception(__('Mailer is not configured.'));
}
return $mta_config;
}
function getMailerLog() {
$mta_log = Setting::getValue('mta_log');
if(!$mta_log) {
$mta_log = array(
'sent' => 0,
'started' => time()
);
Setting::setValue('mta_log', $mta_log);
}
return $mta_log;
}
function updateMailerLog() {
$this->mta_log['sent']++;
Setting::setValue('mta_log', $this->mta_log);
}
function getProcessingMethod() {
return ($this->mta_config['method'] === 'MailPoet') ?
'bulk' :
'individual';
}
function prepareSubscriberForSending(array $subscriber) {
return $this->mailer->transformSubscriber($subscriber);
}
function send($prepared_newsletters, $prepared_subscribers) {
return $this->mailer->mailer_instance->send(
$prepared_newsletters,
$prepared_subscribers
);
}
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) {
$this->mta_log = array(
'sent' => 0,
'started' => time()
);
Setting::setValue('mta_log', $this->mta_log);
}
}
}

106
lib/Cron/Workers/SendingQueue/Tasks/Newsletter.php Normal file
View File

@@ -0,0 +1,106 @@
<?php
namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
use MailPoet\Cron\Workers\SendingQueue\Tasks\Links as LinksTask;
use MailPoet\Cron\Workers\SendingQueue\Tasks\Posts as PostsTask;
use MailPoet\Cron\Workers\SendingQueue\Tasks\Shortcodes as ShortcodesTask;
use MailPoet\Models\Newsletter as NewsletterModel;
use MailPoet\Models\Setting;
use MailPoet\Newsletter\Links\Links as NewsletterLinks;
use MailPoet\Newsletter\Renderer\PostProcess\OpenTracking;
use MailPoet\Newsletter\Renderer\Renderer;
use MailPoet\Util\Helpers;
if(!defined('ABSPATH')) exit;
class Newsletter {
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;
}
function getAndPreProcess(array $queue) {
$newsletter = $this->get($queue['newsletter_id']);
if(!$newsletter) {
return false;
}
// if the newsletter was previously rendered, return it
// otherwise, process/render it
if(!is_null($queue['newsletter_rendered_body'])) {
$newsletter['rendered_body'] = json_decode($queue['newsletter_rendered_body'], true);
return $newsletter;
}
// if tracking is enabled, do additional processing
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 = $this->render($newsletter);
// hash and save all links
$newsletter = LinksTask::process($newsletter, $queue);
} else {
// 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');
if($newsletter['type'] === 'notification' && !$newsletter_contains_posts) {
return false;
}
// save all posts
$newsletter = PostsTask::extractAndSave($newsletter);
return $newsletter;
}
function render($newsletter) {
$renderer = new Renderer($newsletter);
$newsletter['rendered_body'] = $renderer->render();
return $newsletter;
}
function prepareNewsletterForSending(
array $newsletter, array $subscriber, array $queue
) {
// shortcodes and links will be replaced in the subject, html and text body
// to speed the processing, join content into a continuous string
$prepared_newsletter = Helpers::joinObject(
array(
$newsletter['subject'],
$newsletter['rendered_body']['html'],
$newsletter['rendered_body']['text']
)
);
$prepared_newsletter = ShortcodesTask::process(
$prepared_newsletter,
$newsletter,
$subscriber,
$queue
);
if($this->tracking_enabled) {
$prepared_newsletter = NewsletterLinks::replaceSubscriberData(
$newsletter['id'],
$subscriber['id'],
$queue['id'],
$prepared_newsletter
);
}
$prepared_newsletter = Helpers::splitObject($prepared_newsletter);
return array(
'subject' => $prepared_newsletter[0],
'body' => array(
'html' => $prepared_newsletter[1],
'text' => $prepared_newsletter[2]
)
);
}
}

28
lib/Cron/Workers/SendingQueue/Tasks/Posts.php Normal file
View File

@@ -0,0 +1,28 @@
<?php
namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
use MailPoet\Models\NewsletterPost;
if(!defined('ABSPATH')) exit;
class Posts {
static function extractAndSave(array $newsletter) {
if(empty($newsletter['rendered_body']['html']) || empty($newsletter['id'])) {
return;
}
preg_match_all(
'/data-post-id="(\d+)"/ism',
$newsletter['rendered_body']['html'],
$matched_posts_ids);
$matched_posts_ids = $matched_posts_ids[1];
if(!count($matched_posts_ids)) {
return $newsletter;
}
foreach($matched_posts_ids as $post_id) {
$newletter_post = NewsletterPost::create();
$newletter_post->newsletter_id = $newsletter['id'];
$newletter_post->post_id = $post_id;
$newletter_post->save();
}
}
}

14
lib/Cron/Workers/SendingQueue/Tasks/Shortcodes.php Normal file
View File

@@ -0,0 +1,14 @@
<?php
namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
use MailPoet\Newsletter\Shortcodes\Shortcodes as NewsletterShortcodes;
if(!defined('ABSPATH')) exit;
class Shortcodes {
static function process($content, array $newsletter, array $subscriber, array $queue) {
$shortcodes = new NewsletterShortcodes($newsletter, $subscriber, $queue);
return $shortcodes->replace($content);
}
}

63
lib/Cron/Workers/SendingQueue/Tasks/Subscribers.php Normal file
View File

@@ -0,0 +1,63 @@
<?php
namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
if(!defined('ABSPATH')) exit;
class Subscribers {
static function get($subscribers) {
$subscribers = unserialize($subscribers);
if(empty($subscribers['processed'])) {
$subscribers['processed'] = array();
}
if(empty($subscribers['failed'])) {
$subscribers['failed'] = array();
}
return $subscribers;
}
static function updateToProcessList(
array $found_subscribers_ids,
array $subscribers_to_process_ids,
array $queue_subscribers
) {
// compare existing subscriber to the ones that queued for processing
$subscibers_to_exclude = array_diff(
$subscribers_to_process_ids,
$found_subscribers_ids
);
// remove nonexistent subscribers from the processing list
$queue_subscribers['to_process'] = array_diff(
$subscibers_to_exclude,
$queue_subscribers['to_process']
);
return $queue_subscribers;
}
static function updateFailedList(
array $failed_subscribers, array $queue_subscribers
) {
$queue_subscribers['failed'] = array_merge(
$queue_subscribers['failed'],
$failed_subscribers
);
$queue_subscribers['to_process'] = array_diff(
$queue_subscribers['to_process'],
$failed_subscribers
);
return $queue_subscribers;
}
static function updateProcessedList(
array $processed_subscribers, array $queue_subscribers
) {
$queue_subscribers['processed'] = array_merge(
$queue_subscribers['processed'],
$processed_subscribers
);
$queue_subscribers['to_process'] = array_diff(
$queue_subscribers['to_process'],
$processed_subscribers
);
return $queue_subscribers;
}
}

18
lib/Models/StatisticsNewsletters.php
View File

@@ -10,15 +10,27 @@ class StatisticsNewsletters extends Model {
parent::__construct();
}
static function createMultiple($data) {
static function createMultiple(array $data) {
$values = array();
foreach($data as $value) {
if(!empty($value['newsletter_id']) &&
!empty($value['subscriber_id']) &&
!empty($value['queue_id'])
) {
$values[] = $value['newsletter_id'];
$values[] = $value['subscriber_id'];
$values[] = $value['queue_id'];
}
}
if (!count($values)) return false;
return self::rawExecute(
'INSERT INTO `' . self::$_table . '` ' .
'(newsletter_id, subscriber_id, queue_id) ' .
'VALUES ' . rtrim(
str_repeat('(?,?,?), ', count($data)/3),
str_repeat('(?,?,?), ', count($values) / 3),
', '
),
$data
$values
);
}
}

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

@@ -3,6 +3,7 @@ namespace MailPoet\Newsletter\Links;
use MailPoet\Models\NewsletterLink;
use MailPoet\Newsletter\Shortcodes\Shortcodes;
use MailPoet\Util\Helpers;
use MailPoet\Util\Security;
class Links {
@@ -39,16 +40,16 @@ class Links {
}, $shortcodes);
// extract urls with href="url" format
preg_match_all($regex, $content, $matched_urls);
$matched_urls_count = count($matched_urls[1]);
$matched_urls_count = count($matched_urls[0]);
if($matched_urls_count) {
for($index = 0; $index <= $matched_urls_count; $index++) {
for($index = 0; $index < $matched_urls_count; $index++) {
$extracted_links[] = array(
'html' => $matched_urls[0][$index],
'link' => $matched_urls[2][$index]
);
}
}
return $extracted_links;
return array_unique($extracted_links, SORT_REGULAR);
}
static function process($content) {
@@ -79,9 +80,9 @@ class Links {
// third, replace text version URL with tracked link: [description](url)
// regex is used to avoid replacing description URLs that are wrapped in round brackets
// i.e., <a href="http://google.com">(http://google.com)</a> => [(http://google.com)](http://tracked_link)
$regex_escaped_tracked_link = preg_quote($tracked_link, '/');
$regex_escaped_extracted_link = preg_quote($extracted_link['link'], '/');
$content = preg_replace(
'/(\[' . $regex_escaped_tracked_link . '\])(\(' . $regex_escaped_tracked_link . '\))/',
'/\[(' . $regex_escaped_extracted_link . ')\](\(' . $regex_escaped_extracted_link . '\))/',
'[$1](' . $tracked_link . ')',
$content
);
@@ -117,8 +118,9 @@ class Links {
return $content;
}
static function save($links, $newsletter_id, $queue_id) {
static function save(array $links, $newsletter_id, $queue_id) {
foreach($links as $link) {
if(empty($link['hash'] || empty($link['url']))) continue;
$newsletter_link = NewsletterLink::create();
$newsletter_link->newsletter_id = $newsletter_id;
$newsletter_link->queue_id = $queue_id;

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

@@ -2,18 +2,26 @@
namespace MailPoet\Newsletter\Renderer\PostProcess;
use MailPoet\Newsletter\Links\Links;
use MailPoet\Newsletter\Renderer\Renderer;
class OpenTracking {
static function process($template) {
$DOM = new \pQuery();
$DOM = $DOM->parseStr($template);
$template = $DOM->query('body');
$open_tracking_link = sprintf(
$open_tracking_image = sprintf(
'<img alt="" class="" src="%s/%s"/>',
home_url(),
esc_attr('?mailpoet&endpoint=track&action=open&data=' . Links::DATA_TAG)
);
$template->html($template->html() . $open_tracking_link);
$template->html($template->html() . $open_tracking_image);
return $DOM->__toString();
}
static function addTrackingImage() {
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;

24
lib/Newsletter/Shortcodes/Shortcodes.php
View File

@@ -5,6 +5,7 @@ class Shortcodes {
public $newsletter;
public $subscriber;
public $queue;
const SHORTCODE_CATEGORY_NAMESPACE = 'MailPoet\Newsletter\Shortcodes\Categories\\';
function __construct(
$newsletter = false,
@@ -22,14 +23,17 @@ class Shortcodes {
$queue;
}
function extract($content, $categories= false) {
function extract($content, $categories = false) {
$categories = (is_array($categories)) ? implode('|', $categories) : false;
$regex = sprintf(
'/\[%s:.*?\]/ism',
($categories) ? '(?:' . $categories . ')' : '(?:\w+)'
);
preg_match_all($regex, $content, $shortcodes);
return array_unique($shortcodes[0]);
$shortcodes = $shortcodes[0];
return (count($shortcodes)) ?
array_unique($shortcodes) :
false;
}
function match($shortcode) {
@@ -43,18 +47,19 @@ class Shortcodes {
function process($shortcodes, $content = false) {
$processed_shortcodes = array_map(
function($shortcode) use($content) {
function($shortcode) use ($content) {
$shortcode_details = $this->match($shortcode);
$shortcode_category = isset($shortcode_details['category']) ?
$shortcode_category = !empty($shortcode_details['category']) ?
ucfirst($shortcode_details['category']) :
false;
$shortcode_action = isset($shortcode_details['action']) ?
$shortcode_action = !empty($shortcode_details['action']) ?
$shortcode_details['action'] :
false;
$shortcode_class =
__NAMESPACE__ . '\\Categories\\' . $shortcode_category;
$shortcode_default_value = isset($shortcode_details['default'])
? $shortcode_details['default'] : false;
self::SHORTCODE_CATEGORY_NAMESPACE . $shortcode_category;
$shortcode_default_value = !empty($shortcode_details['default']) ?
$shortcode_details['default'] :
false;
if(!class_exists($shortcode_class)) {
$custom_shortcode = apply_filters(
'mailpoet_newsletter_shortcode',
@@ -82,6 +87,9 @@ class Shortcodes {
function replace($content, $categories = false) {
$shortcodes = $this->extract($content, $categories);
if(!$shortcodes) {
return $content;
}
$processed_shortcodes = $this->process($shortcodes, $content);
$shortcodes = array_intersect_key($shortcodes, $processed_shortcodes);
return str_replace($shortcodes, $processed_shortcodes, $content);

22
lib/Util/Helpers.php
View File

@@ -2,19 +2,21 @@
namespace MailPoet\Util;
class Helpers {
const DIVIDER = '***MailPoet***';
static function getMaxPostSize($bytes = false) {
$maxPostSize = ini_get('post_max_size');
if(!$bytes) return $maxPostSize;
switch (substr($maxPostSize, -1)) {
switch(substr($maxPostSize, -1)) {
case 'M':
case 'm':
return (int)$maxPostSize * 1048576;
return (int) $maxPostSize * 1048576;
case 'K':
case 'k':
return (int)$maxPostSize * 1024;
return (int) $maxPostSize * 1024;
case 'G':
case 'g':
return (int)$maxPostSize * 1073741824;
return (int) $maxPostSize * 1073741824;
default:
return $maxPostSize;
}
@@ -69,13 +71,13 @@ class Helpers {
$paramsIndexKey = null;
if(isset($params[2])) {
if(is_float($params[2]) || is_int($params[2])) {
$paramsIndexKey = (int)$params[2];
$paramsIndexKey = (int) $params[2];
} else {
$paramsIndexKey = (string) $params[2];
}
}
$resultArray = array();
foreach ($paramsInput as $row) {
foreach($paramsInput as $row) {
$key = $value = null;
$keySet = $valueSet = false;
if($paramsIndexKey !== null && array_key_exists($paramsIndexKey, $row)) {
@@ -113,4 +115,12 @@ class Helpers {
$func = create_function('$c', 'return "_" . strtolower($c[1]);');
return preg_replace_callback('/([A-Z])/', $func, $str);
}
static function joinObject($object = array()) {
return implode(self::DIVIDER, $object);
}
static function splitObject($object = array()) {
return explode(self::DIVIDER, $object);
}
}

7
tests/unit/Models/NewsletterTest.php
View File

@@ -113,7 +113,9 @@ class NewsletterTest extends MailPoetTest {
$sending_queue->save();
$subscriber = Subscriber::createOrUpdate(array(
'email' => 'john.doe@mailpoet.com'
'email' => 'john.doe@mailpoet.com',
'first_name' => 'John',
'last_name' => 'Doe'
));
$opens = StatisticsOpens::create();
@@ -123,7 +125,7 @@ class NewsletterTest extends MailPoetTest {
$opens->save();
$newsletter->queue = $newsletter->getQueue()->asArray();
$statistics = $newsletter->getStatistics();
$statistics = $newsletter->getStatistics( $sending_queue->id);
expect($statistics->opened)->equals(1);
expect($statistics->clicked)->equals(0);
expect($statistics->unsubscribed)->equals(0);
@@ -250,5 +252,6 @@ class NewsletterTest extends MailPoetTest {
ORM::raw_execute('TRUNCATE ' . Segment::$_table);
ORM::raw_execute('TRUNCATE ' . NewsletterSegment::$_table);
ORM::raw_execute('TRUNCATE ' . SendingQueue::$_table);
ORM::raw_execute('TRUNCATE ' . StatisticsOpens::$_table);
}
}

4
tests/unit/Models/SubscriberSegmentTest.php
View File

@@ -8,6 +8,8 @@ class SubscriberSegmentTest extends MailPoetTest {
function _before() {
$this->subscriber = Subscriber::createOrUpdate(array(
'email' => 'john.doe@mailpoet.com',
'first_name' => 'John',
'last_name' => 'Doe',
'status' => Subscriber::STATUS_SUBSCRIBED
));
$this->segment_1 = Segment::createOrUpdate(array('name' => 'Segment 1'));
@@ -133,6 +135,8 @@ class SubscriberSegmentTest extends MailPoetTest {
// create a second subscriber
$subscriber_2 = Subscriber::createOrUpdate(array(
'email' => 'jane.doe@mailpoet.com',
'first_name' => 'Jane',
'last_name' => 'Doe',
'status' => Subscriber::STATUS_SUBSCRIBED
));
// subscribe her to segments

4
tests/unit/Models/SubscriberTest.php
View File

@@ -316,6 +316,8 @@ class SubscriberTest extends MailPoetTest {
function testItCannotTrashAWPUser() {
$wp_subscriber = Subscriber::createOrUpdate(array(
'email' => 'some.wp.user@mailpoet.com',
'first_name' => 'Some',
'last_name' => 'WP User',
'wp_user_id' => 1
));
expect($wp_subscriber->trash())->equals(false);
@@ -328,6 +330,8 @@ class SubscriberTest extends MailPoetTest {
function testItCannotDeleteAWPUser() {
$wp_subscriber = Subscriber::createOrUpdate(array(
'email' => 'some.wp.user@mailpoet.com',
'first_name' => 'Some',
'last_name' => 'WP User',
'wp_user_id' => 1
));
expect($wp_subscriber->delete())->equals(false);

2
tests/unit/Statistics/Track/ClicksTest.php
View File

@@ -18,6 +18,8 @@ class ClicksTest extends MailPoetTest {
// create subscriber
$subscriber = Subscriber::create();
$subscriber->email = 'test@example.com';
$subscriber->first_name = 'First';
$subscriber->last_name = 'Last';
$this->subscriber = $subscriber->save();
// create queue
$queue = SendingQueue::create();

2
tests/unit/Statistics/Track/OpensTest.php
View File

@@ -16,6 +16,8 @@ class OpensTest extends MailPoetTest {
// create subscriber
$subscriber = Subscriber::create();
$subscriber->email = 'test@example.com';
$subscriber->first_name = 'First';
$subscriber->last_name = 'Last';
$this->subscriber = $subscriber->save();
// create queue
$queue = SendingQueue::create();

2
tests/unit/Statistics/Track/UnsubscribesTest.php
View File

@@ -18,6 +18,8 @@ class UnsubscribesTest extends MailPoetTest {
// create subscriber
$subscriber = Subscriber::create();
$subscriber->email = 'test@example.com';
$subscriber->first_name = 'First';
$subscriber->last_name = 'Last';
$this->subscriber = $subscriber->save();
// create queue
$queue = SendingQueue::create();