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 string
345801
public function queueSubscribe($subject, $queue, \Closure $callback)
    {
        $sid = $this->randomGenerator->generateString(16);
        $msg = 'SUB '.$subject.' '.$queue.' '.$sid;
        $this->send($msg);
        $this->subscriptions[$sid] = $callback;
        return $sid;
    }
792570
public function closeSubjectAction(Subject $subject)
    {
        $forum = $subject->getCategory()->getForum();
        $this->checkAccess($forum);
        $this->manager->closeSubject($subject);

        return new RedirectResponse(
            $this->generateUrl('claro_forum_subjects', array('category' => $subject->getCategory()->getId()))
        );
    }
828226
public function actionMessages($queue_name, $subscriber_id = null)
    {
        if (($subscriber_id = trim($subscriber_id)) === '') {
            $subscriber_id = null;
        }
        /** @var QueueInterface $queue */
        list($queue, $authItems) = $this->loadQueue($queue_name, ['nfy.message.read', 'nfy.message.create']);
        $this->verifySubscriber($queue, $subscriber_id);

        $formModel = new models\MessageForm();
        if ($authItems['nfy.message.create'] && isset($_POST['MessageForm'])) {
            $formModel->attributes = $_POST['MessageForm'];
            if ($formModel->validate()) {
                $queue->send($formModel->content, $formModel->category);

                return $this->redirect([
                    'messages',
                    'queue_name' => $queue_name,
                    'subscriber_id' => $subscriber_id,
                ]);
            }
        }

        $dataProvider = null;
        if ($authItems['nfy.message.read']) {
            $dataProvider = new \yii\data\ArrayDataProvider([
                'allModels' => $queue->peek($subscriber_id, 200, [
                    components\Message::AVAILABLE,
                    components\Message::RESERVED,
                    components\Message::DELETED,
                ]),
                'sort' => [
                    'attributes' => ['id'],
                    'defaultOrder' => ['id' => SORT_DESC],
                ],
            ]);
            // reverse display order to simulate a chat window, where latest message is right above the message form
            $dataProvider->setModels(array_reverse($dataProvider->getModels()));
        }

        return $this->render('messages', [
            'queue' => $queue,
            'queue_name' => $queue_name,
            'dataProvider' => $dataProvider,
            'model' => $formModel,
            'authItems' => $authItems,
        ]);
    }
329372
public function updateSubject(int $conversationId, string $subject): void
    {
        $patch = Patch::replace('subject', $subject);
        $this->patchConversation($conversationId, $patch);
    }
1204369
public function subject($subject)
	{
		if ($this->config['encode_headers'])
		{
			$subject = $this->encode_mimeheader((string) $subject);
		}
		$this->subject = (string) $subject;

		return $this;
	}
1090703
public function executeSending( $subject, $body, $subscriber )
    {
        $email = $subscriber->attribute( 'email' );
        $parameters = array();
        $parameters['content_type'] = $this->emailContentType;
        $parameters['from'] = $this->emailFrom;
        $transport = eZNotificationTransport::instance( 'ezmail' );
        $result = $transport->send( array( $email ), $subject, $body, null, $parameters );
        if ( $result === false )
        {
            throw new Exception( 'Send email error! Subscriber id:' .$subscriber->attribute( 'id' ) );
        }
        eZDebugSetting::writeNotice( 'extension-ezcomments', "An email has been sent to '$email' (subject: $subject)", __METHOD__ );
    }
248693
public function call(Closure $callback = null, $caller = null)
    {
        if ($callback instanceof Closure) {
            // Make dependency injection possible
            $parameters = $this->resolveMethodDependencies(
                [$caller], new ReflectionFunction($callback)
            );
            call_user_func_array($callback, $parameters);
        }

        return $caller;
    }
674960
private function useReflectionToGetMessage($subject)
    {
        $reflection = new \ReflectionClass($subject);
        $property   = $reflection->getProperty('_message');
        $property->setAccessible(true);
        $message = $property->getValue($subject);

        return $message;
    }
187121
public function injectMessageReplacements($message, $subject, $substitutions) {
		// Handle empty messages
		if(empty($subject) && empty($message)) return array(null, null);

		// Check if there's a role specific message
		$subjectText = str_replace(
			array_keys($substitutions),
			array_values($substitutions),
			$subject ?: $message
		);
		$messageText = str_replace(
			array_keys($substitutions),
			array_values($substitutions),
			$message ?: $subject
		);
		return array($subjectText, $messageText);
	}
663937
public function subscribe($topicQueueName, callable $handler)
    {
        $isSubscriptionLoopActive = true;
        while ($isSubscriptionLoopActive) {
            $request = new SubscribeMessageRequest([
                'queueName'          => $topicQueueName,
                'pollingWaitSeconds' => 0,
            ]);

            try {
                $message = $this->client->send($request);
            } catch (RequestException $e) {
                throw $e;
            } catch (ResponseException $e) {
                $message = $e->getBody();

                if (!$message->has('code')) {
                    throw $e;
                } else if (!in_array($message->get('code'), ['7000', '6070'])) {    // 队列没有消息或队列中有太多不可见或者延时消息
                    throw $e;
                }
            } catch (\Exception $e) {
                throw $e;
            }

            if ($message === null || $message->get('msgBody') === null) {
                sleep(mt_rand(1, 5));
                continue;
            }

            call_user_func($handler,
                Utils::unserializeMessagePayload($message->get('msgBody')),
                new MessageHandler($this->client, $topicQueueName, $message)
            );

            unset($message);
        }
    }
716792
public function unstickSubject(Subject $subject)
    {
        $this->om->startFlushSuite();
        $subject->setIsSticked(false);
        $this->om->persist($subject);
        $this->dispatch(new UnstickSubjectEvent($subject));
        $this->om->endFlushSuite();
    }
1285071
public void askSubscribers( Actor sender, String topic, Object message, Callback cb ) {
        List<ReceiverActor> subscriber = topic2Subscriber.get(topic);
        if ( subscriber != null ) {
            List<IPromise> results = subscriber.stream()
                    .filter(subs -> !subs.equals(sender))
                    .map(subs -> subs.receiveAsk(topic, message))
                    .collect(Collectors.toList());
            try {
                all((List) results).await(5000); // is non-blocking
            } catch (Exception ex) {
                // timeout goes here
                Log.Info(this, "timeout in broadcast");
            }
            results.forEach( promise -> cb.pipe(promise.get()));
            cb.finish(); // important to release callback mapping in remoting !
        }
    }
793954
def subscribe(
            self,
            subject="",
            queue="",
            cb=None,
            future=None,
            max_msgs=0,
            is_async=False,
            pending_msgs_limit=DEFAULT_SUB_PENDING_MSGS_LIMIT,
            pending_bytes_limit=DEFAULT_SUB_PENDING_BYTES_LIMIT,
    ):
        
        if self.is_closed:
            raise ErrConnectionClosed

        if self.is_draining:
            raise ErrConnectionDraining

        self._ssid += 1
        sid = self._ssid
        sub = Subscription(
            subject=subject,
            queue=queue,
            cb=cb,
            future=future,
            max_msgs=max_msgs,
            is_async=is_async,
            sid=sid,
        )
        self._subs[sid] = sub

        if cb is not None:
            sub.pending_msgs_limit = pending_msgs_limit
            sub.pending_bytes_limit = pending_bytes_limit
            sub.pending_queue = tornado.queues.Queue(
                maxsize=pending_msgs_limit)

            @tornado.gen.coroutine
            def wait_for_msgs():
                while True:
                    sub = wait_for_msgs.sub
                    err_cb = wait_for_msgs.err_cb

                    try:
                        sub = wait_for_msgs.sub
                        if sub.closed:
                            break

                        msg = yield sub.pending_queue.get()
                        if msg is None:
                            break

                        sub.received += 1
                        sub.pending_size -= len(msg.data)
                        if sub.max_msgs > 0 and sub.received >= sub.max_msgs:
                            # If we have hit the max for delivered msgs, remove sub.
                            self._subs.pop(sub.sid, None)
                            self._remove_subscription(sub)

                        # Invoke depending of type of handler.
                        if sub.is_async:
                            # NOTE: Deprecate this usage in a next release,
                            # the handler implementation ought to decide
                            # the concurrency level at which the messages
                            # should be processed.
                            self._loop.spawn_callback(sub.cb, msg)
                        else:
                            yield sub.cb(msg)
                    except Exception as e:
                        # All errors from calling an async subscriber
                        # handler are async errors.
                        if err_cb is not None:
                            yield err_cb(e)

            # Bind the subscription and error cb if present
            wait_for_msgs.sub = sub
            wait_for_msgs.err_cb = self._error_cb
            self._loop.spawn_callback(wait_for_msgs)

        elif future is not None:
            # Used to handle the single response from a request
            # based on auto unsubscribe.
            sub.future = future

        # Send SUB command...
        sub_cmd = b''.join([
            SUB_OP, _SPC_,
            sub.subject.encode(), _SPC_,
            sub.queue.encode(), _SPC_, ("%d" % sid).encode(), _CRLF_
        ])
        yield self.send_command(sub_cmd)
        yield self._flush_pending()
        raise tornado.gen.Return(sid)
76003
public function unstickSubjectAction(Subject $subject)
    {
        $forum = $subject->getCategory()->getForum();
        $this->checkAccess($forum);
        $this->manager->unstickSubject($subject);

        return new RedirectResponse(
            $this->generateUrl('claro_forum_subjects', array('category' => $subject->getCategory()->getId()))
        );
    }
1353963
public function await(QueueClientInterface $queue, ReceiveParametersInterface $params = null)
    {
        $worker = $this;
        $this->await = true;
        $lastResult = [];

        $callback = function(QueueEvent $event) use($worker, $queue, &$lastResult) {

            $messages = $event->getMessages();

            foreach ($messages as $message) {
                $lastResult = $worker->process($message);

                if ($queue->canDeleteMessage()) {
                    $queue->delete($message);
                }
            }

            if ($worker->isAwaitingStopped()) {
                $event->stopAwait(true);
            }

            return !$worker->isAwaitingStopped();
        };


        $callbackHandler = $queue->getEventManager()->attach(QueueEvent::EVENT_RECEIVE, $callback);

        $queue->await($params);

        $queue->getEventManager()->detach($callbackHandler);


        return $lastResult;
    }
873573
public function consumeMessages($queue, $callback, $flags = AMQP_NOPARAM)
    {
        if (!isset($this->_queues[$queue])) {
            throw new InvalidKeyException(['consume message from queue failed: `:queue queue is NOT exists`', 'queue' => $queue]);
        }

        try {
            $this->_queues[$queue]->consume(function (\AMQPEnvelope $envelope) use ($callback, $queue) {
                return $callback(new Message($this, $queue, $envelope));
            }, $flags);
        } catch (\Exception $e) {
            throw new AmqpException('consume `:queue` queue message failed: ', $e->getMessage());
        }
    }
905379
public function subject($subject = null)
    {
        if ($subject !== null) {
            $this->_subject = $subject;

            return $this;
        }

        return $this->_subject;
    }