// Publish creates a new Command to write a message to a given topic
761403
func Publish(topic string, body []byte) *Command {
	var params = [][]byte{[]byte(topic)}
	return &Command{[]byte("PUB"), params, body}
}
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
}
1077977
func (c *confirms) Publish() uint64 {
	c.m.Lock()
	defer c.m.Unlock()

	c.published++
	return c.published
}
292902
func (c FakeClient) Publish(id *charm.URL, channels []params.Channel, resources map[string]int) error {
	return c.repo.Publish(id, channels, resources)
}
237197
func Publish(topic string, body []byte) error {
	return DefaultProducer.Publish(topic, body)
}
29156
func (r Repository) Publish(id *charm.URL, channels []params.Channel, resources map[string]int) error {
	for _, channel := range channels {
		published := r.published[channel]
		published.Add(id.String())
		r.published[channel] = published
	}
	return nil
}
74126
func (p *NatsPublisher) Publish(subject string, r io.Reader) error {
	_, err := p.server.initSubject(subject)
	p.server.queue[subject] <- r
	return err
}
429457
func (em *Emitter) Publish(ctx context.Context, message interface{}, tags query.Tagged) error {
	if em == nil || em.pubsubServer == nil {
		return nil
	}
	return em.pubsubServer.PublishWithTags(ctx, message, tags)
}
817976
func (conn *Connection) Publish(msg *Message) error {
	transport, err := msg.Transport()
	if err != nil {
		return fmt.Errorf("Cannot publish Message %s: %s", msg.RequestID, err)
	}

	transport.RecordNetworkHop(conn.ConnectedServer(), conn.choria.Config.Identity, conn.ConnectedServer())

	if msg.CustomTarget != "" {
		return conn.publishConnectedBroadcast(msg, transport)
	}

	if conn.choria.IsFederated() {
		return conn.publishFederated(msg, transport)
	}

	return conn.publishConnected(msg, transport)
}
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
}
489298
func publish(pcc *PubControlClient, channel string, item *Item) error {
	export, err := item.Export()
	if err != nil {
		return err
	}
	export["channel"] = channel
	uri := ""
	auth := ""
	pcc.lock.Lock()
	uri = pcc.uri
	auth, err = pcc.generateAuthHeader()
	pcc.lock.Unlock()
	if err != nil {
		return err
	}
	err = pcc.pubCall(pcc, uri, auth, [](map[string]interface{}){export})
	if err != nil {
		return err
	}
	return nil
}
1148357
func (c *Client) Publish(channel string, message interface{}) (int64, error) {
	var ret int64

	err := c.command(
		&ret,
		[]byte("PUBLISH"),
		[]byte(channel),
		to.Bytes(message),
	)

	return ret, err
}
114305
func (b *sqsBroker) Publish(queueName string, msg *broker.Message, opts ...broker.PublishOption) error {
	queueURL, err := b.urlFromQueueName(queueName)
	if err != nil {
		return err
	}

	input := &sqs.SendMessageInput{
		MessageBody: aws.String(string(msg.Body[:])),
		QueueUrl:    &queueURL,
	}
	input.MessageAttributes = copyMessageHeader(msg)
	input.MessageDeduplicationId = b.generateDedupID(msg)
	input.MessageGroupId = b.generateGroupID(msg)

	log.Log(fmt.Sprintf("Publishing SQS message, %d bytes", len(msg.Body)))
	_, err = b.svc.SendMessage(input)

	if err != nil {
		return err
	}

	// Broker interfaces don't let us do anything with message ID or sequence number
	return nil
}
370048
func (ai *Amqp) Publish(exchange, key string, opts amqp.Publishing) error {
	return ai.ch.Publish(exchange, key, false, false, opts)
}
92879
func (service *BaseService) Publish(
	topic string,
	timeout time.Duration,
	heartbeat time.Duration) Service {
	if timeout <= 0 {
		timeout = service.Timeout
	}
	if heartbeat <= 0 {
		heartbeat = service.Heartbeat
	}
	t := newTopic(heartbeat)
	service.topicLock.Lock()
	service.topics[topic] = t
	service.topicLock.Unlock()
	return service.AddFunction(topic, func(id string) interface{} {
		message := t.get(id)
		if message == nil {
			message = make(chan interface{})
			t.put(id, message)
			fireSubscribeEvent(topic, id, service)
		}
	receiveMessage:
		select {
		case result := <-message:
			if result == heartbeatSignals {
				goto receiveMessage
			}
			return result
		case <-time.After(timeout):
			go func() {
				select {
				case message <- heartbeatSignals:
					break
				case <-time.After(t.heartbeat):
					service.offline(t, topic, id)
				}
			}()
			return nil
		}
	}, Options{})
}
172977
func (p *Publisher) Publish(val interface{}) {
	p.m.Lock()
	defer p.m.Unlock()

	msg := newMsg(val)
	if p.lastMsg != nil {
		p.lastMsg.next <- msg
	}
	p.lastMsg = msg
}
666680
func (b *RTMBroker) Publish(e Publishable) error {
	d, err := e.Publishable()
	if err != nil {
		return err
	}

	return b.conn.WriteMessage(websocket.TextMessage, d)
}