From c83ab0886f5bbf6848f6b67d33eae0771a3d40e4 Mon Sep 17 00:00:00 2001 From: Vlad Date: Fri, 17 Jun 2016 15:00:46 -0400 Subject: [PATCH] - Rebases master --- .../Workers/SendingQueue/SendingQueue.php | 170 ++++++++++-------- lib/Cron/Workers/SendingQueue/Tasks/Posts.php | 11 +- .../Workers/SendingQueue/Tasks/Statistics.php | 12 +- .../SendingQueue/Tasks/Subscribers.php | 40 ++++- .../Renderer/PostProcess/OpenTracking.php | 5 + 5 files changed, 144 insertions(+), 94 deletions(-) diff --git a/lib/Cron/Workers/SendingQueue/SendingQueue.php b/lib/Cron/Workers/SendingQueue/SendingQueue.php index c4a63f08b4..f794838c70 100644 --- a/lib/Cron/Workers/SendingQueue/SendingQueue.php +++ b/lib/Cron/Workers/SendingQueue/SendingQueue.php @@ -6,8 +6,8 @@ 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\Setting; use MailPoet\Models\Subscriber; use MailPoet\Util\Helpers; @@ -15,20 +15,21 @@ if(!defined('ABSPATH')) exit; class SendingQueue { public $mta_config; - public $mta_log; private $timer; const BATCH_SIZE = 50; - const STATUS_COMPLETED = 'completed'; function __construct($timer = false) { $this->mta_config = MailerTask::getMailerConfig(); - $this->mta_log = MailerTask::getMailerLog(); $this->timer = ($timer) ? $timer : microtime(true); CronHelper::checkExecutionTimer($this->timer); } function process() { + return; + $mta_log = MailerTask::getMailerLog(); + MailerTask::checkSendingLimit($this->mta_config, $mta_log); foreach($this->getQueues() as $queue) { + // get and pre-process newsletter (render, replace shortcodes/links, etc.) $newsletter = NewsletterTask::getAndPreProcess($queue->asArray()); if(!$newsletter) { $queue->delete(); @@ -36,22 +37,26 @@ class SendingQueue { } 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()); // 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); - foreach(array_chunk($queue->subscribers['to_process'], self::BATCH_SIZE) as - $subscribers_to_process_ids) { - $subscribers = Subscriber::whereIn('id', $subscribers_to_process_ids) - ->findArray(); + foreach(array_chunk($queue->subscribers['to_process'], self::BATCH_SIZE) + as $subscribers_to_process_ids + ) { + $found_subscribers = Subscriber::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($subscribers) !== count($subscribers_to_process_ids)) { - $queue->subscribers['to_process'] = Subscribers::updateCount( - Helpers::arrayColumn($subscribers, 'id'), + if(count($found_subscribers_ids) !== count($subscribers_to_process_ids)) { + $queue->subscribers = SubscribersTask::updateToProcessList( + $found_subscribers_ids, $subscribers_to_process_ids, - $queue->subscribers['to_process'] + $queue->subscribers ); } if(!count($queue->subscribers['to_process'])) { @@ -59,15 +64,24 @@ class SendingQueue { continue; } $queue->subscribers = call_user_func_array( - array($this, $processing_method), - array($mailer, $newsletter, $subscribers, $queue) + array( + $this, + $processing_method + ), + array( + $mailer, + $mta_log, + $newsletter, + $found_subscribers, + $queue + ) ); } } } - function processBulkSubscribers($mailer, $newsletter, $subscribers, $queue) { - $subscriber_log = array(); + // TODO: merge processBulkSubscribers with processIndividualSubscriber + function processBulkSubscribers($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 @@ -87,62 +101,80 @@ class SendingQueue { ); } // send - $result = MailerTask::send($mailer, $prepared_newsletters, $prepared_subscribers); - if(!$result) { - // record failed subscribers - $subscriber_log['failed'] = SubscribersTask::updateFailedList( - $queue->subscribers['failed'], - $subscribers_ids + $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 { - StatisticsTask::updateBulkNewsletterStatistics( + // update processed/to process list + $queue->subscribers = SubscribersTask::updateProcessedList( + $subscribers_ids, + $queue->subscribers + ); + // log statistics + StatisticsTask::processAndLogBulkNewsletterStatistics( $subscribers_ids, $newsletter['id'], $queue->id ); - MailerTask::updateMailerLog($this->mta_log); - $subscriber_log['processed'] = array_merge( - $queue->subscribers['processed'], - $subscribers_ids - ); + // keep track of sent items + $mta_log = MailerTask::updateMailerLog($mta_log); } - // TODO - //$queue = $this->updateQueue($queue, $subscriber_log); - MailerTask::checkSendingLimit($this->mta_config, $this->mta_log); + $this->updateQueue($queue); + MailerTask::checkSendingLimit($this->mta_config, $mta_log); CronHelper::checkExecutionTimer($this->timer); - return $queue->subscribers; } - /* function processIndividualSubscriber($mailer, $newsletter, $subscribers, $queue) { + function processIndividualSubscriber($mailer, $mta_log, $newsletter, $subscribers, $queue) { + $subscribers_ids = Helpers::arrayColumn($subscribers, 'id'); foreach($subscribers as $subscriber) { - $this->checkSendingLimit(); - $processed_newsletter = $this->prepareNewsletterForSending($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 + // render shortcodes and replace subscriber data in tracked links + $prepared_newsletter = + NewsletterTask::prepareNewsletterForSending( + $newsletter, + $subscriber, + $queue->asArray() ); - $this->updateMailerLog(); - $this->updateNewsletterStatistics($newsletter_statistics); + if(!$queue->newsletter_rendered_subject) { + $queue->newsletter_rendered_subject = $prepared_newsletter['subject']; } - $this->updateQueue($queue); + // format subscriber name/address according to mailer settings + $prepared_subscriber = MailerTask::prepareSubscriberForSending( + $mailer, + $subscriber + ); + $send_result = MailerTask::send($mailer, $prepared_newsletter, $prepared_subscriber); + 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::logStatistics( + array( + $newsletter['id'], + $subscriber['id'], + $queue->id + ) + ); + // keep track of sent items + $mta_log = MailerTask::updateMailerLog($mta_log); + } + $queue = $this->updateQueue($queue); + MailerTask::checkSendingLimit($this->mta_config, $mta_log); CronHelper::checkExecutionTimer($this->timer); } - return $queue->subscribers; - }*/ + } function getQueues() { return SendingQueueModel::orderByDesc('priority') @@ -152,19 +184,6 @@ class SendingQueue { } function updateQueue($queue) { - // TODO - return; - $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']); @@ -173,9 +192,16 @@ class SendingQueue { $queue->count_processed + $queue->count_to_process; if(!$queue->count_to_process) { $queue->processed_at = current_time('mysql'); - $queue->status = self::STATUS_COMPLETED; + $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; } } \ No newline at end of file diff --git a/lib/Cron/Workers/SendingQueue/Tasks/Posts.php b/lib/Cron/Workers/SendingQueue/Tasks/Posts.php index c5bd23cee6..c91da44aaa 100644 --- a/lib/Cron/Workers/SendingQueue/Tasks/Posts.php +++ b/lib/Cron/Workers/SendingQueue/Tasks/Posts.php @@ -13,17 +13,16 @@ class Posts { preg_match_all( '/data-post-id="(\d+)"/ism', $newsletter['rendered_body']['html'], - $matached_posts); - $matached_posts = $matached_posts[1]; - if(!count($matached_posts)) { + $matched_posts_ids); + $matched_posts_ids = $matched_posts_ids[1]; + if(!count($matched_posts_ids)) { return $newsletter; } - foreach($matached_posts as $post) { + foreach($matched_posts_ids as $post_id) { $newletter_post = NewsletterPost::create(); $newletter_post->newsletter_id = $newsletter['id']; - $newletter_post->post_id = $post; + $newletter_post->post_id = $post_id; $newletter_post->save(); } - return $newsletter; } } \ No newline at end of file diff --git a/lib/Cron/Workers/SendingQueue/Tasks/Statistics.php b/lib/Cron/Workers/SendingQueue/Tasks/Statistics.php index ceb2f368ee..92e6b6d8ad 100644 --- a/lib/Cron/Workers/SendingQueue/Tasks/Statistics.php +++ b/lib/Cron/Workers/SendingQueue/Tasks/Statistics.php @@ -7,7 +7,7 @@ use MailPoet\Util\Helpers; if(!defined('ABSPATH')) exit; class Statistics { - static function updateBulkNewsletterStatistics( + static function processAndLogBulkNewsletterStatistics( array $processed_subscribers_ids, $newsletter_id, $queue_id ) { $newsletter_statistics = array(); @@ -19,12 +19,10 @@ class Statistics { ); } $newsletter_statistics = Helpers::flattenArray($newsletter_statistics); + return self::logStatistics($newsletter_statistics); + } + + static function logStatistics($newsletter_statistics) { return StatisticsNewsletters::createMultiple($newsletter_statistics); } - - static function updateBulkNewsletterStatistics( - array $processed_subscribers_ids, $newsletter_id, $queue_id - ) { - - } } \ No newline at end of file diff --git a/lib/Cron/Workers/SendingQueue/Tasks/Subscribers.php b/lib/Cron/Workers/SendingQueue/Tasks/Subscribers.php index 0d58694d2f..6f9c74cb67 100644 --- a/lib/Cron/Workers/SendingQueue/Tasks/Subscribers.php +++ b/lib/Cron/Workers/SendingQueue/Tasks/Subscribers.php @@ -16,26 +16,48 @@ class Subscribers { } static function updateToProcessList( - array $existing_subscribers_ids, + array $found_subscribers_ids, array $subscribers_to_process_ids, - array $queue_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, - $existing_subscribers_ids + $found_subscribers_ids ); // remove nonexistent subscribers from the processing list - return array_diff( - $queue_subscribers_to_process_ids, - $subscibers_to_exclude + $queue_subscribers['to_process'] = array_diff( + $subscibers_to_exclude, + $queue_subscribers['to_process'] ); + return $queue_subscribers; } - static function updateFailedList(array $subscribers, array $failed_subscribers) { - return array_merge( - $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( + $failed_subscribers, + $queue_subscribers['to_process'] + ); + 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( + $processed_subscribers, + $queue_subscribers['to_process'] + ); + return $queue_subscribers; } } \ No newline at end of file diff --git a/lib/Newsletter/Renderer/PostProcess/OpenTracking.php b/lib/Newsletter/Renderer/PostProcess/OpenTracking.php index a5f256aa2f..780939969e 100644 --- a/lib/Newsletter/Renderer/PostProcess/OpenTracking.php +++ b/lib/Newsletter/Renderer/PostProcess/OpenTracking.php @@ -4,6 +4,8 @@ namespace MailPoet\Newsletter\Renderer\PostProcess; use MailPoet\Newsletter\Links\Links; class OpenTracking { + const OPEN_TRACKING_FILTER_NAME = 'mailpoet_rendering_post_process'; + static function process($template) { $DOM = new \pQuery(); $DOM = $DOM->parseStr($template); @@ -18,6 +20,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) { return OpenTracking::process($template); });