Add async command system to handle asynchronous operations
This commit is contained in:
parent
f5b6226732
commit
74ae7b8929
|
@ -35,7 +35,7 @@ abstract class QueuedJob extends Job {
|
|||
* @param \OC\Log $logger
|
||||
*/
|
||||
public function execute($jobList, $logger = null) {
|
||||
$jobList->remove($this);
|
||||
$jobList->remove($this, $this->argument);
|
||||
parent::execute($jobList, $logger);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,70 @@
|
|||
<?php
|
||||
/**
|
||||
* Copyright (c) 2015 Robin Appelman <icewind@owncloud.com>
|
||||
* This file is licensed under the Affero General Public License version 3 or
|
||||
* later.
|
||||
* See the COPYING-README file.
|
||||
*/
|
||||
|
||||
namespace OC\Command;
|
||||
|
||||
use OCP\Command\IBus;
|
||||
use OCP\Command\ICommand;
|
||||
use SuperClosure\Serializer;
|
||||
|
||||
/**
|
||||
* Asynchronous command bus that uses the background job system as backend
|
||||
*/
|
||||
class AsyncBus implements IBus {
|
||||
/**
|
||||
* @var \OCP\BackgroundJob\IJobList
|
||||
*/
|
||||
private $jobList;
|
||||
|
||||
/**
|
||||
* @param \OCP\BackgroundJob\IJobList $jobList
|
||||
*/
|
||||
function __construct($jobList) {
|
||||
$this->jobList = $jobList;
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule a command to be fired
|
||||
*
|
||||
* @param \OCP\Command\ICommand | callable $command
|
||||
*/
|
||||
public function push($command) {
|
||||
$this->jobList->add($this->getJobClass($command), $this->serializeCommand($command));
|
||||
}
|
||||
|
||||
/**
|
||||
* @param \OCP\Command\ICommand | callable $command
|
||||
* @return string
|
||||
*/
|
||||
private function getJobClass($command) {
|
||||
if ($command instanceof \Closure) {
|
||||
return 'OC\Command\ClosureJob';
|
||||
} else if (is_callable($command)) {
|
||||
return 'OC\Command\CallableJob';
|
||||
} else if ($command instanceof ICommand) {
|
||||
return 'OC\Command\CommandJob';
|
||||
} else {
|
||||
throw new \InvalidArgumentException('Invalid command');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param \OCP\Command\ICommand | callable $command
|
||||
* @return string
|
||||
*/
|
||||
private function serializeCommand($command) {
|
||||
if ($command instanceof \Closure) {
|
||||
$serializer = new Serializer();
|
||||
return $serializer->serialize($command);
|
||||
} else if (is_callable($command) or $command instanceof ICommand) {
|
||||
return serialize($command);
|
||||
} else {
|
||||
throw new \InvalidArgumentException('Invalid command');
|
||||
}
|
||||
}
|
||||
}
|
|
@ -0,0 +1,22 @@
|
|||
<?php
|
||||
/**
|
||||
* Copyright (c) 2015 Robin Appelman <icewind@owncloud.com>
|
||||
* This file is licensed under the Affero General Public License version 3 or
|
||||
* later.
|
||||
* See the COPYING-README file.
|
||||
*/
|
||||
|
||||
namespace OC\Command;
|
||||
|
||||
use OC\BackgroundJob\QueuedJob;
|
||||
|
||||
class CallableJob extends QueuedJob {
|
||||
protected function run($serializedCallable) {
|
||||
$callable = unserialize($serializedCallable);
|
||||
if (is_callable($callable)) {
|
||||
$callable();
|
||||
} else {
|
||||
throw new \InvalidArgumentException('Invalid serialized callable');
|
||||
}
|
||||
}
|
||||
}
|
|
@ -0,0 +1,24 @@
|
|||
<?php
|
||||
/**
|
||||
* Copyright (c) 2015 Robin Appelman <icewind@owncloud.com>
|
||||
* This file is licensed under the Affero General Public License version 3 or
|
||||
* later.
|
||||
* See the COPYING-README file.
|
||||
*/
|
||||
|
||||
namespace OC\Command;
|
||||
|
||||
use OC\BackgroundJob\QueuedJob;
|
||||
use SuperClosure\Serializer;
|
||||
|
||||
class ClosureJob extends QueuedJob {
|
||||
protected function run($serializedCallable) {
|
||||
$serializer = new Serializer();
|
||||
$callable = $serializer->unserialize($serializedCallable);
|
||||
if (is_callable($callable)) {
|
||||
$callable();
|
||||
} else {
|
||||
throw new \InvalidArgumentException('Invalid serialized callable');
|
||||
}
|
||||
}
|
||||
}
|
|
@ -0,0 +1,26 @@
|
|||
<?php
|
||||
/**
|
||||
* Copyright (c) 2015 Robin Appelman <icewind@owncloud.com>
|
||||
* This file is licensed under the Affero General Public License version 3 or
|
||||
* later.
|
||||
* See the COPYING-README file.
|
||||
*/
|
||||
|
||||
namespace OC\Command;
|
||||
|
||||
use OC\BackgroundJob\QueuedJob;
|
||||
use OCP\Command\ICommand;
|
||||
|
||||
/**
|
||||
* Wrap a command in the background job interface
|
||||
*/
|
||||
class CommandJob extends QueuedJob {
|
||||
protected function run($serializedCommand) {
|
||||
$command = unserialize($serializedCommand);
|
||||
if ($command instanceof ICommand) {
|
||||
$command->handle();
|
||||
} else {
|
||||
throw new \InvalidArgumentException('Invalid serialized command');
|
||||
}
|
||||
}
|
||||
}
|
|
@ -38,6 +38,7 @@ use OC\AppFramework\Http\Request;
|
|||
use OC\AppFramework\Db\Db;
|
||||
use OC\AppFramework\Utility\SimpleContainer;
|
||||
use OC\Cache\UserCache;
|
||||
use OC\Command\AsyncBus;
|
||||
use OC\Diagnostics\NullQueryLogger;
|
||||
use OC\Diagnostics\EventLogger;
|
||||
use OC\Diagnostics\QueryLogger;
|
||||
|
@ -291,6 +292,10 @@ class Server extends SimpleContainer implements IServerContainer {
|
|||
$this->registerService('IniWrapper', function ($c) {
|
||||
return new IniGetWrapper();
|
||||
});
|
||||
$this->registerService('AsyncCommandBus', function (Server $c) {
|
||||
$jobList = $c->getJobList();
|
||||
return new AsyncBus($jobList);
|
||||
});
|
||||
$this->registerService('TrustedDomainHelper', function ($c) {
|
||||
return new TrustedDomainHelper($this->getConfig());
|
||||
});
|
||||
|
@ -777,6 +782,13 @@ class Server extends SimpleContainer implements IServerContainer {
|
|||
return $this->query('IniWrapper');
|
||||
}
|
||||
|
||||
/**
|
||||
* @return \OCP\Command\IBus
|
||||
*/
|
||||
function getAsyncCommandBus(){
|
||||
return $this->query('AsyncCommandBus');
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the trusted domain helper
|
||||
*
|
||||
|
|
|
@ -0,0 +1,18 @@
|
|||
<?php
|
||||
/**
|
||||
* Copyright (c) 2015 Robin Appelman <icewind@owncloud.com>
|
||||
* This file is licensed under the Affero General Public License version 3 or
|
||||
* later.
|
||||
* See the COPYING-README file.
|
||||
*/
|
||||
|
||||
namespace OCP\Command;
|
||||
|
||||
interface IBus {
|
||||
/**
|
||||
* Schedule a command to be fired
|
||||
*
|
||||
* @param \OCP\Command\ICommand | callable $command
|
||||
*/
|
||||
public function push($command);
|
||||
}
|
|
@ -0,0 +1,16 @@
|
|||
<?php
|
||||
/**
|
||||
* Copyright (c) 2015 Robin Appelman <icewind@owncloud.com>
|
||||
* This file is licensed under the Affero General Public License version 3 or
|
||||
* later.
|
||||
* See the COPYING-README file.
|
||||
*/
|
||||
|
||||
namespace OCP\Command;
|
||||
|
||||
interface ICommand {
|
||||
/**
|
||||
* Run the command
|
||||
*/
|
||||
public function handle();
|
||||
}
|
|
@ -317,4 +317,9 @@ interface IServerContainer {
|
|||
* @return \bantu\IniGetWrapper\IniGetWrapper
|
||||
*/
|
||||
function getIniWrapper();
|
||||
|
||||
/**
|
||||
* @return \OCP\Command\IBus
|
||||
*/
|
||||
function getAsyncCommandBus();
|
||||
}
|
||||
|
|
|
@ -21,13 +21,18 @@ class DummyJobList extends \OC\BackgroundJob\JobList {
|
|||
|
||||
private $last = 0;
|
||||
|
||||
public function __construct(){}
|
||||
public function __construct() {
|
||||
}
|
||||
|
||||
/**
|
||||
* @param \OC\BackgroundJob\Job|string $job
|
||||
* @param mixed $argument
|
||||
*/
|
||||
public function add($job, $argument = null) {
|
||||
if (is_string($job)) {
|
||||
/** @var \OC\BackgroundJob\Job $job */
|
||||
$job = new $job;
|
||||
}
|
||||
$job->setArgument($argument);
|
||||
if (!$this->has($job, null)) {
|
||||
$this->jobs[] = $job;
|
||||
|
|
|
@ -0,0 +1,143 @@
|
|||
<?php
|
||||
|
||||
/**
|
||||
* Copyright (c) 2015 Robin Appelman <icewind@owncloud.com>
|
||||
* This file is licensed under the Affero General Public License version 3 or
|
||||
* later.
|
||||
* See the COPYING-README file.
|
||||
*/
|
||||
|
||||
namespace Test\Command;
|
||||
|
||||
use OCP\Command\ICommand;
|
||||
use Test\BackgroundJob\DummyJobList;
|
||||
use Test\TestCase;
|
||||
|
||||
class SimpleCommand implements ICommand {
|
||||
public function handle() {
|
||||
AsyncBus::$lastCommand = 'SimpleCommand';
|
||||
}
|
||||
}
|
||||
|
||||
class StateFullCommand implements ICommand {
|
||||
private $state;
|
||||
|
||||
function __construct($state) {
|
||||
$this->state = $state;
|
||||
}
|
||||
|
||||
public function handle() {
|
||||
AsyncBus::$lastCommand = $this->state;
|
||||
}
|
||||
}
|
||||
|
||||
function basicFunction() {
|
||||
AsyncBus::$lastCommand = 'function';
|
||||
}
|
||||
|
||||
class AsyncBus extends TestCase {
|
||||
/**
|
||||
* Basic way to check output from a command
|
||||
*
|
||||
* @var string
|
||||
*/
|
||||
public static $lastCommand;
|
||||
|
||||
/**
|
||||
* @var \OCP\BackgroundJob\IJobList
|
||||
*/
|
||||
private $jobList;
|
||||
|
||||
/**
|
||||
* @var \OCP\Command\IBus
|
||||
*/
|
||||
private $bus;
|
||||
|
||||
public static function DummyCommand() {
|
||||
self::$lastCommand = 'static';
|
||||
}
|
||||
|
||||
public function setUp() {
|
||||
$this->jobList = new DummyJobList();
|
||||
$this->bus = new \OC\Command\AsyncBus($this->jobList);
|
||||
self::$lastCommand = '';
|
||||
}
|
||||
|
||||
public function testSimpleCommand() {
|
||||
$command = new SimpleCommand();
|
||||
$this->bus->push($command);
|
||||
$this->runJobs();
|
||||
$this->assertEquals('SimpleCommand', self::$lastCommand);
|
||||
}
|
||||
|
||||
public function testStateFullCommand() {
|
||||
$command = new StateFullCommand('foo');
|
||||
$this->bus->push($command);
|
||||
$this->runJobs();
|
||||
$this->assertEquals('foo', self::$lastCommand);
|
||||
}
|
||||
|
||||
public function testStaticCallable() {
|
||||
$this->bus->push(['\Test\Command\AsyncBus', 'DummyCommand']);
|
||||
$this->runJobs();
|
||||
$this->assertEquals('static', self::$lastCommand);
|
||||
}
|
||||
|
||||
public function testMemberCallable() {
|
||||
$command = new StateFullCommand('bar');
|
||||
$this->bus->push([$command, 'handle']);
|
||||
$this->runJobs();
|
||||
$this->assertEquals('bar', self::$lastCommand);
|
||||
}
|
||||
|
||||
public function testFunctionCallable() {
|
||||
$this->bus->push('\Test\Command\BasicFunction');
|
||||
$this->runJobs();
|
||||
$this->assertEquals('function', self::$lastCommand);
|
||||
}
|
||||
|
||||
public function testClosure() {
|
||||
$this->bus->push(function () {
|
||||
AsyncBus::$lastCommand = 'closure';
|
||||
});
|
||||
$this->runJobs();
|
||||
$this->assertEquals('closure', self::$lastCommand);
|
||||
}
|
||||
|
||||
public function testClosureSelf() {
|
||||
$this->bus->push(function () {
|
||||
self::$lastCommand = 'closure-self';
|
||||
});
|
||||
$this->runJobs();
|
||||
$this->assertEquals('closure-self', self::$lastCommand);
|
||||
}
|
||||
|
||||
private function privateMethod() {
|
||||
self::$lastCommand = 'closure-this';
|
||||
}
|
||||
|
||||
public function testClosureThis() {
|
||||
$this->bus->push(function () {
|
||||
$this->privateMethod();
|
||||
});
|
||||
$this->runJobs();
|
||||
$this->assertEquals('closure-this', self::$lastCommand);
|
||||
}
|
||||
|
||||
public function testClosureBind() {
|
||||
$state = 'bar';
|
||||
$this->bus->push(function () use ($state) {
|
||||
self::$lastCommand = 'closure-' . $state;
|
||||
});
|
||||
$this->runJobs();
|
||||
$this->assertEquals('closure-bar', self::$lastCommand);
|
||||
}
|
||||
|
||||
|
||||
private function runJobs() {
|
||||
$jobs = $this->jobList->getAll();
|
||||
foreach ($jobs as $job) {
|
||||
$job->execute($this->jobList);
|
||||
}
|
||||
}
|
||||
}
|
Loading…
Reference in New Issue