Synchronous stream of events (Message instances).

@param string $topic
@param array $options

@return Message[] A traversable stream of Message
320568
public function subscribe(string $topic, array $options = [])
    {
        for (;;) {
            if (isset($this->messages[$topic])) {
                foreach ($this->messages[$topic] as $key => $message) {
                    yield $message;

                    unset($this->messages[$topic][$key]);
                }
            }

            sleep(1);
        }
    }
998829
public function message($message, $options = null) {
		$options = $this->_mergeOptions($options);

		$this->set($message, $options);

		$this->_assertSessionStackSize($options);
	}
150463
public function onPublish(Conn $conn, $topic, $event, array $exclude, array $eligible)
    {
        /*
         * $topic contains the FULL uri, so if anything extra was added to that, you can find out the full path based on that.
         */

        $topic->broadcast(array(
            "sender" => $conn->resourceId,
            "topic" => $topic->getId(),
            "event" => $event
        ));
    }
1094592
public function fromArray($options)
    {
        if (!empty($options['KeyFrameInterval'])) {
            Validate::isString($options['KeyFrameInterval'], 'options[KeyFrameInterval]');
            $this->_keyFrameInterval = $options['KeyFrameInterval'];
        }

        if (isset($options['StreamingProtocol'])) {
            Validate::isString($options['StreamingProtocol'], 'options[StreamingProtocol]');
            $this->_streamingProtocol = $options['StreamingProtocol'];
        }

        if (isset($options['AccessControl'])) {
            Validate::isArray($options['AccessControl'], 'options[AccessControl]');
            $this->_accessControl = ChannelInputAccessControl::createFromOptions($options['AccessControl']);
        }

        if (!empty($options['Endpoints'])) {
            Validate::isArray($options['Endpoints'], 'options[Endpoints]');
            foreach ($options['Endpoints'] as $endpoint) {
                $this->_endpoints[] = ChannelEndpoint::createFromOptions($endpoint);
            }
        }
    }
368029
public function subscribeToTopic($topic, $registrationTokenOrTokens): array
    {
        $topic = $topic instanceof Topic ? $topic : Topic::fromValue($topic);
        $tokens = $this->ensureArrayOfRegistrationTokens($registrationTokenOrTokens);

        $response = $this->topicManagementApi->subscribeToTopic($topic, $tokens);

        return JSON::decode((string) $response->getBody(), true);
    }
482170
public function save(array &$a, array $options = [])
    {
        if ($this->eventManager->hasListeners(Events::preSave)) {
            $eventArgs = new EventArgs($this, $a, $options);
            $this->eventManager->dispatchEvent(Events::preSave, $eventArgs);
            $a = $eventArgs->getData();
            $options = $eventArgs->getOptions();
        }

        $result = $this->doSave($a, $options);

        if ($this->eventManager->hasListeners(Events::postSave)) {
            $eventArgs = new MutableEventArgs($this, $result);
            $this->eventManager->dispatchEvent(Events::postSave, $eventArgs);
            $result = $eventArgs->getData();
        }

        return $result;
    }
611842
public function deserialize($data, Message $message, array $options = [])
    {
        $message = $this->messageSerializer->deserialize($data, $message, $options);

        if (isset($data['subject'])) {
            $subject = $this->om->getObject($data['subject'], Subject::class);

            if (!empty($subject)) {
                $message->setSubject($subject);
            }
        }

        if (isset($data['parent'])) {
            $parent = $this->om->getRepository($this->getClass())->findOneBy(['uuid' => $data['parent']['id']]);

            if ($parent) {
                $message->setParent($parent);
            }
        }
        $this->sipe('meta.flagged', 'setFlagged', $data, $message);
        $this->sipe('meta.moderation', 'setModerated', $data, $message);

        return $message;
    }
1020823
def topics(options = {})
      batches = Enumerator.new do |y|
        resp = @client.list_topics(options)
        resp.each_page do |page|
          batch = []
          page.data.topics.each do |t|
            batch << Topic.new(
              arn: t.topic_arn,
              client: @client
            )
          end
          y.yield(batch)
        end
      end
      Topic::Collection.new(batches)
    end
476410
public function serialize(AnnouncementAggregate $announcements, array $options = [])
    {
        $announcePosts = $announcements->getAnnouncements()->toArray();

        if (in_array(static::VISIBLE_POSTS_ONLY, $options)) {
            // filter embed announces to only get visible ones
            $now = new \DateTime('now');
            $announcePosts = array_values(// reindex array for correct serialization
                array_filter($announcePosts, function (Announcement $announcement) use ($now) {
                    return $announcement->isVisible()
                        && (empty($announcement->getVisibleFrom()) || $announcement->getVisibleFrom() <= $now)
                        && (empty($announcement->getVisibleUntil()) || $announcement->getVisibleUntil() > $now);
                })
            );
        }

        return [
            'id' => $announcements->getUuid(),
            'posts' => array_map(function (Announcement $announcement) {
                return $this->announcementSerializer->serialize($announcement);
            }, $announcePosts),
        ];
    }
825237
public function publish($topic, $arguments = null, $argumentsKw = null, $options = null)
    {
        return $this->getClient()->getSession()
            ->publish($this->prefix . $topic, $arguments, $argumentsKw, $options);
    }
601107
public function publish(array $message, array $options = [])
    {
        $mergedOptions = $this->mergeOptions($options);

        if (isset($options['message_deduplication_id'])) {
            $mergedOptions['message_deduplication_id'] = $options['message_deduplication_id'];
        }

        if (isset($options['message_group_id'])) {
            $mergedOptions['message_group_id'] = $options['message_group_id'];
        }

        $options = $mergedOptions;

        $publishStart = microtime(true);

        // ensures that the SQS Queue and SNS Topic exist
        if (!$this->queueExists()) {
            $this->create();
        }

        if ($options['push_notifications']) {

            if (!$this->topicExists()) {
                $this->create();
            }

            $message    = [
                'default' => $this->getNameWithPrefix(),
                'sqs'     => json_encode($message),
                'http'    => $this->getNameWithPrefix(),
                'https'   => $this->getNameWithPrefix(),
            ];

            $result = $this->sns->publish([
                'TopicArn'         => $this->topicArn,
                'Subject'          => $this->getName(),
                'Message'          => json_encode($message),
                'MessageStructure' => 'json'
            ]);

            $context = [
                'TopicArn'           => $this->topicArn,
                'MessageId'          => $result->get('MessageId'),
                'push_notifications' => $options['push_notifications'],
                'publish_time'       => microtime(true) - $publishStart
            ];
            $this->log(200,"Message published to SNS", $context);

            return $result->get('MessageId');
        }

        $arguments = [
            'QueueUrl'      => $this->queueUrl,
            'MessageBody'   => json_encode($message),
            'DelaySeconds'  => $options['message_delay']
        ];

        if ($this->isQueueFIFO()) {
            if (isset($options['message_deduplication_id'])) {
                // Always use user supplied dedup id
                $arguments['MessageDeduplicationId'] = $options['message_deduplication_id'];
            } elseif ($options['content_based_deduplication'] !== true) {
                // If none is supplied and option "content_based_deduplication" is not set, generate default
                $arguments['MessageDeduplicationId'] = hash('sha256', json_encode($message));
            }

            $arguments['MessageGroupId'] = $this->getNameWithPrefix();
            if (isset($options['message_group_id'])) {
                $arguments['MessageGroupId'] = $options['message_group_id'];
            }
        }

        $result = $this->sqs->sendMessage($arguments);

        $context = [
            'QueueUrl'              => $this->queueUrl,
            'MessageId'             => $result->get('MessageId'),
            'push_notifications'    => $options['push_notifications'],
            'fifo'                  => $options['fifo']
        ];

        if ($this->isQueueFIFO()) {
            if (isset($arguments['MessageDeduplicationId'])) {
                $context['message_deduplication_id'] = $arguments['MessageDeduplicationId'];
            }
            $context['message_group_id'] = $arguments['MessageGroupId'];
        }

        $this->log(200,"Message published to SQS", $context);

        return $result->get('MessageId');
    }
906890
public function getUnreadNotifications(ComponentInterface $subject, $context = "GLOBAL", array $options = array())
    {
        $options['context'] = $context;
        $options['type']    = 'notification';

        return $this->timelineManager->getTimeline($subject, $options);
    }
1134075
public function publish(array $message, array $options = [])
    {
        $options      = $this->mergeOptions($options);
        $publishStart = microtime(true);

        if (!$this->queueExists()) {
            $this->create();
        }

        $result = $this->ironmq->postMessage(
            $this->getNameWithPrefix(),
            json_encode($message + ['_qpush_queue' => $this->name]),
            [
                'timeout'       => $options['message_timeout'],
                'delay'         => $options['message_delay'],
                'expires_in'    => $options['message_expiration']
            ]
        );

        $context = [
            'message_id'    => $result->id,
            'publish_time'  => microtime(true) - $publishStart
        ];
        $this->log(200, "Message has been published.", $context);

        return (int) $result->id;
    }
385841
private function deserializeTags(Subject $subject, array $tags = [], array $options = [])
    {
        //  if ($this->hasOption(Transfer::PERSIST_TAG, $options)) {
        $event = new GenericDataEvent([
            'tags' => $tags,
            'data' => [
                [
                    'class' => Subject::class,
                    'id' => $subject->getUuid(),
                    'name' => $subject->getTitle(),
                ],
            ],
            'replace' => true,
        ]);

        $this->eventDispatcher->dispatch('claroline_tag_multiple_data', $event);
        //}
    }
1293512
public function checkMessages(array $options = [])
    {
        $this->buildCall('/text');

        $rawMessages = $this->getRequest()->xml();

        return $this->makeMessages($rawMessages->Text);
    }
759869
public synchronized List<KafkaDependencyInstanceContext> getDepsByTopicAndEvent(final String topic,
      final String regex) {
    final Map<String, List<KafkaDependencyInstanceContext>> regexMap = this.topicEventMap.get(topic);
    if (regexMap != null) {
      return regexMap.get(regex);
    }
    return Collections.emptyList();
  }
671080
public function onPublish(ConnectionInterface $connection, Topic $topic, WampRequest $request, $event, array $exclude, array $eligible)
    {
        /*
            $topic->getId() will contain the FULL requested uri, so you can proceed based on that

            if ($topic->getId() == "chat/channel/shout") ????????????????¿¿¿¿¿¿¿¿¿¿¿¿¿¿
               //shout something to all subs.
        */

        $topic->broadcast([
            'msg' => $event
        ]);
    }