Subscribes to a topic.

@param consumer
@param topic
@since 1.3.2
9183
private void _checkAndSubscribe(KafkaConsumer<?, ?> consumer, String topic) {
        synchronized (consumer) {
            Set<String> subscription = consumer.subscription();
            if (subscription == null || !subscription.contains(topic)) {
                consumer.subscribe(Arrays.asList(topic));
            }
        }
    }
358980
protected Collection loadEvenHandler(String topic) {
		Collection ehs = new ArrayList();
		Collection<String> consumers = (Collection<String>) containerWrapper.lookup(ConsumerLoader.TOPICNAME + topic);
		if (consumers == null || consumers.size() == 0) {
			Debug.logWarning("[Jdonframework]there is no any consumer class annotated with @Consumer(" + topic + ") ", module);
			return ehs;
		}
		for (String consumerName : consumers) {
			DomainEventHandler eh = (DomainEventHandler) containerWrapper.lookup(consumerName);
			ehs.add(eh);
		}

		return ehs;

	}
728023
synchronized void add(ConsumerSessionImpl consumer)
  {
    consumer.setId(_consumerCount);
    
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "add", new Long(consumer.getIdInternal()) );

    _consumers.put(new Long(_consumerCount), consumer);

    _consumerCount++;

    if (tc.isEntryEnabled())
      SibTr.exit(tc, "add");
  }
1311462
def community_topic_subscription_create(self, topic_id, data, **kwargs):
        ""
        api_path = "/api/v2/community/topics/{topic_id}/subscriptions.json"
        api_path = api_path.format(topic_id=topic_id)
        return self.call(api_path, method="POST", data=data, **kwargs)
151556
@JsTopic
	@WsDataService
	public Integer subscribe(@JsTopicName(prefix = Constants.Topic.SUBSCRIBERS) String topic, Session session) throws IllegalAccessException {
		return topicManager.registerTopicSession(topic, session);
	}
1039357
public boolean seekToBeginning(String consumerGroupId, String topic) {
        KafkaMsgConsumer consumer = getKafkaConsumer(consumerGroupId, false);
        return consumer.seekToBeginning(topic);
    }
262106
@Override
	public void registerConsumer(LocalMessageConsumer consumer)
    {
        super.registerConsumer(consumer);
        
        subscriptionsLock.writeLock().lock();
        try
		{
        	LocalTopicSubscription subscription = subscriptionMap.remove(consumer.getSubscriberId());
        	if (subscription == null)
        	{
        		// New subscription
        		storeNewSubscription(consumer);
        	}
        	else
        	{
        		// Remove old subscription
        		removeSubscription(subscription);
        		// Replace by new subscription
        		storeNewSubscription(consumer);
        	}
		}
        finally
        {
        	subscriptionsLock.writeLock().unlock();
        }
    }
905949
public QueueConsumer createQueueListener(final String topic, final ConsumerCallback<Message> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new QueueConsumer(connectionFactory, jmsConfig, topic, messageCallback);
    }
504149
public void tellSubscribers( Actor sender, String topic, Object message) {
        List<ReceiverActor> subscriber = topic2Subscriber.get(topic);
        if ( subscriber != null ) {
            subscriber.stream()
                .filter(subs -> !subs.equals(sender)) // do not receive self sent
                .forEach(subs -> subs.receiveTell(topic, message) );
        }
    }
972769
public <T> TopicProducer<T> createTopicProducer(final String topic, final ProducerCallback<T> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new TopicProducer<T>(connectionFactory, jmsConfig, topic, messageCallback);
    }
1147832
public function onSubscribe(ConnectionInterface $connection, Topic $topic, WampRequest $request)
    {
        //this will broadcast the message to ALL subscribers of this topic.
        $topic->broadcast(['msg' => $connection->resourceId . " has joined " . $topic->getId()]);
    }
293951
@Override
	public TopicSubscriber createSubscriber(Topic topic) throws JMSException
    {
        return createSubscriber(topic,null,false);
    }
434446
protected void sendUnsubscribe(String topic) {
    JsonObject msg = Json.createObject().set(TYPE, "unregister").set(TOPIC, topic);
    send(msg);
  }
412718
public TopicConsumer createTopicTextMessageListener(final String topic, final ConsumerCallback<String> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new TopicConsumer(connectionFactory, jmsConfig, topic, new TextMessageConsumerCallback(messageCallback));
    }
225820
def consume(topic, conf):
    
    from confluent_kafka.avro import AvroConsumer
    from confluent_kafka.avro.serializer import SerializerError

    print("Consuming user records from topic {} with group {}. ^c to exit.".format(topic, conf["group.id"]))

    c = AvroConsumer(conf, reader_value_schema=record_schema)
    c.subscribe([topic])

    while True:
        try:
            msg = c.poll(1)

            # There were no messages on the queue, continue polling
            if msg is None:
                continue

            if msg.error():
                print("Consumer error: {}".format(msg.error()))
                continue

            record = User(msg.value())
            print("name: {}\n\tfavorite_number: {}\n\tfavorite_color: {}\n".format(
                record.name, record.favorite_number, record.favorite_color))
        except SerializerError as e:
            # Report malformed record, discard results, continue polling
            print("Message deserialization failed {}".format(e))
            continue
        except KeyboardInterrupt:
            break

    print("Shutting down consumer..")
    c.close()
632013
function subscribe() {
  console.log('Subscribing client ' + clientId + ' to topic: ' + topics.subscribe);
  client.subscribe(topics.subscribe);
  isSubscribed = true;
}
742376
def succeeded(topic, event)
      subscribers_for(topic).each{ |subscriber| subscriber.succeeded(event) }
    end