// Publish queues up the message if and only if we have an active connection to
// the target apiserver.
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)
			}
		}
	}
}
711512
func (mr *MockInstanceConnectorMockRecorder) Publish(msg interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Publish", reflect.TypeOf((*MockInstanceConnector)(nil).Publish), 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)
}
1226178
func (w *Producer) Publish(topic string, body []byte) error {
	return w.sendCommand(Publish(topic, body))
}
527891
func (c *syncClient) Publish(arg0 string, arg1 []byte) (rcvCnt int64, err Error) {
	arg0bytes := []byte(arg0)
	arg1bytes := arg1

	var resp Response
	resp, err = c.conn.ServiceRequest(&PUBLISH, [][]byte{arg0bytes, arg1bytes})
	if err == nil {
		rcvCnt = resp.GetNumberValue()
	}
	return rcvCnt, err
}
1321682
func (br *defaultBroker) Publish(pub Sender, msg *Publish) {
	pubID := NewID()
	evtTemplate := Event{
		Publication: pubID,
		Arguments:   msg.Arguments,
		ArgumentsKw: msg.ArgumentsKw,
		Details:     make(map[string]interface{}),
	}

	excludePublisher := true
	if exclude, ok := msg.Options["exclude_me"].(bool); ok {
		excludePublisher = exclude
	}

	for id, sub := range br.routes[msg.Topic] {
		// shallow-copy the template
		event := evtTemplate
		event.Subscription = id
		// don't send event to publisher
		if (excludePublisher && sub != pub) || !excludePublisher {
			sub.Send(&event)
		}
	}

	// only send published message if acknowledge is present and set to true
	if doPub, _ := msg.Options["acknowledge"].(bool); doPub {
		pub.Send(&Published{Request: msg.Request, Publication: pubID})
	}
}
325053
func (Pipe) Publish(ctx *context.Context) error {
	if len(ctx.Config.S3) == 0 {
		return pipe.Skip("s3 section is not configured")
	}
	var g = semerrgroup.New(ctx.Parallelism)
	for _, conf := range ctx.Config.S3 {
		conf := conf
		g.Go(func() error {
			return upload(ctx, conf)
		})
	}
	return g.Wait()
}
135488
func (p *Publisher) Publish(topic string, body []byte) error {
	pub := p.topic(topic)
	res := pub.Publish(context.Background(), &pubsub.Message{
		Data: body,
	})
	_, err := res.Get(context.Background())
	return err
}
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
}
68320
func (m *MockConnector) Publish(msg *Message) error {
	ret := m.ctrl.Call(m, "Publish", msg)
	ret0, _ := ret[0].(error)
	return ret0
}
1250017
func (m *Miniredis) Publish(channel, message string) int {
	m.Lock()
	defer m.Unlock()
	return m.publish(channel, message)
}
645507
func (fn LogPublisherFunc) Publish(ctx context.Context, message api.LogMessage) error {
	return fn(ctx, message)
}
1139570
func (p *IpnsPublisher) Publish(ctx context.Context, k ci.PrivKey, value path.Path) error {
	log.Debugf("Publish %s", value)
	return p.PublishWithEOL(ctx, k, value, time.Now().Add(DefaultRecordEOL))
}
996820
func Publish(ctx context.Context, msg Message, opts ...PublishOption) error {
	return DefaultClient.Publish(ctx, msg, opts...)
}
227791
func (sc *conn) Publish(subject string, data []byte) error {
	// Need to make this a buffered channel of 1 in case
	// a publish call is blocked in pubAckChan but cleanupOnClose()
	// is trying to push the error to this channel.
	ch := make(chan error, 1)
	_, err := sc.publishAsync(subject, data, nil, ch)
	if err == nil {
		err = <-ch
	}
	return err
}
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,
	})
}
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
}