// Publish tells the emitter to publish with the given message and tags
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)
}
1186266
func (f FPublisher) Publish(mts []plugin.Metric, cfg plugin.Config) error {
	file, err := cfg.GetString("file")
	if err != nil {
		return err
	}
	if val, err := cfg.GetBool("return_error"); err == nil && val {
		return errors.New("Houston we have a problem")
	}
	fileHandle, _ := os.Create(file)
	writer := bufio.NewWriter(fileHandle)
	defer fileHandle.Close()

	for _, m := range mts {
		fmt.Fprintf(writer, "%s|%v|%d|%s|%s|%s|%v|%v\n",
			m.Namespace.Strings(),
			m.Data,
			m.Version,
			m.Unit,
			m.Description,
			m.Timestamp,
			m.Tags,
			m.Config,
		)
	}
	writer.Flush()

	return nil
}
711512
func (mr *MockInstanceConnectorMockRecorder) Publish(msg interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Publish", reflect.TypeOf((*MockInstanceConnector)(nil).Publish), msg)
}
101407
func (np NoopProvider) Publish(ctx context.Context, topic string, m *Msg) error {
	return nil
}
431427
func (c *client) Publish(topic string, qos byte, retained bool, payload interface{}) Token {
	token := newToken(packets.Publish).(*PublishToken)
	DEBUG.Println(CLI, "enter Publish")
	switch {
	case !c.IsConnected():
		token.setError(ErrNotConnected)
		return token
	case c.connectionStatus() == reconnecting && qos == 0:
		token.flowComplete()
		return token
	}
	pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
	pub.Qos = qos
	pub.TopicName = topic
	pub.Retain = retained
	switch payload.(type) {
	case string:
		pub.Payload = []byte(payload.(string))
	case []byte:
		pub.Payload = payload.([]byte)
	default:
		token.setError(fmt.Errorf("Unknown payload type"))
		return token
	}

	if pub.Qos != 0 && pub.MessageID == 0 {
		pub.MessageID = c.getID(token)
		token.messageID = pub.MessageID
	}
	persistOutbound(c.persist, pub)
	if c.connectionStatus() == reconnecting {
		DEBUG.Println(CLI, "storing publish message (reconnecting), topic:", topic)
	} else {
		DEBUG.Println(CLI, "sending publish message, topic:", topic)
		c.obound <- &PacketAndToken{p: pub, t: token}
	}
	return token
}
292902
func (c FakeClient) Publish(id *charm.URL, channels []params.Channel, resources map[string]int) error {
	return c.repo.Publish(id, channels, resources)
}
1140454
func (nc *Conn) Publish(subj string, data []byte) error {
	return nc.publish(subj, _EMPTY_, data)
}
797943
func (p *PubSub) Publish(i interface{}) {
	p.m.Lock()
	defer p.m.Unlock()

	for _, v := range p.subscribers {
		v.OnEvent(i)
	}
}
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
}
398599
func (q *Queue) Publish(item events.Event) {
	q.broadcast.Write(item)
}
1105141
func (e *eventer) Publish(name string, data interface{}) {
	evt := NewEvent(name, data)
	e.in <- evt
}
370048
func (ai *Amqp) Publish(exchange, key string, opts amqp.Publishing) error {
	return ai.ch.Publish(exchange, key, false, false, opts)
}
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
}
546903
func (r *remoteServer) Publish(message *params.PubSubMessage) {
	select {
	case <-r.tomb.Dying():
		r.logger.Tracef("dying, don't send %q", message.Topic)
	default:
		r.mutex.Lock()
		// Only queue the message up if we are currently connected.
		notifyData := false
		if r.connection != nil {
			r.logger.Tracef("queue up topic %q", message.Topic)
			r.pending.PushBack(message)
			notifyData = r.pending.Len() == 1

		} else {
			r.logger.Tracef("skipping %q for %s as not connected", message.Topic, r.target)
		}
		r.mutex.Unlock()
		if notifyData {
			select {
			case r.data <- struct{}{}:
			case <-r.connectionReset:
				r.logger.Debugf("connection reset while notifying %q for %s", message.Topic, r.target)
			}
		}
	}
}
1296132
func (w *clientWrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) (err error) {
	t := newPublicationTracker(p, ClientProfile)
	ctx = t.start(ctx, true)

	defer func() { t.end(ctx, err) }()

	ctx = injectTraceIntoCtx(ctx, t.span)

	err = w.Client.Publish(ctx, p, opts...)
	return
}
1151423
func (p *Provider) Publish(ctx context.Context, topic string, m *pubsub.Msg) error {
	w, err := p.writerForTopic(ctx, topic)
	if err != nil {
		return err
	}

	u1, err := uuid.NewV1()
	if err != nil {
		return err
	}

	return w.WriteMessages(ctx, kafka.Message{
		Key:   u1.Bytes(),
		Value: m.Data,
	})
}
892340
func (r *Redis) Publish(channel, message string) (int64, error) {
	rp, err := r.ExecuteCommand("PUBLISH", channel, message)
	if err != nil {
		return 0, err
	}
	return rp.IntegerValue()
}