// Topic returns a new listener for Topic messages
172798
func Topic(pattern string, fn func(res *Response) error) handler {
	return &Handler{Method: TOPIC, Run: fn}
}
198170
func (s Subject) String() string {
	var comp []string
	if len(s.Country) > 0 {
		comp = append(comp, "C="+strings.Join(s.Country, ", C="))
	}
	if len(s.Organisation) > 0 {
		comp = append(comp, "O="+strings.Join(s.Organisation, ", O="))
	}
	if len(s.OrgUnit) > 0 {
		comp = append(comp, "OU="+strings.Join(s.OrgUnit, ", OU="))
	}
	if len(s.CommonName) > 0 {
		comp = append(comp, "CN="+s.CommonName)
	}
	return strings.Join(comp, ", ")
}
556145
func (tl TopicList) Find(name string) *Topic {
	for _, topic := range tl {
		if topic.Name == name {
			return topic
		}
	}
	return nil
}
671370
func (b *pubsubBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
	t := b.client.Topic(topic)
	ctx := context.Background()

	exists, err := t.Exists(ctx)
	if err != nil {
		return err
	}

	if !exists {
		tt, err := b.client.CreateTopic(ctx, topic)
		if err != nil {
			return err
		}
		t = tt
	}

	m := &pubsub.Message{
		ID:         "m-" + uuid.New().String(),
		Data:       msg.Body,
		Attributes: msg.Header,
	}

	pr := t.Publish(ctx, m)
	_, err = pr.Get(ctx)
	return err
}
309117
func (t *Tree) String() string {
	t.mutex.RLock()
	defer t.mutex.RUnlock()

	return fmt.Sprintf("topic.Tree:%s", t.root.string(0))
}
1344643
func (t *Tree) Search(topic string) []interface{} {
	t.mutex.RLock()
	defer t.mutex.RUnlock()

	segments := strings.Split(topic, t.Separator)
	values := t.search([]interface{}{}, 0, segments, t.root)

	return t.clean(values)
}
51514
func (m *Model) topic(suffix string) string {
	return modelTopic(m.details.ModelUUID, suffix)
}
920344
func (t *Topic) Config() (map[string]string, error) {
	value, _, err := t.kz.conn.Get(t.configPath())
	if err != nil {
		return nil, err
	}

	return t.parseConfig(value)
}
838937
func (t *Topics) Get(index int) (hashes *Hashes, _ error) {
	if index < 0 || index >= len(t.topics) {
		return nil, errors.New("index out of bounds")
	}
	return &Hashes{t.topics[index]}, nil
}
1095901
func (t *Tree) Empty(topic string) {
	t.mutex.Lock()
	defer t.mutex.Unlock()

	t.remove(nil, 0, strings.Split(topic, t.Separator), t.root)
}
1295182
func Register(topic string, channel string) *Command {
	params := [][]byte{[]byte(topic)}
	if len(channel) > 0 {
		params = append(params, []byte(channel))
	}
	return &Command{[]byte("REGISTER"), params, nil}
}
426063
func (c *Client) Publish(topic string, args []interface{}, kwargs map[string]interface{}) error {
	return c.Send(&Publish{
		Request:     NewID(),
		Options:     make(map[string]interface{}),
		Topic:       URI(topic),
		Arguments:   args,
		ArgumentsKw: kwargs,
	})
}
1358404
func (cli *Client) handleMessage(topicName, message []byte) {
	// Get the string of the Topic Name.
	topicNameStr := string(topicName)

	for topicFilter, handler := range cli.conn.ackedSubs {
		if handler == nil || !match(topicNameStr, topicFilter) {
			continue
		}

		// Execute the handler.
		go handler(topicName, message)
	}
}
53819
func (t ApplicationTopic) String() string {
	appID := simpleWildcard
	if t.AppID != "" {
		appID = t.AppID
	}
	if t.Type == AppEvents && t.Field == "" {
		t.Field = wildcard
	}
	topic := fmt.Sprintf("%s/%s", appID, t.Type)
	if t.Type == AppEvents && t.Field != "" {
		topic += "/" + t.Field
	}
	return topic
}
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
}
362850
func (c *Client) Listener(kind ListenerType, event string) *Listener {
	listener := &Listener{
		Type:      kind,
		Event:     event,
		Messages:  make(chan redis.Message),
		PMessages: make(chan redis.PMessage),
		Client:    c,
	}
	c.Subscribe(listener)

	return listener
}
329650
func (h *grpcHandler) Publish(ctx context.Context, msg *proto.Message) (*proto.Empty, error) {
	if len(msg.Topic) == 0 {
		return nil, merr.InternalServerError("go.micro.broker", "Topic not found")
	}

	m := &broker.Message{
		Header: msg.Header,
		Body:   msg.Body,
	}

	p := &grpcPublication{m: m, t: msg.Topic}

	h.g.RLock()
	for _, subscriber := range h.g.subscribers[msg.Topic] {
		if msg.Id == subscriber.id {
			// sub is sync; crufty rate limiting
			// so we don't hose the cpu
			subscriber.fn(p)
		}
	}
	h.g.RUnlock()
	return new(proto.Empty), nil
}