contamobi / CmobiRabbitmqBundle

use Cmobi\RabbitmqBundle\Queue\QueueBuilderInterface;

use Cmobi\RabbitmqBundle\Queue\QueueServiceInterface;

class RpcServerBuilder implements QueueBuilderInterface

{

    private $queue;

    private $connectionManager;

    private $connectionName;

    public function __construct(ConnectionManager $connManager, $connectionName = 'default')

    {

        $this->connectionManager = $connManager;

        $this->connectionName = $connectionName;

        $this->channel = null;

    }

    /**

     * @param $queueName

     * @param QueueServiceInterface $queueService

     * @param QueueBagInterface $queueBag

     * @return Queue

     * @throws \Exception

     */

    public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag)

    {

        if (! $queueBag instanceof RpcQueueBag) {

            throw new \Exception('Unsupported QueueBag');

        }

        $queue = new Queue($this->getConnectionManager(), $queueBag, $this->connectionName);

        $queueCallback = new RpcQueueCallback($queueService);

        $queue->setCallback($queueCallback);

        $this->queue = $queue;

        return $queue;

    }

    /**

     * @return ConnectionManager

     */

    public function getConnectionManager()

    {

        return $this->connectionManager;

    }

    /**

     * @return Queue

     */

    public function getQueue()

    {

        return $this->queue;

    }

}

use Cmobi\RabbitmqBundle\Queue\QueueBuilderInterface;

use Cmobi\RabbitmqBundle\Queue\QueueServiceInterface;

class SubscriberBuilder implements QueueBuilderInterface

{

    private $queue;

    private $connectionManager;

    private $connectionName;

    public function __construct(ConnectionManager $connManager, $connectionName = 'default')

    {

        $this->connectionManager = $connManager;

        $this->connectionName = $connectionName;

        $this->channel = null;

    }

    /**

     * @param $queueName

     * @param QueueServiceInterface $queueService

     * @param QueueBagInterface $queueBag

     * @return Queue

     * @throws \Exception

     */

    public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag)

    {

        if (! $queueBag instanceof SubscriberQueueBag) {

            throw new \Exception('Unsupported QueueBag');

        }

        $queue = new Queue($this->getConnectionManager(), $queueBag, $this->connectionName);

        $queueCallback = new SubscriberQueueCallback($queueService);

        $queue->setCallback($queueCallback);

        $this->queue = $queue;

        return $queue;

    }

    /**

     * @return ConnectionManager

     */

    public function getConnectionManager()

    {

        return $this->connectionManager;

    }

    /**

     * @return Queue

     */

    public function getQueue()

    {

        return $this->queue;

    }

}

use Cmobi\RabbitmqBundle\Queue\QueueBuilderInterface;

use Cmobi\RabbitmqBundle\Queue\QueueServiceInterface;

class WorkerBuilder implements QueueBuilderInterface

{

    private $queue;

    private $connectionManager;

    private $connectionName;

    public function __construct(ConnectionManager $connManager, $connectionName = 'default')

    {

        $this->connectionManager = $connManager;

        $this->connectionName = $connectionName;

        $this->channel = null;

    }

    /**

     * @param $queueName

     * @param QueueServiceInterface $queueService

     * @param QueueBagInterface $queueBag

     * @return Queue

     * @throws \Exception

     */

    public function buildQueue($queueName, QueueServiceInterface $queueService, QueueBagInterface $queueBag)

    {

        if (! $queueBag instanceof WorkerQueueBag) {

            throw new \Exception('Unsupported QueueBag');

        }

        $queue = new Queue($this->getConnectionManager(), $queueBag, $this->connectionName);

        $queueCallback = new WorkerQueueCallback($queueService);

        $queue->setCallback($queueCallback);

        $this->queue = $queue;

        return $queue;

    }

    /**

     * @return ConnectionManager

     */

    public function getConnectionManager()

    {

        return $this->connectionManager;

    }

    /**

     * @return Queue

     */

    public function getQueue()

    {

        return $this->queue;

    }

}