ulabox / gearman-bundle
Gearman support for Symfony2.
Installs: 2 669
Dependents: 0
Suggesters: 0
Security: 0
Stars: 0
Watchers: 16
Forks: 1
Open Issues: 0
Type:symfony-bundle
Requires
- php: >=5.3.3
- doctrine/common: ~2.2
- jms/metadata: >=0.11
- jms/serializer-bundle: 0.12.*@dev
- symfony/framework-bundle: ~2.3
Requires (Dev)
- phpspec/php-diff: dev-master
- phpspec/phpspec2: dev-develop
This package is not auto-updated.
Last update: 2022-02-01 12:27:08 UTC
README
The bundle provides an interface between symfony2 projects and Gearman Job Server.
Quick installation
Require the bundle in your composer.json file:
{
"require": {
"ulabox/gearman-bundle": "*",
}
}
Run composer update command:
$ composer update ulabox/gearman-bundle
Now add the Bundle to your Kernel:
<?php
// app/AppKernel.php
public function registerBundles()
{
$bundles = array(
// ...
new Ulabox\Bundle\GearmanBundle\UlaboxGearmanBundle(),
// ...
);
}
Gearman dependency
To install Gearman Job Server for Debain/Ubuntu Package using the following commands:
$ sudo apt-get install gearman-job-server
Once the job server has been installed, can be started by running:
$ service gearman-job-server start
Install the Gearman driver
To install Gearman driver using the following commands:
$ pecl install channel://pecl.php.net/gearman-0.8.3
Now we just need to enable the module. Edit your /etc/php5/conf.d/gearman.ini file and add the line:
extension=gearman.so
Configuration
By default, the bundle doesn't need any configuration, but you can configure it editing your config.yml:
ulabox_gearman:
# Define your own servers.
# By default is set to localhost 127.0.0.1:4730.
# If annotations defined, will be overwritten.
servers:
localhost:
host: 127.0.0.1
port: 4730
server2:
host: myotherhost
port: 4567
# Define the default method to execute a task.
# By default is set to doBackgroundJob.
# Available methods (doBackgroundJob, doHighJob, doHighBackgroundJob, doLowJob, doLowBackgroundJob, doNormalJob,
# addTask, addTaskBackground, addTaskHigh, addTaskHighBackground, addTaskLow, addTaskLowBackground, runTasks).
default_method: doLowBackgroundJob
# Define the default number of executions before job dies.
# By default is set to 100.
# If annotations defined, will be overwritten.
iterations: 150
# Define your workers location directory.
# By default is set to Gearman/Worker
worker_dir: MyWorkerDir
# Define your clients location directory.
# By default is set to Gearman/Client
client_dir: MyClientDir
# Define the way the included event worker triggers the events
# cli for getting the event dispatcher, request for doing a curl
# against the configured request_context URL
# By default is set to request
use_via: [cli|request]
Writing Simple Worker
For a bundle located at src/Acme/DemoBundle, the worker classes by default should live inside src/Acme/DemoBundle/Gearman/Worker. We can create a simple worker as shown below:
<?php
namespace Acme\DemoBundle\Gearman\Worker;
use Ulabox\Bundle\GearmanBundle\Model\ContainerAwareWorker;
use Ulabox\Bundle\GearmanBundle\Annotation\Worker;
use Ulabox\Bundle\GearmanBundle\Annotation\Job;
/**
* The worker.
*
* @Worker()
*/
class AcmeWorker extends ContainerAwareWorker
{
/**
* The hello world job.
*
* @param \GearmanJob $job The GearmanJob instance
* @return boolean
* @Job()
*/
public function hello_world(\GearmanJob $job)
{
echo "Received hello world job: " . $job->handle() . "\n";
$workload = $job->workload();
$workloadSize = $job->workloadSize();
echo "Workload: $workload ($workloadSize)\n";
// This status loop is not needed, just to show how work
for ($i = 0; $i < $workloadSize; $i ++) {
echo "Sending status: " . $i . "/$workloadSize completed\n";
$job->sendStatus($i, $workloadSize);
sleep(1);
}
echo "Result: '".$workload."' loaded\n";
return true;
}
/**
* This job is never call, because is not marked with @Job annotation.
*
* @param \GearmanJob $job The GearmanJob instance
* @return boolean
*/
public function never_call_job(\GearmanJob $job)
{
...
}
}
Note how the the worker class is marked with @Worker() annotation and each method considered as a job is also marked with the @Job() annotation.
Executing the job
Once your job have been written, can be run in two ways:
By code:
// get the gearman manager
$gearmanManager = $this->container->get('ulabox_gearman.manager');
// get the generic gearman client
$client = $gearmanManager->getClient('UlaboxGearmanBundle:GearmanClient');
// find your worker
$worker = $gearmanManager->getWorker('AcmeDemoBundle:AcmeWorker');
// now you should tell the client that worker must be run
$client->setWorker($worker);
// and finally do the job
$client->doNormalJob('hello_world', json_encode(array('foo' => 'bar')));
// do the job in backgroud
//$client->doBackgroundJob('hello_world', json_encode(array('foo' => 'bar')));
To view the result open a console and run the command:
$ php app/console gearman:worker:execute --worker=AcmeDemoBundle:AcmeWorker
Then run the code above.
By the command line:
Open the first console and run:
$ php app/console gearman:worker:execute --worker=AcmeDemoBundle:AcmeWorker
Now open another console and run:
$ php app/console gearman:client:execute --client=UlaboxGearmanBundle:GearmanClient:hello_world --worker=AcmeDemoBundle:AcmeWorker --params="{\"foo\": \"bar\" }"
The commands come with a few options, you can see more details in the commands section.
Annotations
Worker annotations
- servers: Array containing servers
- iterations: The number of executions before job dies
<?php
namespace Acme\DemoBundle\Gearman\Worker;
use Ulabox\Bundle\GearmanBundle\Model\ContainerAwareWorker;
use Ulabox\Bundle\GearmanBundle\Annotation\Worker;
use Ulabox\Bundle\GearmanBundle\Annotation\Job;
/**
* The worker.
*
* @Worker(servers={"127.0.0.1:4730"}, iterations=10)
*/
class AcmeWorker extends ContainerAwareWorker
{
....
}
Client annotations
- worker: The worker name
- servers: Array containing servers
<?php
namespace Acme\DemoBundle\Gearman\Client;
use Ulabox\Bundle\GearmanBundle\Annotation\Client;
use Ulabox\Bundle\GearmanBundle\Model\Client as BaseClient;
/**
* The client
*
* @Client(worker="MyWorker", servers={"127.0.0.1:4730"})
*/
class AcmeClient extends BaseClient
....
}
Job annotations
- name: The job name
<?php
namespace Acme\DemoBundle\Gearman\Worker;
use Ulabox\Bundle\GearmanBundle\Model\ContainerAwareWorker;
use Ulabox\Bundle\GearmanBundle\Annotation\Worker;
use Ulabox\Bundle\GearmanBundle\Annotation\Job;
/**
* The worker.
*
* @Worker(servers={"127.0.0.1:4730"}, iterations=10)
*/
class AcmeWorker extends ContainerAwareWorker
{
/**
* The hello world job.
*
* @param \GearmanJob $job The GearmanJob instance
* @return boolean
*
* @Job(name="acme_hello_world")
*/
public function hello_world(\GearmanJob $job)
{
....
}
}
Writing Simple Client
For a bundle located at src/Acme/DemoBundle, the client classes by default should live inside src/Acme/DemoBundle/Gearman/Client. The client class by default is associated with the Worker class with the same name, for example, the AcmeClient will be associated with AcmeWorker:
<?php
namespace Acme\DemoBundle\Gearman\Client;
use Ulabox\Bundle\GearmanBundle\Annotation\Client;
use Ulabox\Bundle\GearmanBundle\Model\Client as BaseClient;
/**
* The client
*
* @Client()
*/
class AcmeClient extends BaseClient
{
public function hello_world_status($status)
{
print_r("AcmeClient::Status ".$status."\n");
}
public function hello_world_data($task)
{
print_r("AcmeClient::Data: ".$task->data()."\n");
}
public function hello_world_fail($task)
{
print_r("AcmeClient::Failed: ".$task->jobHandle()."\n");
}
public function hello_world_success($result)
{
print_r("AcmeClient::Success: ".$result);
}
}
Note how the the client class has a callback methods to be notified when the worker do the job, but it only makes sense when the job is executing with doNormalJob method.
Executing the job through client
Once your job have been written, can be run in two ways:
By code:
// get the gearman manager
$gearmanManager = $this->container->get('ulabox_gearman.manager');
// get the acme client
$client = $gearmanManager->getClient('AcmeDemoBundle:AcmeClient');
// and finally do the job
$client->doNormalJob('hello_world', json_encode(array('foo' => 'bar')));
// do the job in backgroud
//$client->doBackgroundJob('hello_world', json_encode(array('foo' => 'bar')));
To view the result open a console and run the command:
$ php app/console gearman:worker:execute --worker=AcmeDemoBundle:AcmeWorker
Then run the code above.
By the command line:
Open the first console and run:
$ php app/console gearman:worker:execute --worker=AcmeDemoBundle:AcmeWorker
Now open another console and run:
$ php app/console gearman:client:execute --client=AcmeDemoBundle:AcmeClient:hello_world --params="{\"foo\": \"bar\" }"
Run Tasks
Add a task to be run in parallel with other tasks is very simple. Adding to the previous worker a new job:
<?php
namespace Acme\DemoBundle\Gearman\Worker;
use Ulabox\Bundle\GearmanBundle\Model\ContainerAwareWorker;
use Ulabox\Bundle\GearmanBundle\Annotation\Worker;
use Ulabox\Bundle\GearmanBundle\Annotation\Job;
/**
* The worker.
*
* @Worker(servers={"127.0.0.1:4730"}, iterations=10)
*/
class AcmeWorker extends ContainerAwareWorker
{
/**
* Execute a job.
*
* @param \GearmanJob $job The GearmanJob instance
*
* @return boolean
*
* @Job()
*/
public function hello_world(\GearmanJob $job)
{
echo "Received hello world job: " . $job->handle() . "\n";
$workload = $job->workload();
$workloadSize = $job->workloadSize();
echo "Workload: $workload ($workloadSize)\n";
// This status loop is not needed, just to show how work
for ($i = 0; $i < $workloadSize; $i ++) {
echo "Sending status: " . $i . "/$workloadSize completed\n";
$job->sendStatus($i, $workloadSize);
sleep(1);
}
echo "Result: '".$workload."' loaded\n";
return true;
}
/**
* Execute a job.
*
* @param \GearmanJob $job The GearmanJob instance
*
* @return boolean
*
* @Job()
*/
public function send_newsletter(\GearmanJob $job)
{
$users = json_decode($job->workload(), true);
foreach ($users as $name => $email) {
echo "Be sent an email to $name\n";
// send the email
echo "The email have been send to $email\n\n";
sleep(1);
}
echo count($users)." mails have been sent \n";
return true;
}
}
Once your job have been written, can be run as show below:
// get the gearman manager
$gearmanManager = $this->container->get('ulabox_gearman.manager');
// get the acme client
$client = $gearmanManager->getClient('AcmeDemoBundle:AcmeClient');
// add multiple background tasks
$client->addTaskBackground('hello_world', json_encode(array('foo' => 'bar')));
$client->addTaskBackground('send_newsletter', json_encode(array('Ivan' => 'ivannis.suarez@gmail.com', 'Ulabox' => 'info@ulabox.com')));
// run tasks
$client->runTasks();
To view the result open a console and run the command:
$ php app/console gearman:worker:execute --worker=AcmeDemoBundle:AcmeWorker
Then run the code above.
Commands
Show all workers registered:
$ php app/console gearman:worker:list
Execute a job:
$ php app/console gearman:client:execute --help
Execute a worker:
$ php app/console gearman:worker:execute --help
Asynchronous Events
Exploiting the potential of Gearman Job Server, we have introduced a new EventDispatcherAsync that replaces the default symfony EventDispatcher, the new dispatcher has a new method called dispatchAsync that sends the event to the gearman queue, gearman processes it and sends it back to the php application, which finally reconstructs the event and run it asynchronously.
Configuration
To use asynchronous events the first thing to do is import the routing.yml file of the gearman bundle.
# app/config/routing.yml
gearman_bundle:
resource: @UlaboxGearmanBundle/Resources/config/routing.yml
prefix: /_gearman
And add the route in the access control from your security.yml file.
# app/config/security.yml
security:
# ...
access_control:
- { path: ^/_gearman*, roles: IS_AUTHENTICATED_ANONYMOUSLY, ips: [127.0.0.1, ::1] }
Enable the asynchronous event dispatcher in your config.yml.
# app/config/config.yml
ulabox_gearman:
enable_asynchronous_event_dispatcher: true
The EventWorker is responsible for processing the event and send it back to the app. This worker needs to generate a url from the command line, and this requires us setup the request context globally.
# app/config/parameters.yml
parameters:
router.request_context.host: example.org
router.request_context.scheme: https
router.request_context.base_url: my/path
Now that everything is configured the following step is run the EventWorker in the command line
$ php app/console gearman:worker:execute --worker=UlaboxGearmanBundle:EventWorker
The EventWorker should always be running, so we recommend using a tool such as the supervisor to ensure that the worker is always running.
Writing an async Event
An async Event is a simple event with the only difference that must implement the interface Ulabox\Bundle\GearmanBundle\Dispatcher\AsyncEventInterface. Here we show a small example:
<?php
namespace Acme\DemoBundle\Event;
use Ulabox\Bundle\GearmanBundle\Dispatcher\AsyncEventInterface;
use Symfony\Component\EventDispatcher\Event;
use Acme\DemoBundle\Entity\User;
/**
* The custom async event.
*/
class FooEvent extends Event implements AsyncEventInterface
{
/**
* The user entity
*
* @var User
*/
protected $user;
/**
* Construstor.
*
* @param User $user The user instance
*/
public function __construct(User $user)
{
$this->user = $user;
}
/**
* {@inheritdoc}
*/
public function getUser()
{
return $this->user;
}
/**
* {@inheritdoc}
*/
public function getArguments()
{
return array(
'user' => $this->user
);
}
/**
* {@inheritdoc}
*/
public function setArguments(array $args = array())
{
$this->user = $args['user'];
}
}
Note that the class implements the two AsyncEventInterface methods (getArguments, setArguments), these are used internally to reconstruct the event.
Create an Event Listener
<?php
namespace Acme\DemoBundle\EventListener;
use Acme\DemoBundle\Event\FooEvent;
/**
* Foo listener
*/
class FooListener
{
/**
* Listener on foo event
*
* @param FooEvent $event The event
*/
public function onFoo(FooEvent $event)
{
$user = $event->getUser();
// do something
}
}
Now that the class is created, you just need to register it as a service:
# app/config/config.yml
services:
acme.listener.foo_event:
class: Acme\DemoBundle\EventListener\FooListener
tags:
- { name: kernel.event_listener, event: foo.event, method: onFoo }
Dispatch the Event
The dispatchAsync() method begins the process and notifies asynchronously all listeners of the given event
<?php
namespace Acme\DemoBundle\Controller;
use Acme\DemoBundle\Event\FooEvent;
/**
* Demo controller
*/
class DemoController
{
public function someAction()
{
...
$user = $this->get('security.context')->getToken()->getUser();
$eventDispatcher = $this->get('event_dispatcher');
// dispatch the event
$eventDispatcher->dispatchAsync('foo.event', new FooEvent($user));
...
}
}