- Renames and updates sending queue worker

- Updates mailer router's send() method
2015-12-04 12:31:54 -05:00
parent beb939df9e
commit c2fde308cb
7 changed files with 125 additions and 62 deletions
--- a/lib/Cron/Workers/SendingQueue.php
+++ b/lib/Cron/Workers/SendingQueue.php
@ -0,0 +1,114 @@
+<?php
+namespace MailPoet\Cron\Workers;
+
+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;
+
+class SendingQueue {
+  function __construct($timer = false) {
+    $this->timer = ($timer) ? $timer : microtime(true);
+  }
+
+  function process() {
+    $queues =
+      \MailPoet\Models\SendingQueue::orderByDesc('priority')
+        ->whereNull('deleted_at')
+        ->whereNull('status')
+        ->findResultSet();
+    foreach($queues as $queue) {
+      $newsletter = Newsletter::findOne($queue->newsletter_id);
+      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
+        )
+      );
+      $subscribers = json_decode($queue->subscribers, true);
+      $subscribersToProcess = $subscribers['to_process'];
+      if(!isset($subscribers['failed'])) $subscribers['failed'] = array();
+      if(!isset($subscribers['processed'])) $subscribers['processed'] = array();
+      foreach(array_chunk($subscribersToProcess, 200) as $subscriberIds) {
+        $dbSubscribers = Subscriber::whereIn('id', $subscriberIds)
+          ->findArray();
+        foreach($dbSubscribers as $i => $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();
+          if($result) {
+            $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();
+        }
+      }
+    }
+  }
+
+  function checkExecutionTimer() {
+    $elapsedTime = microtime(true) - $this->timer;
+    if($elapsedTime >= 28) throw new \Exception('Maximum execution time reached.');
+  }
+}