- Refactors sending queue worker by breaking it into smaller tasks

- Adds arrayUnique method to Helpers for multidimensional arrays

lib/Cron/Workers/SendingQueue/SendingQueue.php

@@ -0,0 +1,181 @@
+<?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\Statistics as StatisticsTask;
+use MailPoet\Cron\Workers\SendingQueue\Tasks\Subscribers as SubscribersTask;
+use MailPoet\Models\SendingQueue as SendingQueueModel;
+use MailPoet\Models\Setting;
+use MailPoet\Models\Subscriber;
+use MailPoet\Util\Helpers;
+
+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() {
+    foreach($this->getQueues() as $queue) {
+      $newsletter = NewsletterTask::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->subscribers = SubscribersTask::get($queue->asArray());
+      // configure mailer with newsletter data (from/reply-to)
+      $mailer = MailerTask::configureMailer($newsletter);
+      $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();
+        // 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'),
+            $subscribers_to_process_ids,
+            $queue->subscribers['to_process']
+          );
+        }
+        if(!count($queue->subscribers['to_process'])) {
+          $this->updateQueue($queue);
+          continue;
+        }
+        $queue->subscribers = call_user_func_array(
+          array($this, $processing_method),
+          array($mailer, $newsletter, $subscribers, $queue)
+        );
+      }
+    }
+  }
+
+  function processBulkSubscribers($mailer, $newsletter, $subscribers, $queue) {
+    $subscriber_log = array();
+    $subscribers_ids = Helpers::arrayColumn($subscribers, 'id');
+    foreach($subscribers as $subscriber) {
+      // render shortcodes and replace subscriber data in tracked links
+      $prepared_newsletters[] =
+        NewsletterTask::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[] = MailerTask::prepareSubscriberForSending(
+        $mailer,
+        $subscriber
+      );
+    }
+    // send
+    $result = MailerTask::send($mailer, $prepared_newsletters, $prepared_subscribers);
+    if(!$result) {
+      // record failed subscribers
+      $subscriber_log['failed'] = SubscribersTask::updateFailedList(
+        $queue->subscribers['failed'],
+        $subscribers_ids
+      );
+    } else {
+      StatisticsTask::updateBulkNewsletterStatistics(
+        $subscribers_ids,
+        $newsletter['id'],
+        $queue->id
+      );
+      MailerTask::updateMailerLog($this->mta_log);
+      $subscriber_log['processed'] = array_merge(
+        $queue->subscribers['processed'],
+        $subscribers_ids
+      );
+    }
+    // TODO
+    //$queue = $this->updateQueue($queue, $subscriber_log);
+    MailerTask::checkSendingLimit($this->mta_config, $this->mta_log);
+    CronHelper::checkExecutionTimer($this->timer);
+    return $queue->subscribers;
+  }
+
+ /* function processIndividualSubscriber($mailer, $newsletter, $subscribers, $queue) {
+    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
+        );
+        $this->updateMailerLog();
+        $this->updateNewsletterStatistics($newsletter_statistics);
+      }
+      $this->updateQueue($queue);
+      CronHelper::checkExecutionTimer($this->timer);
+    }
+    return $queue->subscribers;
+  }*/
+
+  function getQueues() {
+    return SendingQueueModel::orderByDesc('priority')
+      ->whereNull('deleted_at')
+      ->whereNull('status')
+      ->findResultSet();
+  }
+
+  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']);
+    $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 = self::STATUS_COMPLETED;
+    }
+    $queue->subscribers = serialize((array) $queue->subscribers);
+    $queue->save();
+  }
+}

lib/Cron/Workers/SendingQueue/Tasks/Links.php

@@ -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']);
+  }
+}
+

lib/Cron/Workers/SendingQueue/Tasks/Mailer.php

@@ -0,0 +1,95 @@
+<?php
+namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
+
+use MailPoet\Mailer\Mailer as MailerFactory;
+use MailPoet\Models\Setting;
+
+if(!defined('ABSPATH')) exit;
+
+class Mailer {
+  static function configureMailer(array $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 MailerFactory($method = false, $sender, $reply_to);
+    return $mailer;
+  }
+
+  static function getMailerConfig() {
+    $mta_config = Setting::getValue('mta');
+    if(!$mta_config) {
+      throw new \Exception(__('Mailer is not configured.'));
+    }
+    return $mta_config;
+  }
+
+  static function updateMailerLog($mta_log) {
+    $mta_log['sent']++;
+    Setting::setValue('mta_log', $mta_log);
+    return $mta_log;
+  }
+
+  static 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;
+  }
+
+  static function getProcessingMethod($mta_config) {
+    return ($mta_config['method'] === 'MailPoet') ?
+      'processBulkSubscribers' :
+      'processIndividualSubscriber';
+  }
+
+  static function prepareSubscriberForSending($mailer, $subscriber) {
+    return ($mailer instanceof \MailPoet\Mailer\Mailer) ?
+      $mailer->transformSubscriber($subscriber) :
+      false;
+  }
+
+  static 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 &&
+      $elapsed_time <= $frequency_interval
+    ) {
+      throw new \Exception(__('Sending frequency limit has been reached.'));
+    }
+    if($elapsed_time > $frequency_interval) {
+      $mta_log = array(
+        'sent' => 0,
+        'started' => time()
+      );
+      Setting::setValue('mta_log', $mta_log);
+    }
+    return;
+  }
+}

lib/Cron/Workers/SendingQueue/Tasks/Newsletter.php

@@ -0,0 +1,96 @@
+<?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 {
+  static 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']);
+    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((boolean) Setting::getValue('tracking.enabled')) {
+      // add tracking image
+      OpenTracking::addTrackingImage();
+      // render newsletter
+      $newsletter = self::render($newsletter);
+      // hash and save all links
+      $newsletter = LinksTask::process($newsletter, $queue);
+    } else {
+      // render newsletter
+      $newsletter = self::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;
+  }
+
+  static function render($newsletter) {
+    $renderer = new Renderer($newsletter);
+    $newsletter['rendered_body'] = $renderer->render();
+    return $newsletter;
+  }
+
+  static 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((boolean) Setting::getValue('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]
+      )
+    );
+  }
+}

lib/Cron/Workers/SendingQueue/Tasks/Posts.php

@@ -0,0 +1,29 @@
+<?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'],
+      $matached_posts);
+    $matached_posts = $matached_posts[1];
+    if(!count($matached_posts)) {
+      return $newsletter;
+    }
+    foreach($matached_posts as $post) {
+      $newletter_post = NewsletterPost::create();
+      $newletter_post->newsletter_id = $newsletter['id'];
+      $newletter_post->post_id = $post;
+      $newletter_post->save();
+    }
+    return $newsletter;
+  }
+}

lib/Cron/Workers/SendingQueue/Tasks/Shortcodes.php

@@ -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);
+  }
+}
+

lib/Cron/Workers/SendingQueue/Tasks/Statistics.php

@@ -0,0 +1,30 @@
+<?php
+namespace MailPoet\Cron\Workers\SendingQueue\Tasks;
+
+use MailPoet\Models\StatisticsNewsletters;
+use MailPoet\Util\Helpers;
+
+if(!defined('ABSPATH')) exit;
+
+class Statistics {
+  static function updateBulkNewsletterStatistics(
+    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 StatisticsNewsletters::createMultiple($newsletter_statistics);
+  }
+
+  static function updateBulkNewsletterStatistics(
+    array $processed_subscribers_ids, $newsletter_id, $queue_id
+  ) {
+
+  }
+}

lib/Cron/Workers/SendingQueue/Tasks/Subscribers.php

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