Merge pull request #238 from mailpoet/queue

Queue

assets/js/src/ajax.js

@@ -10,7 +10,8 @@ define('ajax', ['mailpoet', 'jquery'], function(MailPoet, jQuery) {
         token: null,
         data: {},
         onSuccess: function(data, textStatus, xhr) {},
-        onError: function(xhr, textStatus, errorThrown) {}
+        onError: function(xhr, textStatus, errorThrown) {},
+        onComplete: function(xhr) {}
       },
       get: function(options) {
         return this.request('get', options);
@@ -18,6 +19,9 @@ define('ajax', ['mailpoet', 'jquery'], function(MailPoet, jQuery) {
       post: function(options) {
         return this.request('post', options);
       },
+      head: function(options) {
+        return this.request('head', options);
+      },
       delete: function(options) {
         return this.request('delete', options);
       },
@@ -60,14 +64,23 @@ define('ajax', ['mailpoet', 'jquery'], function(MailPoet, jQuery) {
             this.options.onSuccess,
             'json'
           );
-        } else {
+        }
+        else if (method === 'head') {
+          jqXHR = jQuery.ajax({
+            url: this.options.url,
+            type : 'head',
+            complete : this.options.onComplete
+          });
+        }
+        else {
           jqXHR = jQuery.ajax({
             url: this.options.url,
             type : 'post',
             data: params,
             dataType: 'json',
             success : this.options.onSuccess,
-            error : this.options.onError
+            error : this.options.onError,
+            complete : this.options.onComplete
           });
         }
 

composer.json

@@ -8,7 +8,9 @@
     "tburry/pquery": "*",
     "j4mie/paris": "1.5.4",
     "swiftmailer/swiftmailer": "^5.4",
-    "phpseclib/phpseclib": "*"
+    "phpseclib/phpseclib": "*",
+    "nesbot/carbon": "*",
+    "mtdowling/cron-expression": "^1.0"
   },
   "require-dev": {
     "codeception/codeception": "*",

lib/Config/Initializer.php

@@ -2,6 +2,7 @@
 namespace MailPoet\Config;
 
 use MailPoet\Models;
+use MailPoet\Queue\Supervisor;
 use MailPoet\Router;
 
 if(!defined('ABSPATH')) exit;
@@ -25,6 +26,8 @@ class Initializer {
     $this->setupAnalytics();
     $this->setupPermissions();
     $this->setupChangelog();
+    $this->setupPublicAPI();
+    $this->runQueueSupervisor();
   }
 
   function setupDB() {
@@ -33,7 +36,8 @@ class Initializer {
     \ORM::configure('password', Env::$db_password);
     \ORM::configure('logging', WP_DEBUG);
     \ORM::configure('driver_options', array(
-      \PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES utf8'
+      \PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES utf8',
+      \PDO::MYSQL_ATTR_INIT_COMMAND => 'SET TIME_ZONE = "+00:00"'
     ));
 
     $subscribers = Env::$db_prefix . 'subscribers';
@@ -48,6 +52,7 @@ class Initializer {
     $subscriber_custom_field = Env::$db_prefix . 'subscriber_custom_field';
     $newsletter_option_fields = Env::$db_prefix . 'newsletter_option_fields';
     $newsletter_option = Env::$db_prefix . 'newsletter_option';
+    $queue = Env::$db_prefix . 'queue';
 
     define('MP_SUBSCRIBERS_TABLE', $subscribers);
     define('MP_SETTINGS_TABLE', $settings);
@@ -61,6 +66,7 @@ class Initializer {
     define('MP_SUBSCRIBER_CUSTOM_FIELD_TABLE', $subscriber_custom_field);
     define('MP_NEWSLETTER_OPTION_FIELDS_TABLE', $newsletter_option_fields);
     define('MP_NEWSLETTER_OPTION_TABLE', $newsletter_option);
+    define('MP_QUEUE_TABLE', $queue);
   }
 
   function setupActivator() {
@@ -110,4 +116,14 @@ class Initializer {
     $changelog = new Changelog();
     $changelog->init();
   }
-}
+  
+  function setupPublicAPI() {
+    $publicAPI = new PublicAPI();
+    $publicAPI->init();
+  }
+
+  function runQueueSupervisor() {
+    $supervisor = new Supervisor();
+    $supervisor->checkQueue();
+  }
+}

lib/Config/Migrator.php

@@ -21,6 +21,7 @@ class Migrator {
       'subscriber_custom_field',
       'newsletter_option_fields',
       'newsletter_option',
+      'queue',
       'forms'
     );
   }
@@ -199,6 +200,21 @@ class Migrator {
     return $this->sqlify(__FUNCTION__, $attributes);
   }
 
+  function queue() {
+    $attributes = array(
+      'id mediumint(9) NOT NULL AUTO_INCREMENT,',
+      'newsletter_id mediumint(9) NOT NULL,',
+      'subscribers longtext,',
+      'total mediumint(9) NOT NULL DEFAULT 0,',
+      'processed mediumint(9) NOT NULL DEFAULT 0,',
+      'created_at TIMESTAMP NOT NULL DEFAULT 0,',
+      'updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,',
+      'deleted_at TIMESTAMP NULL DEFAULT NULL,',
+      'PRIMARY KEY  (id)',
+    );
+    return $this->sqlify(__FUNCTION__, $attributes);
+  }
+
   function forms() {
     $attributes = array(
       'id mediumint(9) NOT NULL AUTO_INCREMENT,',

lib/Config/PublicAPI.php

@@ -0,0 +1,44 @@
+<?php
+namespace MailPoet\Config;
+
+use MailPoet\Queue\Queue;
+use Symfony\Component\Config\Definition\Exception\Exception;
+
+if(!defined('ABSPATH')) exit;
+
+class PublicAPI {
+  function __construct() {
+    # http://example.com/?mailpoet-api&section=&action=&data=
+    $this->api = isset($_GET['mailpoet-api']) ? true : false;
+    $this->section = isset($_GET['section']) ? $_GET['section'] : false;
+    $this->action = isset($_GET['action']) ? $_GET['action'] : false;
+    $this->data = isset($_GET['data']) ? $_GET['data'] : false;
+  }
+
+  function init() {
+    if(!$this->api && !$this->section) return;
+    if(method_exists($this, $this->section)) {
+      call_user_func(
+        array(
+          $this,
+          $this->section
+        ));
+    }
+    else {
+      header('HTTP/1.0 404 Not Found');
+    }
+    exit;
+  }
+
+  function queue() {
+    $method = str_replace('_', '', lcfirst(ucwords($this->action, '_')));
+    $queue = new Queue();
+    if(method_exists($queue, $method)) {
+      call_user_func(
+        array(
+          $queue,
+          $method
+        ));
+    }
+  }
+}

lib/Mailer/SMTP.php

@@ -20,6 +20,7 @@ class SMTP {
       $message = $this->createMessage($newsletter, $subscriber);
       $result = $this->mailer->send($message);
     } catch (\Exception $e) {
+      return $e->getMessage();
       $result = false;
     }
     return ($result === 1);

lib/Models/Queue.php

@@ -0,0 +1,12 @@
+<?php
+namespace MailPoet\Models;
+
+if(!defined('ABSPATH')) exit;
+
+class Queue extends Model {
+  public static $_table = MP_QUEUE_TABLE;
+
+  function __construct() {
+    parent::__construct();
+  }
+}

lib/Queue/Queue.php

@@ -0,0 +1,125 @@
+<?php
+namespace MailPoet\Queue;
+
+use Carbon\Carbon;
+use MailPoet\Models\Setting;
+
+if(!defined('ABSPATH')) exit;
+
+class Queue {
+  function __construct() {
+    $this->checkRequestMethod();
+    set_time_limit(0);
+    ignore_user_abort();
+    ob_start();
+    list ($this->queue, $this->queueData) = $this->getQueue();
+  }
+
+  function start() {
+    $_start = function ($queue, $queueData) {
+      $queue->value = serialize($queueData);
+      $queue->updated_at = date('Y-m-d H:i:s');
+      $queue->save();
+      $this->setHeadersAndFlush($queueData['status']);
+      $this->callSelf();
+    };
+    if(!$this->queue) {
+      $queue = Setting::create();
+      $queueData = array(
+        'status' => 'started',
+        'executionCounter' => 0
+      );
+      $queue->name = 'queue';
+      $_start($queue, $queueData);
+    }
+    $queue = $this->queue;
+    $queueData = $this->queueData;
+    if($queueData['status'] === 'stopped' || $queueData['status'] === 'paused') {
+      if($queueData['status'] !== 'paused') $queueData['executionCounter'] = 0;
+      $queueData['status'] = 'started';
+      $_start($queue, $queueData);
+    }
+    $this->setHeadersAndFlush('already running', true);
+  }
+
+  function process() {
+    if(!$this->queue) {
+      $this->setHeadersAndFlush('not running', true);
+    }
+    $queue = $this->queue;
+    $queueData = $this->queueData;
+    if($queueData['status'] !== 'started') {
+      $this->setHeadersAndFlush('not running', true);
+    }
+    // TODO: check if the queue is already being executed
+      $currentTime = Carbon::now('UTC');
+      $lastUpdateTime = Carbon::createFromFormat('Y-m-d H:i:s', $queue->updated_at, 'UTC');
+      $timeSinceLastStart = $currentTime->diffInSeconds($lastUpdateTime);
+      $this->setHeadersAndFlush('processing');
+    sleep(30); // THIS WILL BE REPLACED BY SENDING LOGIC
+    list($queue, $queueData) = $this->getQueue();
+    $queueData['executionCounter']++;
+    $queue->value = serialize($queueData);
+    $queue->save();
+    // TODO: remove
+        $setting = Setting::create();
+        $setting->name = date('H:i:s');
+        $setting->save();
+    $this->callSelf();
+  }
+
+  function pause() {
+    $this->updateQueue('paused');
+  }
+
+  function stop() {
+    $this->updateQueue('stopped');
+  }
+
+  function updateQueue($status = false) {
+    if (!$status) return;
+    if(!$this->queue || $this->queueData['status'] !== 'started') {
+      $this->setHeadersAndFlush('not running', true);
+    }
+    $queue = $this->queue;
+    $queueData = $this->queueData;
+    $queueData['status'] = $status;
+    $queue->value = serialize($queueData);
+    $queue->save();
+    $this->setHeadersAndFlush($queueData['status'], true);
+  }
+
+  function setHeadersAndFlush($status, $terminate = false) {
+    ob_end_clean();
+    header('Connection: close');
+    header('X-MailPoet-Queue: ' . $status);
+    ob_end_flush();
+    ob_flush();
+    flush();
+    if($terminate) exit;
+  }
+
+  function callSelf() {
+    stream_context_set_default(array('http' => array('method' => 'HEAD')));
+    get_headers(home_url() . '/?mailpoet-api&section=queue&action=process', 1);
+    exit;
+  }
+
+  function getQueue() {
+    $queue = Setting::where('name', 'queue')
+      ->findOne();
+    $queueData = ($queue) ? unserialize($queue->value) : false;
+    return array(
+      $queue,
+      $queueData
+    );
+  }
+
+  function checkRequestMethod() {
+    $method = $_SERVER['REQUEST_METHOD'];
+    if($method !== 'HEAD') {
+      header('HTTP/1.0 405 Method Not Allowed');
+      exit;
+    }
+  }
+}

lib/Queue/Supervisor.php

@@ -0,0 +1,48 @@
+<?php
+namespace MailPoet\Queue;
+
+use Carbon\Carbon;
+use MailPoet\Models\Setting;
+
+if(!defined('ABSPATH')) exit;
+
+class Supervisor {
+  function __construct() {
+    list ($this->queue, $this->queueData) = $this->getQueue();
+  }
+
+  function checkQueue() {
+    if(!$this->queue) {
+      $this->startQueue();
+    } else {
+      if($this->queueData['status'] !== 'paused' &&
+        $this->queueData['status'] !== 'stopped'
+      ) {
+        return;
+      }
+      $currentTime = Carbon::now('UTC');
+      $lastUpdateTime = Carbon::createFromFormat('Y-m-d H:i:s', $this->queue->updated_at, 'UTC');
+      $timeSinceLastStart = $currentTime->diffInSeconds($lastUpdateTime);
+      if($timeSinceLastStart < 50) return;
+      $this->queueData['status'] = 'paused';
+      $this->queue->value = serialize($this->queueData);
+      $this->queue->save();
+      $this->startQueue();
+    }
+  }
+
+  function startQueue() {
+    stream_context_set_default(array('http' => array('method' => 'HEAD')));
+    get_headers(home_url() . '/?mailpoet-api&section=queue&action=start', 1);
+  }
+
+  function getQueue() {
+    $queue = Setting::where('name', 'queue')
+      ->findOne();
+    $queueData = ($queue) ? unserialize($queue->value) : false;
+    return array(
+      $queue,
+      $queueData
+    );
+  }
+}

views/queue.html

@@ -0,0 +1,22 @@
+<% extends 'layout.html' %>
+<% block content %>
+<br>
+<a href="#" class="button-primary" id="queue" data-status="stopped">Start Queue</a>
+<script type="text/javascript">
+  jQuery(document).ready(function () {
+    jQuery('#queue[data-status="stopped"]').click(function () {
+      var element = jQuery(this);
+      MailPoet.Ajax
+       .head({
+         // TODO: update with WP site_url();
+         url: jQuery('location').attr('host') + '?mailpoet&action=start_queue',
+       })
+       .complete(function (xhr) {
+         if (xhr.getResponseHeader('X-MailPoet-Queue') === 'started') {
+           element.text('Started').attr('data-id', 'running');
+         }
+       });
+    });
+  });
+</script>
+<% endblock %>