// SubscribeToTopic subscribes a single device/token to a topic
171338
func (this *FcmClient) SubscribeToTopic(instanceIdToken string, topic string) (*SubscribeResponse, error) {

	request, err := http.NewRequest("POST", generateSubToTopicUrl(instanceIdToken, topic), nil)
	request.Header.Set("Authorization", this.apiKeyHeader())
	request.Header.Set("Content-Type", "application/json")

	if err != nil {
		return nil, err
	}

	client := &http.Client{}
	response, err := client.Do(request)
	if err != nil {
		return nil, err
	}

	defer response.Body.Close()

	body, err := ioutil.ReadAll(response.Body)
	if err != nil {
		return nil, err
	}

	subResponse, err := parseSubscribeResponse(body, response)
	if err != nil {
		return nil, err
	}

	return subResponse, nil
}
274577
func Subscribe(topic string, channel string) *Command {
	var params = [][]byte{[]byte(topic), []byte(channel)}
	return &Command{[]byte("SUB"), params, nil}
}
590810
func (api *Client) SetChannelTopicContext(ctx context.Context, channelID, topic string) (string, error) {
	values := url.Values{
		"token":   {api.token},
		"channel": {channelID},
		"topic":   {topic},
	}

	response, err := api.channelRequest(ctx, "channels.setTopic", values)
	if err != nil {
		return "", err
	}
	return response.Topic, nil
}
945076
func (api *Client) SetGroupTopic(group, topic string) (string, error) {
	return api.SetGroupTopicContext(context.Background(), group, topic)
}
1081174
func (c *DefaultClient) SubscribeDeviceDownlink(appID string, devID string, handler DownlinkHandler) Token {
	topic := DeviceTopic{appID, devID, DeviceDownlink, ""}
	return c.subscribe(topic.String(), func(mqtt MQTT.Client, msg MQTT.Message) {
		// Determine the actual topic
		topic, err := ParseDeviceTopic(msg.Topic())
		if err != nil {
			c.ctx.Warnf("mqtt: received message on invalid downlink topic: %s", msg.Topic())
			return
		}

		// Unmarshal the payload
		dataDown := &types.DownlinkMessage{}
		err = json.Unmarshal(msg.Payload(), dataDown)
		if err != nil {
			c.ctx.Warnf("mqtt: could not unmarshal downlink: %s", err)
			return
		}
		dataDown.AppID = topic.AppID
		dataDown.DevID = topic.DevID

		// Call the Downlink handler
		handler(c, topic.AppID, topic.DevID, *dataDown)
	})
}
1196681
func (c *Client) SubscribeMultiple(filters map[string]byte, callback MessageHandler) Token {
	var err error
	token := newToken(packets.Subscribe).(*SubscribeToken)
	DEBUG.Println(CLI, "enter SubscribeMultiple")
	if !c.IsConnected() {
		token.err = ErrNotConnected
		token.flowComplete()
		return token
	}
	sub := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
	if sub.Topics, sub.Qoss, err = validateSubscribeMap(filters); err != nil {
		token.err = err
		return token
	}

	if callback != nil {
		for topic := range filters {
			c.msgRouter.addRoute(topic, callback)
		}
	}
	token.subs = make([]string, len(sub.Topics))
	copy(token.subs, sub.Topics)
	c.oboundP <- &PacketAndToken{p: sub, t: token}
	DEBUG.Println(CLI, "exit SubscribeMultiple")
	return token
}
627659
func (tm *topicManager) IsSubscribed(topic string) bool {
	tm.locker.RLock()
	defer tm.locker.RUnlock()
	return tm.allTopics[topic] != nil
}
373354
func (s *PublishInput) SetTopic(v string) *PublishInput {
	s.Topic = &v
	return s
}
812470
func BytesToTopic(b []byte) (t TopicType) {
	sz := TopicLength
	if x := len(b); x < TopicLength {
		sz = x
	}
	for i := 0; i < sz; i++ {
		t[i] = b[i]
	}
	return t
}
247956
func (br *defaultBroker) Subscribe(sub Sender, msg *Subscribe) {
	if _, ok := br.routes[msg.Topic]; !ok {
		br.routes[msg.Topic] = make(map[ID]Sender)
	}
	id := NewID()
	br.routes[msg.Topic][id] = sub

	br.subscriptions[id] = msg.Topic

	sub.Send(&Subscribed{Request: msg.Request, Subscription: id})
}
1373924
func (c *Client) Subscribe(topic string, qos byte, callback MessageHandler) Token {
	token := newToken(packets.Subscribe).(*SubscribeToken)
	DEBUG.Println(CLI, "enter Subscribe")
	if !c.IsConnected() {
		token.err = ErrNotConnected
		token.flowComplete()
		return token
	}
	sub := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
	if err := validateTopicAndQos(topic, qos); err != nil {
		token.err = err
		return token
	}
	sub.Topics = append(sub.Topics, topic)
	sub.Qoss = append(sub.Qoss, qos)
	DEBUG.Println(sub.String())

	if callback != nil {
		c.msgRouter.addRoute(topic, callback)
	}

	token.subs = append(token.subs, topic)
	c.oboundP <- &PacketAndToken{p: sub, t: token}
	DEBUG.Println(CLI, "exit Subscribe")
	return token
}
71340
func (conn *ProtoConnection) ConsumeTopic(msgClb func(messaging.ProtoMessage), topics ...string) error {
	conn.multiplexer.rwlock.Lock()
	defer conn.multiplexer.rwlock.Unlock()

	if conn.multiplexer.started {
		return fmt.Errorf("ConsumeTopic can be called only if the multiplexer has not been started yet")
	}

	byteClb := func(bm *client.ConsumerMessage) {
		pm := client.NewProtoConsumerMessage(bm, conn.serializer)
		msgClb(pm)
	}

	for _, topic := range topics {
		// check if we have already consumed the topic
		var found bool
		var subs *consumerSubscription
	LoopSubs:
		for _, subscription := range conn.multiplexer.mapping {
			if subscription.manual == true {
				// do not mix dynamic and manual mode
				continue
			}
			if subscription.topic == topic {
				found = true
				subs = subscription
				break LoopSubs
			}
		}

		if !found {
			subs = &consumerSubscription{
				manual:         false, // non-manual example
				topic:          topic,
				connectionName: conn.name,
				byteConsMsg:    byteClb,
			}
			// subscribe new topic
			conn.multiplexer.mapping = append(conn.multiplexer.mapping, subs)
		}

		// add subscription to consumerList
		subs.byteConsMsg = byteClb
	}

	return nil
}
1368405
func (s *ChangesService) SetTopic(changeID string, input *TopicInput) (*string, *Response, error) {
	u := fmt.Sprintf("changes/%s/topic", changeID)

	req, err := s.client.NewRequest("PUT", u, input)
	if err != nil {
		return nil, nil, err
	}

	v := new(string)
	resp, err := s.client.Do(req, v)
	if err != nil {
		return nil, resp, err
	}

	return v, resp, err
}
1176984
func ParseApplicationTopic(topic string) (*ApplicationTopic, error) {
	pattern := regexp.MustCompile("^([0-9a-z](?:[_-]?[0-9a-z]){1,35}|\\+)/(events)([0-9a-z/-]+|/#)?$")
	matches := pattern.FindStringSubmatch(topic)
	if len(matches) < 3 {
		return nil, fmt.Errorf("Invalid topic format")
	}
	var appID string
	if matches[1] != simpleWildcard {
		appID = matches[1]
	}
	topicType := ApplicationTopicType(matches[2])
	appTopic := &ApplicationTopic{appID, topicType, ""}
	if topicType == AppEvents && len(matches) > 3 {
		appTopic.Field = strings.Trim(matches[3], "/")
	}
	return appTopic, nil
}
990764
func (em *Emitter) Subscribe(ctx context.Context, subscriber string, queryable query.Queryable, bufferSize int) (<-chan interface{}, error) {
	qry, err := queryable.Query()
	if err != nil {
		return nil, err
	}
	return em.pubsubServer.Subscribe(ctx, subscriber, qry, bufferSize)
}
326004
func (ctl *taskController) PrepareTopic(ctx context.Context, publisher string) (topic string, token string, err error) {
	return ctl.eng.prepareTopic(ctx, &topicParams{
		jobID:     ctl.JobID(),
		invID:     ctl.InvocationID(),
		manager:   ctl.manager,
		publisher: publisher,
	})
}
1227369
func (s *ticketStore) addTopic(topic Topic, register bool) {
	log.Trace("Adding discovery topic", "topic", topic, "register", register)
	if s.radius[topic] == nil {
		s.radius[topic] = newTopicRadius(topic)
	}
	if register && s.tickets[topic] == nil {
		s.tickets[topic] = &topicTickets{buckets: make(map[timeBucket][]ticketRef)}
	}
}