Worker Question

[tapankumar]

Hi,

I am able to submit job to queue (under my controller) and now would like to know how will i pick the same from queue.

If i am running below command (Under zend public folder), i am not getting anything on console.

php index.php queue beanstalkd encodingQueue

Should i make some worker code inside my module to fetch the job from queue? If yes, please can you give some example code.

Thanks

[tapankumar]

To understand how this module works for a newbie is really a tough business.

1. I have realized that if a run below command then the code inside BeanstalkdWorker.php (SlmQueueBeanstalkd\Worker) gets executed.

php index.php queue beanstalkd encodingQueue --timeout=5

Issues 1: As per SlmQueue below code should add data under content array.

public function encodeAction()
{
    // Here the parameters come from simple GET parameters
    $job = new EncodingJob(array(
        'originalUrl'       => $this->params()->fromQuery('originalUrl'),
        'destinationFormat' => $this->params()->fromQuery('destinationFormat')
    ));

    // Get the queue plugin manager
    $queueManager = $this->serviceLocator->get('SlmQueue\Queue\QueuePluginManager');
    $queue        = $queueManager->get('encodingQueue');

    $queue->push($job);
}

However, it is not adding data inside content array.

We have to call setContent method then only content array is getting filled with data.

Issue 2: How can i make my own worker controller which can do similar task which is being currently done by BeanstalkdWorker.php (SlmQueueBeanstalkd\Worker).

Thanks & Regards Tapan Thapa

[juriansluiman]

Can you please explain what you have done so far? It is important to know:

1. How you defined the queue
2. How you defined the job
3. How you have pushed a job into the queue
4. Perhaps more extensively how you start your worker

Working with queues might be tricky sometimes, there are multiple steps which can go wrong :)

[juriansluiman]

Oh, we crossed comments (this page was open for more than 32 minutes).

Yes we know the docs are lacking, so currently I am writing docs for SlmQueue (see juriansluiman/SlmQueue#77). Subsequently, we need to write better documentation for all adapter modules.

With your new code: what is the constructor of your EncodingJob? If you don't accept these parameters as arguments, the payload won't be stored. For your second question: what do you want to achieve? Do you need to modify the controller or the worker class? And for what reason?

[tapankumar]

Hi Juriansluiman,

Please find my inline reply.

1. How you defined the queue Comment: As per docs (https://github.com/juriansluiman/SlmQueueBeanstalkd), I have added below config under my module config.

'slm_queue' => array(
        'queue_manager' => array(
            'factories' => array(
                'encodingQueue' => 'SlmQueueBeanstalkd\Factory\BeanstalkdQueueFactory'
            )
        )
    ),

1. How you defined the job. Comment: I have created a Job folder inside src folder of my Module and added below class (Below is my Job Class)

<?php

namespace Api\Job;

use SlmQueue\Job\AbstractJob;

class EncodingJob extends AbstractJob {

    /**
     * Encode the video!
     */
    public function execute() {
        $originalUrl = $this->content['originalUrl'];
        $destinationFormat = $this->content['destinationFormat'];
        $this->content['destinationFormat'] = $destinationFormat . '.txt';
    }

}

1. How you have pushed a job into the queue Comment: I am making an API for which i have made below class. Through this i am adding job to queue.

<?php

namespace Api\Controller;

use Zend\Mvc\Controller\AbstractRestfulController;
use Zend\View\Model\JsonModel;
use Api\Job\EncodingJob;

class PushController extends AbstractRestfulController {

    protected $albumTable;

    public function getList() {
        //return new JsonModel(array('data' => "Welcome to the Zend Framework Album API example"));

        $username = $this->getEvent()->getRouteMatch()->getParam('username');
        $password = $this->getEvent()->getRouteMatch()->getParam('password');
        $receiver = $this->getEvent()->getRouteMatch()->getParam('receiver');
        $message = $this->getEvent()->getRouteMatch()->getParam('message');
        $sender = $this->getEvent()->getRouteMatch()->getParam('sender');

        // Here the parameters come from simple GET parameters
        $job = new EncodingJob();
        $job->setContent(array(
            'originalUrl' => 'http://www.yahoo.com',
            'destinationFormat' => '.mov'
        ));

        // Get the queue plugin manager        
        $queueManager = $this->serviceLocator->get('SlmQueue\Queue\QueuePluginManager');
        $queue = $queueManager->get('encodingQueue');

        $queue->push($job);
        return new JsonModel(array('status' => 'OK', 'id' => 1));
    }

    public function get($id) {
        //return new JsonModel(array('data' => $id));
        return new JsonModel(array('data' => $this->params()->fromQuery()));
    }

    public function create($data) {
// code...
    }

    public function update($id, $data) {
// code...
    }

    public function delete($id) {
// code...
    }

}

1. Perhaps more extensively how you start your worker Comment: I have started my worker via below command (I have not made any Worker class so far). php index.php queue beanstalkd encodingQueue --timeoout=5 Note: Getting below on my command line (I have modified BeanstalkdWorker.php under queue-beanstalkd\src\SlmQueueBeanstalkd\Worker to print content).

array(2) {
  'originalUrl' =>
  string(20) "http://www.yahoo.com"
  'destinationFormat' =>
  string(8) ".mov.txt"
}

Now what i want is as follows: List all tubes inside Beanstalkd and loop between all tubes so that my worker pics one job from each queue.

Let me know if i can provide more info here.

[tapankumar]

One more thing.

I have tried adding one worker inside my module and would like to know, how i will call from command line.

<?php

namespace Api\Worker;

use SlmQueue\Controller\AbstractWorkerController;

class WorkerController extends AbstractWorkerController {

    public function work() {
        $job = $this->processAction();
        var_dump($job);
    }

}

[juriansluiman]

@tapankumar there are a few issues here as I read along, most of them are independent from each other.

However, it is not adding data inside content array.

What you are doing is this:

$originalUrl = $this->content['originalUrl'];
$destinationFormat = $this->content['destinationFormat'];
$this->content['destinationFormat'] = $destinationFormat . '.txt';

The actual code flow is this: first you push a job with a certain payload into the queue, then the job is popped and inside the execute() method you can perform the work you need to do. There is no reason to "add" new content, since the execution of the job is the final step in the process. Why you set the content ($this->content['destinationFormat'] = $destinationFormat . '.txt';) is unclear to me. What you might be doing, as an example in the job, is more like this:

$originalUrl = $this->content['originalUrl'];
$destinationFormat = $this->content['destinationFormat'];
echo  $originalUrl . $destinationFormat . "\n";

When the worker is started, you should see some lines printed in your console.

List all tubes inside Beanstalkd and loop between all tubes so that my worker pics one job from each queue.

It is usually a better idea to know what tubes you are using. Then you define a queue for every tube you use. You specifically use a single queue in your controllers to push jobs into the queue. Next, you spin up as many workers as you have queues defined: one for every queue.

Why would you want to loop through all the tubes in a single worker?

[tapankumar]

@juriansluiman Thanks for the explanation of actual code flow. Now i got to know if i start worker with below command, how execute method of any job gets executed. php index.php queue beanstalkd encodingQueue --timeout=5 Please if possible, correct your documentation (Or explain in more better way) for newbie.

Now coming back to list all tubes question: I have almost 100-200 users and they all invoke my PHP API (Created earlier with PHP native code and now converting to Zend2) with their credentials. I want to store received messages to their own queue (user wise) so that i can pass the same to second API which i have.Priority of all user queue is same. No FIFO.

Now let say if i have 100-200 queues (user wise) with no job in it and if i am running 100-200 workers queue wise, it is just a resources wastage (Am i right don't know?).

My idea is to get list of tubes from beanstalkd (As beanstalkd removes the queue if they have no job in it) and loop between them so that all user queue will get equal weighted.

Let me know if this can be accomplished with this module or you have some better idea.

Thanks in advance.

[juriansluiman]

@tapankumar is this still an issue? I am still not sure why you need a queue per user. It is not the common way to do things, as you need a worker per queue (and this creates a number of issuers what you're aware of too).

If you really need a worker listening to multiple queues, I would extend the SlmQueueBeanstalk feature and add the option to listen to multiple tubes (which is possible with Beanstalkd iirc).

[juriansluiman]

@tapankumar closing now, please reopen if you have still questions about this.