- Adds Queue router

- Updates logic for Queue and Supervisor
- #227

lib/Config/PublicAPI.php

@@ -0,0 +1,45 @@
+<?php
+namespace MailPoet\Config;
+
+use MailPoet\Queue\Queue;
+
+if(!defined('ABSPATH')) exit;
+
+class PublicAPI {
+  function __construct() {
+    # http://example.com/?mailpoet-api&section=&action=&payload=
+    $this->api = isset($_GET['mailpoet-api']) ? true : false;
+    $this->section = isset($_GET['section']) ? $_GET['section'] : false;
+    $this->action = isset($_GET['action']) ?
+      str_replace('_', '', lcfirst(ucwords($_GET['action'], '_'))) :
+      false;
+    $this->payload = isset($_GET['payload']) ?
+      json_decode(urldecode($_GET['payload']), true) :
+      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() {
+    $queue = new Queue($this->payload);
+    if(method_exists($queue, $this->action)) {
+      call_user_func(
+        array(
+          $queue,
+          $this->action
+        ));
+    }
+  }
+}

lib/Queue/Queue.php

@@ -0,0 +1,125 @@
+<?php
+namespace MailPoet\Queue;
+
+use MailPoet\Models\Setting;
+use MailPoet\Util\Security;
+
+require_once(ABSPATH . 'wp-includes/pluggable.php');
+
+if(!defined('ABSPATH')) exit;
+
+class Queue {
+  function __construct($payload = array()) {
+    set_time_limit(0);
+    ignore_user_abort();
+    list ($this->queue, $this->queueData) = $this->getQueue();
+    $this->refreshedToken = $this->refreshToken();
+    $this->payload = $payload;
+  }
+  
+  function start() {
+    if(!isset($this->payload['token'])) {
+      $this->abortWithError('missing token');
+    }
+    $queue = $this->queue;
+    $queueData = $this->queueData;
+    if(!$queue) {
+      $queue = Setting::create();
+      $queue->name = 'queue';
+      $queue->value = serialize(array('status' => 'stopped'));
+      $queue->save();
+    }
+    if(!preg_match('!stopped|paused!', $queueData['status'])
+    ) {
+      $queueData = array(
+        'status' => 'started',
+        'token' => $this->refreshedToken,
+        'executionCounter' => ($queueData['status'] === 'paused') ?
+          $queueData['executionCounter']
+          : 0,
+        'log' => array(
+          'token' => $this->payload['token'],
+          'message' => 'started'
+        )
+      );
+      $queue->value = serialize($queueData);
+      $queue->save();
+      $this->callSelf();
+    } else {
+      $queueData['log'] = array(
+        'token' => $this->payload['token'],
+        'status' => 'already started'
+      );
+      $queue->value = serialize($queueData);
+      $queue->save();
+    }
+  }
+  
+  function process() {
+    if(!$this->queue || $this->queueData['status'] !== 'started') {
+      $this->abortWithError('not running');
+    }
+    if(!isset($this->payload['token']) ||
+      $this->payload['token'] !== $this->queueData['token']
+    ) {
+      $this->abortWithError('invalid token');
+    }
+    
+    /*
+     * LOGIC WILL HAPPEN HERE
+     *
+     */
+    sleep(30);
+    
+    // after each execution, read queue in case its status was modified
+    list($queue, $queueData) = $this->getQueue();
+    $queueData['executionCounter']++;
+    $queueData['token'] = $this->refreshedToken;
+    $queue->value = serialize($queueData);
+    $queue->save();
+    $this->callSelf();
+  }
+
+  function callSelf() {
+    $payload = json_encode(array('token' => $this->refreshedToken));
+    $args = array(
+      'timeout' => 1,
+      'user-agent' => 'MailPoet (www.mailpoet.com)'
+    );
+    wp_remote_get(
+      site_url() .
+      '/?mailpoet-api&section=queue&action=process&payload=' . urlencode($payload),
+      $args
+    );
+    exit;
+  }
+  
+  function abortWithError($error) {
+    wp_send_json(
+      array(
+        'result' => false,
+        'error' => $error
+      ));
+    exit;
+  }
+  
+  function getQueue() {
+    $queue = Setting::where('name', 'queue')
+      ->findOne();
+    return array(
+      ($queue) ? $queue : null,
+      ($queue) ? unserialize($queue->value) : null
+    );
+  }
+  
+  function checkAuthorization() {
+    if(!current_user_can('manage_options')) {
+      header('HTTP/1.0 401 Not Authorized');
+      exit;
+    }
+  }
+  
+  function refreshToken() {
+    return Security::generateRandomString(5);
+  }
+}

lib/Queue/Supervisor.php

@@ -4,22 +4,24 @@ namespace MailPoet\Queue;
 use Carbon\Carbon;
 use MailPoet\Config\Env;
 use MailPoet\Models\Setting;
+use MailPoet\Util\Security;
 
 if(!defined('ABSPATH')) exit;
 
 class Supervisor {
-  function __construct() {
+  function __construct($forceStart = false) {
+    $this->forceStart = $forceStart;
     $this->checkDBReadiness();
     list ($this->queue, $this->queueData) = $this->getQueue();
   }
 
   function checkQueue() {
     if(!$this->queue) {
-      $this->startQueue();
+      return $this->startQueue();
     } else {
-      if($this->queueData['status'] === 'paused' &&
+      if(!$this->forceStart && ($this->queueData['status'] === 'paused' ||
         $this->queueData['status'] === 'stopped'
-      ) {
+      )) {
         return;
       }
       $currentTime = Carbon::now('UTC');
@@ -32,13 +34,29 @@ class Supervisor {
       $this->queueData['status'] = 'paused';
       $this->queue->value = serialize($this->queueData);
       $this->queue->save();
-      $this->startQueue();
+      return $this->startQueue();
     }
   }
 
   function startQueue() {
-    stream_context_set_default(array('http' => array('method' => 'HEAD')));
-    get_headers(home_url() . '/?mailpoet-api&section=queue&action=start', 1);
+    $args = array(
+      'timeout' => 1,
+      'user-agent' => 'MailPoet (www.mailpoet.com)'
+    );
+    $token = Security::generateRandomString(5);
+    $payload = json_encode(
+      array(
+        'token' => $token
+      )
+    );
+    wp_remote_get(
+      site_url() .
+      '/?mailpoet-api&section=queue&action=start&payload=' .
+      urlencode($payload),
+      $args
+    );
+    list ($queue, $queueData) = $this->getQueue();
+    return ($queueData && $queueData['token'] === $token) ? true : false;
   }
 
   function getQueue() {
@@ -53,7 +71,8 @@ class Supervisor {
 
   function checkDBReadiness() {
     $db = \ORM::forTable('')
-      ->rawQuery('SELECT COUNT(*) as settings FROM information_schema.tables ' .
+      ->rawQuery(
+        'SELECT COUNT(*) as settings FROM information_schema.tables ' .
         'WHERE table_schema = "' . Env::$db_name . '" ' .
         'AND table_name = "' . MP_SETTINGS_TABLE . '";'
       )

lib/Router/Queue.php

@@ -0,0 +1,51 @@
+<?php
+namespace MailPoet\Router;
+
+use MailPoet\Queue\Supervisor;
+
+if(!defined('ABSPATH')) exit;
+
+class Queue {
+  function start() {
+    $supervisor = new Supervisor();
+    wp_send_json(
+      array(
+        'result' => ($supervisor->checkQueue($forceStart = true)) ?
+          true :
+          false
+      )
+    );
+  }
+
+  function pause() {
+    wp_send_json(
+      array(
+        'result' => ($this->updateQueueStatus('paused') ?
+          true :
+          false
+        )
+      )
+    );
+  }
+
+  function stop() {
+    wp_send_json(
+      array(
+        'result' => ($this->updateQueueStatus('stopped') ?
+          true :
+          false
+        )
+      )
+    );
+  }
+
+  private function updateQueueStatus($status) {
+    $queue = new \MailPoet\Queue\Queue();
+    if(!$queue->queue || $queue->queueData['status'] !== 'started') {
+      return false;
+    }
+    $queue->queueData['status'] = $status;
+    $queue->queue->value = serialize($queue->queueData);
+    return $queue->queue->save();
+  }
+}

lib/Util/Security.php

@@ -1,8 +1,14 @@
 <?php
 namespace MailPoet\Util;
 
+require_once(ABSPATH . 'wp-includes/pluggable.php');
+
 class Security {
   static function generateToken() {
     return wp_create_nonce('mailpoet_token');
   }
+
+  static function generateRandomString($length) {
+    return substr(str_shuffle("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"), 0, $length);
+  }
 }