// AddTopic adds a given topic config to the general bucket notification config
491807
func (b *BucketNotification) AddTopic(topicConfig NotificationConfig) bool {
	newTopicConfig := TopicConfig{NotificationConfig: topicConfig, Topic: topicConfig.Arn.String()}
	for _, n := range b.TopicConfigs {
		// If new config matches existing one
		if n.Topic == newTopicConfig.Arn.String() && newTopicConfig.Filter == n.Filter {

			existingConfig := set.NewStringSet()
			for _, v := range n.Events {
				existingConfig.Add(string(v))
			}

			newConfig := set.NewStringSet()
			for _, v := range topicConfig.Events {
				newConfig.Add(string(v))
			}

			if !newConfig.Intersection(existingConfig).IsEmpty() {
				return false
			}
		}
	}
	b.TopicConfigs = append(b.TopicConfigs, newTopicConfig)
	return true
}
1328828
func (fcm *FCM) SubscribeDevicesToTopic(ctx context.Context, registrationTokens []string, topic string) (*Response, error) {
	return fcm.sendFirebaseTopicRequest(ctx, fcmTopicAddEndpoint, Topic{To: normalizeTopicName(topic), RegistrationTokens: registrationTokens})
}
489861
func (r *RoomService) SetTopic(id string, topic string) (*http.Response, error) {
	topicReq := &SetTopicRequest{Topic: topic}

	req, err := r.client.NewRequest("PUT", fmt.Sprintf("room/%s/topic", id), nil, topicReq)
	if err != nil {
		return nil, err
	}

	return r.client.Do(req, nil)
}
199041
func (s *RegisterEventTopicInput) SetTopicName(v string) *RegisterEventTopicInput {
	s.TopicName = &v
	return s
}
209024
func (kz *Kazoo) DeleteTopic(name string) error {
	node := fmt.Sprintf("%s/admin/delete_topics/%s", kz.conf.Chroot, name)

	exists, err := kz.exists(node)
	if err != nil {
		return err
	}
	if exists {
		return ErrTopicMarkedForDelete
	}

	if err := kz.create(node, nil, false); err != nil {
		return err
	}
	return nil
}
1161329
func (n *NSQD) GetTopic(topicName string) *Topic {
	// most likely, we already have this topic, so try read lock first.
	n.RLock()
	t, ok := n.topicMap[topicName]
	n.RUnlock()
	if ok {
		return t
	}

	n.Lock()

	t, ok = n.topicMap[topicName]
	if ok {
		n.Unlock()
		return t
	}
	deleteCallback := func(t *Topic) {
		n.DeleteExistingTopic(t.name)
	}
	t = NewTopic(topicName, &context{n}, deleteCallback)
	n.topicMap[topicName] = t

	n.Unlock()

	n.logf(LOG_INFO, "TOPIC(%s): created", t.name)
	// topic is created but messagePump not yet started

	// if loading metadata at startup, no lookupd connections yet, topic started after load
	if atomic.LoadInt32(&n.isLoading) == 1 {
		return t
	}

	// if using lookupd, make a blocking call to get the topics, and immediately create them.
	// this makes sure that any message received is buffered to the right channels
	lookupdHTTPAddrs := n.lookupdHTTPAddrs()
	if len(lookupdHTTPAddrs) > 0 {
		channelNames, err := n.ci.GetLookupdTopicChannels(t.name, lookupdHTTPAddrs)
		if err != nil {
			n.logf(LOG_WARN, "failed to query nsqlookupd for channels to pre-create for topic %s - %s", t.name, err)
		}
		for _, channelName := range channelNames {
			if strings.HasSuffix(channelName, "#ephemeral") {
				continue // do not create ephemeral channel with no consumer client
			}
			t.GetChannel(channelName)
		}
	} else if len(n.getOpts().NSQLookupdTCPAddresses) > 0 {
		n.logf(LOG_ERROR, "no available nsqlookupd to query for channels to pre-create for topic %s", t.name)
	}

	// now that all channels are added, start topic messagePump
	t.Start()
	return t
}
684477
func (fcm *FCM) SendToTopic(ctx context.Context, notificationKey string, payload Message) (*Response, error) {
	return fcm.SendToDevice(ctx, normalizeTopicName(notificationKey), payload)
}
444324
func NewTopic(name string, config TopicConfig, client *Client) *Topic {
	return &Topic{
		handle: C.rd_kafka_topic_new(client.handle, C.CString(name), config.handle),
		client: client,
		name:   name,
	}
}
936734
func ProtocolTopic(spec *protocols.Spec) Topic {
	return BytesToTopic([]byte(fmt.Sprintf("%s:%d", spec.Name, spec.Version)))
}
217592
func (b *Broker) CreateTopics(request *CreateTopicsRequest) (*CreateTopicsResponse, error) {
	response := new(CreateTopicsResponse)

	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}

	return response, nil
}
452600
func ensureTopic(pubsubClient *pubsub.Client, topicName string) *pubsub.Topic {
	var topic *pubsub.Topic
	ctx, _ := context.WithTimeout(context.Background(), ContextDuration)
	topic = pubsubClient.Topic(topicName)
	topicExists, err := topic.Exists(ctx)

	if err != nil {
		log.Fatalf("Could not check if topic exists: %v", err)
	}
	if !topicExists {
		new_topic, err := pubsubClient.CreateTopic(ctx, topicName)
		if err != nil {
			log.Fatalf("Could not create PubSub topic: %v", err)
		}
		topic = new_topic
	}

	return topic
}
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
}
515239
func (s *WorkmailAction) SetTopicArn(v string) *WorkmailAction {
	s.TopicArn = &v
	return s
}
190875
func (b *Broker) CreateTopic(topics []proto.TopicInfo, timeout time.Duration, validateOnly bool) (*proto.CreateTopicsResp, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	var resp *proto.CreateTopicsResp
	err := b.callOnClusterController(func(c *connection) error {
		var err error
		req := proto.CreateTopicsReq{
			Timeout:              timeout,
			CreateTopicsRequests: topics,
			ValidateOnly:         validateOnly,
		}
		resp, err = c.CreateTopic(&req)
		return err
	})
	return resp, err
}
887055
func (c *Client) ListTopicPermissions() (rec []TopicPermissionInfo, err error) {
	req, err := newGETRequest(c, "topic-permissions/")
	if err != nil {
		return []TopicPermissionInfo{}, err
	}

	if err = executeAndParseRequest(c, req, &rec); err != nil {
		return []TopicPermissionInfo{}, err
	}

	return rec, nil
}
1256845
func (pn *PushNotification) AddPayload(p *Payload) {
	// This deserves some explanation.
	//
	// Setting an exported field of type int to 0
	// triggers the omitempty behavior if you've set it.
	// Since the badge is optional, we should omit it if
	// it's not set. However, we want to include it if the
	// value is 0, so there's a hack in push_notification.go
	// that exploits the fact that Apple treats -1 for a
	// badge value as though it were 0 (i.e. it clears the
	// badge but doesn't stop the notification from going
	// through successfully.)
	//
	// Still a hack though :)
	if p.Badge == 0 {
		p.Badge = -1
	}
	pn.Set("aps", p)
}
1286186
func (c *s3Client) AddNotificationConfig(arn string, events []string, prefix, suffix string) *probe.Error {
	bucket, _ := c.url2BucketAndObject()
	// Validate total fields in ARN.
	fields := strings.Split(arn, ":")
	if len(fields) != 6 {
		return errInvalidArgument()
	}

	// Get any enabled notification.
	mb, e := c.api.GetBucketNotification(bucket)
	if e != nil {
		return probe.NewError(e)
	}

	accountArn := minio.NewArn(fields[1], fields[2], fields[3], fields[4], fields[5])
	nc := minio.NewNotificationConfig(accountArn)

	// Configure events
	for _, event := range events {
		switch event {
		case "put":
			nc.AddEvents(minio.ObjectCreatedAll)
		case "delete":
			nc.AddEvents(minio.ObjectRemovedAll)
		case "get":
			nc.AddEvents(minio.ObjectAccessedAll)
		default:
			return errInvalidArgument().Trace(events...)
		}
	}
	if prefix != "" {
		nc.AddFilterPrefix(prefix)
	}
	if suffix != "" {
		nc.AddFilterSuffix(suffix)
	}

	switch fields[2] {
	case "sns":
		if !mb.AddTopic(nc) {
			return errInvalidArgument().Trace("Overlapping Topic configs")
		}
	case "sqs":
		if !mb.AddQueue(nc) {
			return errInvalidArgument().Trace("Overlapping Queue configs")
		}
	case "lambda":
		if !mb.AddLambda(nc) {
			return errInvalidArgument().Trace("Overlapping lambda configs")
		}
	default:
		return errInvalidArgument().Trace(fields[2])
	}

	// Set the new bucket configuration
	if err := c.api.SetBucketNotification(bucket, mb); err != nil {
		return probe.NewError(err)
	}
	return nil
}