// BytesToTopic hashes an arbitrary length byte slice and truncates it to the length of a topic, using only the first bytes of the digest
961293
func BytesToTopic(b []byte) Topic {
	topicHashMutex.Lock()
	defer topicHashMutex.Unlock()
	topicHashFunc.Reset()
	topicHashFunc.Write(b)
	return Topic(whisper.BytesToTopic(topicHashFunc.Sum(nil)))
}
345436
func Digest(bundleHash []int8, signatureFragment Trits, startOffset int, spongeFunc ...SpongeFunction) (Trits, error) {
	sigLength := len(signatureFragment)

	secLvl, err := GetSecurityLevel(bundleHash)
	if err != nil {
		return nil, err
	}

	sigFrag := make(Trits, sigLength)
	copy(sigFrag, signatureFragment[:sigLength])

	h := GetSpongeFunc(spongeFunc, kerl.NewKerl)
	defer h.Reset()

	dig := make(Trits, HashTrinarySize)
	sig := make(Trits, HashTrinarySize)

	i := startOffset % int(secLvl)
	for digLength := 0; digLength < sigLength; {
		for j := i * ISSChunkLength; j < (i+1)*ISSChunkLength && digLength < sigLength; j += TrinaryRadix {
			copy(sig, sigFrag[digLength:digLength+HashTrinarySize])

			to := (bundleHash[j] + bundleHash[j+1]*3 + bundleHash[j+2]*9) - MinTryteValue

			for k := 0; k < int(to); k++ {
				err := h.Absorb(sig)
				if err != nil {
					return nil, err
				}
				sig, err = h.Squeeze(HashTrinarySize)
				if err != nil {
					return nil, err
				}
				h.Reset()
			}

			copy(sigFrag[digLength:], sig)
			digLength += HashTrinarySize
		}

		i = (i + 1) % int(secLvl)
	}

	err = h.Absorb(sigFrag)
	if err != nil {
		return nil, err
	}

	dig, err = h.Squeeze(HashTrinarySize)
	if err != nil {
		return nil, err
	}

	return dig, nil
}
151491
func (kz *Kazoo) CreateTopic(name string, partitionCount int, replicationFactor int, topicConfig map[string]string) error {
	topic := kz.Topic(name)

	// Official kafka sdk checks if topic exists, then always writes the config unconditionally
	// but only writes the partition map if ones does not exist.
	exists, err := topic.Exists()
	if err != nil {
		return err
	} else if exists {
		return ErrTopicExists
	}

	brokerList, err := kz.brokerIDList()
	if err != nil {
		return err
	}

	partitionList, err := topic.generatePartitionAssignments(brokerList, partitionCount, replicationFactor)
	if err != nil {
		return err
	}

	configData, err := topic.marshalConfig(topicConfig)
	if err != nil {
		return err
	}

	partitionData, err := topic.marshalPartitions(partitionList)
	if err != nil {
		return err
	}

	if err = kz.createOrUpdate(topic.configPath(), configData, false); err != nil {
		return err
	}

	if err = kz.create(topic.metadataPath(), partitionData, false); err != nil {
		return err
	}

	return nil
}
1080566
func generateSubToTopicUrl(instaceId string, topic string) string {
	Tmptopic := strings.ToLower(topic)
	if strings.Contains(Tmptopic, "/topics/") {
		tmp := strings.Split(topic, "/")
		topic = tmp[len(tmp)-1]
	}
	return fmt.Sprintf(subscribe_instanceid_to_topic_srv_url, instaceId, topic)
}
500308
func HashTo(v Value, p []byte) {
	h := hashPool.Get().(hash.Hash)
	h.Reset()
	defer hashPool.Put(h)
	if len(p) < HashSize {
		panic("buffer too small to fit the hash")
	}
	if v != nil {
		// TODO(kortschak,dennwc) Remove dependence on String() method.
		h.Write([]byte(v.String()))
	}
	h.Sum(p[:0])
}
373354
func (s *PublishInput) SetTopic(v string) *PublishInput {
	s.Topic = &v
	return s
}
1360181
func (res *MCResponse) Bytes() []byte {
	data := make([]byte, res.Size())

	pos := res.fillHeaderBytes(data)

	copy(data[pos:pos+len(res.Body)], res.Body)

	return data
}
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,
	})
}
929382
func (b *HttpRequest) Bytes() ([]byte, error) {
	if b.body != nil {
		return b.body, nil
	}
	resp, err := b.getResponse()
	if err != nil {
		return nil, err
	}
	if resp.Body == nil {
		return nil, nil
	}
	defer resp.Body.Close()
	if b.setting.Gzip && resp.Header.Get("Content-Encoding") == "gzip" {
		reader, err := gzip.NewReader(resp.Body)
		if err != nil {
			return nil, err
		}
		b.body, err = ioutil.ReadAll(reader)
	} else {
		b.body, err = ioutil.ReadAll(resp.Body)
	}
	return b.body, err
}
526776
func (p *Publisher) SubscribeTopic(topic topicFunc) chan interface{} {
	ch := make(chan interface{}, p.buffer)
	p.m.Lock()
	p.subscribers[ch] = topic
	p.m.Unlock()
	return ch
}
601067
func (s *ListTopicRulesInput) SetTopic(v string) *ListTopicRulesInput {
	s.Topic = &v
	return s
}
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
}
684477
func (fcm *FCM) SendToTopic(ctx context.Context, notificationKey string, payload Message) (*Response, error) {
	return fcm.SendToDevice(ctx, normalizeTopicName(notificationKey), payload)
}
1352417
func (t *Topics) Append(topics *Hashes) {
	t.topics = append(t.topics, topics.hashes)
}
1032085
func updateTopics(query, doc utils.M) error {
	return db.Conn.Update(db.TopicCollection, query, doc)
}
952147
func fixAndHashPayload(g *GlobalContext, payload []byte, linkID LinkID) []byte {
	toHash := payload
	if s, ok := badWhitespaceChainLinks[linkID.Export()]; ok {
		last := len(payload) - 1
		if payload[last] == '\n' {
			g.Log.Debug("Fixing payload hash by stripping newline on link '%s': %s", linkID.Export(), s)
			toHash = payload[0:last]
		}
	}
	ret := sha256.Sum256(toHash)
	return ret[:]
}
373484
func findMembersByTopic(members []memberGroupMetadata) map[string][]memberGroupMetadata {
	membersByTopic := map[string][]memberGroupMetadata{}
	for _, member := range members {
		for _, topic := range member.Metadata.Topics {
			membersByTopic[topic] = append(membersByTopic[topic], member)
		}
	}

	// normalize ordering of members to enabling grouping across topics by partitions
	//
	// Want:
	// 		C0 [T0/P0, T1/P0]
	// 		C1 [T0/P1, T1/P1]
	//
	// Not:
	// 		C0 [T0/P0, T1/P1]
	// 		C1 [T0/P1, T1/P0]
	//
	// Even though the later is still round robin, the partitions are crossed
	//
	for _, members := range membersByTopic {
		sort.Slice(members, func(i, j int) bool {
			return members[i].MemberID < members[j].MemberID
		})
	}

	return membersByTopic
}