Subscribes to an specific event given a subject and a queue.

@param string   $subject  Message topic.
@param string   $queue    Queue name.
@param \Closure $callback Closure to be executed as callback.

@return void
187768
public function queueSubscribe($subject, $queue, \Closure $callback)
    {
        $c = function ($message) use ($callback) {
            $message->setBody($this->encoder->decode($message->getBody()));
            $callback($message);
        };
        parent::queueSubscribe($subject, $queue, $c);
    }
939558
public function setSubject(NotificationInterface $notification, $subject, $flush = false)
    {
        $notification->setSubject($subject);
        $this->flush($flush);

        $event = new NotificationEvent($notification);
        $this->dispatcher->dispatch(MgiletNotificationEvents::MODIFIED, $event);

        return $notification;
    }
575703
public function subscribe(callable $handler)
    {
        $this->handlers[] = $handler;

        // laze subscribe
        if (!$this->subscribed) {
            $this->client->subscribe($this->queue_name, function ($message) {
                $this->handle($message);
            });
            $this->subscribed = true;
        }
    }
820708
public function update(SplSubject $subject)
    {
        if ($this->hasMethod('__update') && is_callable([$this, '__update'])) {
            $this->__update($subject);
        }
    }
761575
public function subscribe(string $event, string $listener, callable $callback)
    {
        $callbackKey = $event . ' ' . $listener;
        $this->events[$event][] = $callbackKey;
        $this->subscriptions[$callbackKey] = $callback;
    }
320667
public function handleSubscribeCallback($message, $expr, callable $handler)
    {
        $event = $this->translator->translate($message);
        if ($event) {
            // we were able to translate the message into an event
            if ($event->matches($expr)) {
                // the event matches the listen expression
                call_user_func($handler, $event);
            } else {
                // pass to listen expr fail handler?
                if ($this->listenExprFailHandler) {
                    call_user_func($this->listenExprFailHandler, $event, $expr);
                }
            }
        } else {
            // pass to translate fail handler?
            if ($this->translateFailHandler) {
                call_user_func($this->translateFailHandler, $message);
            }
        }
    }
624928
public function addConsumer($queue, ConsumerInterface $consumer, $priority = 0)
    {
        $this->messageDispatcher->addListener($queue, array($consumer, 'consume'), $priority);
    }
1014002
public function listen( Closure $callback )
	{
		if ( !isset( $this->dispatcher ) )
			throw new RuntimeException( 'Events dispatcher has not been set.' );

		Hooks::addHook( 'logger', $callback );
	}
225219
public function createMessageAction(Subject $subject)
    {
        $form = $this->container->get('form.factory')->create(new MessageType, new Message());
        $form->handleRequest($this->get('request'));


        if ($form->isValid()) {
            $message = $form->getData();
            $this->manager->createMessage($message, $subject);
        }

        return new RedirectResponse(
            $this->generateUrl('claro_forum_messages', array('subject' => $subject->getId()))
        );
    }
580083
public function addMessageToQueue($message, $queue)
    {
        $this->queue->sendMessage([
            'QueueUrl' => $this->getQueueURL($queue),
            'MessageBody' => $message,
        ]);
    }
470370
public function setSubject($subject = '')
    {
        $this->subject = $subject;
        $this->debug->debug(__FUNCTION__, 'setSubject: ', $this->subject);
    }
476809
public function update(SplSubject $subject, $event = null)
    {
        switch ($event) {
            case 'log':
                /* @var Doozr_Logging $subject */
                $logs = $subject->getCollectionRaw();

                foreach ($logs as $log) {
                    $this->log(
                        $log['type'],
                        $log['message'],
                        unserialize($log['context']),
                        $log['time'],
                        $log['fingerprint'],
                        $log['separator']
                    );
                }
                break;
        }
    }
566300
public function request($subject, $payload, \Closure $callback)
    {
        $inbox = uniqid('_INBOX.');
        $sid   = $this->subscribe(
            $inbox,
            $callback
        );
        $this->unsubscribe($sid, 1);
        $this->publish($subject, $payload, $inbox);
        $this->wait(1);
    }
39745
public function push(callable $callback)
    {
        if ($callback instanceof self) {
            $this->queue = array_merge($this->queue, $callback->queue);
            return;
        }

        $this->queue[] = $callback;
    }
903383
public function listen(Closure $callback)
    {
        if (! isset($this->dispatcher)) {
            throw new RuntimeException('Events dispatcher has not been set.');
        }

        $this->dispatcher->listen(MessageLogged::class, $callback);
    }
362259
protected function _error(Subject $subject)
    {
        $subject->set(['success' => false, 'created' => false]);

        $this->_trigger('afterSave', $subject);
        $this->setFlash('error', $subject);
        $this->_trigger('beforeRender', $subject);
    }
811733
public function stickSubject(Subject $subject)
    {
        $this->om->startFlushSuite();
        $subject->setIsSticked(true);
        $this->om->persist($subject);
        $this->dispatch(new StickSubjectEvent($subject));
        $this->om->endFlushSuite();
    }