// Publisher returns a LogPublisher for the given subscription
// as well as a cancel function that should be called when the log stream
// is completed.
14953
func (a *Agent) Publisher(ctx context.Context, subscriptionID string) (exec.LogPublisher, func(), error) {
	// TODO(stevvooe): The level of coordination here is WAY too much for logs.
	// These should only be best effort and really just buffer until a session is
	// ready. Ideally, they would use a separate connection completely.

	var (
		err       error
		publisher api.LogBroker_PublishLogsClient
	)

	err = a.withSession(ctx, func(session *session) error {
		publisher, err = api.NewLogBrokerClient(session.conn.ClientConn).PublishLogs(ctx)
		return err
	})
	if err != nil {
		return nil, nil, err
	}

	// make little closure for ending the log stream
	sendCloseMsg := func() {
		// send a close message, to tell the manager our logs are done
		publisher.Send(&api.PublishLogsMessage{
			SubscriptionID: subscriptionID,
			Close:          true,
		})
		// close the stream forreal
		publisher.CloseSend()
	}

	return exec.LogPublisherFunc(func(ctx context.Context, message api.LogMessage) error {
			select {
			case <-ctx.Done():
				sendCloseMsg()
				return ctx.Err()
			default:
			}

			return publisher.Send(&api.PublishLogsMessage{
				SubscriptionID: subscriptionID,
				Messages:       []api.LogMessage{message},
			})
		}), func() {
			sendCloseMsg()
		}, nil
}
489623
func endSubscriber(m *Miniredis, c *server.Peer) {
	ctx := getCtx(c)
	if sub := ctx.subscriber; sub != nil {
		m.removeSubscriber(sub) // will Close() the sub
	}
	ctx.subscriber = nil
}
19217
func NewPublisher(etcdHost string, etcdPort int, etcdPath string, publishInterval int,
	publishTTL int, logHost string, logPort int) (*Publisher, error) {
	etcdClient := etcd.NewClient([]string{fmt.Sprintf("http://%s:%d", etcdHost, etcdPort)})
	ticker := time.NewTicker(time.Duration(publishInterval) * time.Second)
	return &Publisher{
		etcdClient: etcdClient,
		etcdPath:   etcdPath,
		publishTTL: uint64(time.Duration(publishTTL) * time.Second),
		logHost:    logHost,
		logPort:    logPort,
		ticker:     ticker,
	}, nil
}
159635
func (p *Publisher) Publish(topic string, body []byte) error {
	return p.prod.Publish(topic, body)
}
828553
func (s *Collector) Unsubscribe(c *container.Container, ch chan interface{}) {
	s.m.Lock()
	publisher := s.publishers[c]
	if publisher != nil {
		publisher.Evict(ch)
		if publisher.Len() == 0 {
			delete(s.publishers, c)
		}
	}
	s.m.Unlock()
}
90236
func NewPublisher(
	publisher *nats.Conn,
	subject string,
	enc EncodeRequestFunc,
	dec DecodeResponseFunc,
	options ...PublisherOption,
) *Publisher {
	p := &Publisher{
		publisher: publisher,
		subject:   subject,
		enc:       enc,
		dec:       dec,
		timeout:   10 * time.Second,
	}
	for _, option := range options {
		option(p)
	}
	return p
}
971012
func NewPublisher(addr string) (*Publisher, error) {
	cfg := nsq.NewConfig()
	prod, err := nsq.NewProducer(addr, cfg)
	if err != nil {
		return nil, err
	}

	return &Publisher{
		prod: prod,
	}, nil
}
800176
func NewPublisher(ctx context.Context, projectID string, opts ...option.ClientOption) (*Publisher, error) {
	client, err := pubsub.NewClient(ctx, projectID, opts...)
	if err != nil {
		return nil, err
	}

	return &Publisher{
		client: client,
		topics: make(map[string]*pubsub.Topic),
	}, nil
}
1212349
func (up *UnbufferedPublisher) Publish(c context.Context, msgs ...*pubsub.Message) ([]string, error) {
	if len(msgs) == 0 {
		return nil, nil
	}

	messages := make([]*pb.PubsubMessage, len(msgs))
	for i, msg := range msgs {
		messages[i] = &pb.PubsubMessage{
			Data:       msg.Data,
			Attributes: msg.Attributes,
		}
	}

	client, err := up.ClientFactory.Client(up.AECtx)
	if err != nil {
		return nil, err
	}

	resp, err := client.Publish(c, &pb.PublishRequest{
		Topic:    string(up.Topic),
		Messages: messages,
	}, up.CallOpts...)
	if err != nil {
		// Optimistically recreate the client.
		up.ClientFactory.RecreateClient()
		logging.Debugf(c, "Recreating a new PubSub client due to error")
		return nil, err
	}
	return resp.MessageIds, nil
}
628157
func (s *SubscriptionProvider) Subscribe() *Subscription {
	s.Lock()
	defer s.Unlock()
	if s.subscriptions == nil {
		s.subscriptions = make(map[string]*Subscription)
	}
	sub := newSubscription(s)
	s.subscriptions[sub.id.String()] = sub

	return sub
}
361873
func (w *Wallet) PublishTransaction(tx *wire.MsgTx) error {
	_, err := w.reliablyPublishTransaction(tx)
	return err
}
534536
func (p *Publisher) Publish(pub amqp.Publishing) error {
	return p.PublishWithRoutingKey(pub, p.key)
}
608803
func (s *StanServer) ackPublisher(iopm *ioPendingMsg) {
	msgAck := &iopm.pa
	msgAck.Guid = iopm.pm.Guid
	needed := msgAck.Size()
	s.tmpBuf = util.EnsureBufBigEnough(s.tmpBuf, needed)
	n, _ := msgAck.MarshalTo(s.tmpBuf)
	if s.trace {
		pm := &iopm.pm
		s.log.Tracef("[Client:%s] Acking Publisher subj=%s guid=%s", pm.ClientID, pm.Subject, pm.Guid)
	}
	s.ncs.Publish(iopm.m.Reply, s.tmpBuf[:n])
}
895156
func (b *pubsubBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
	if b.err != nil {
		return b.err
	}
	t := b.topic(topic)
	return t.Send(context.Background(), &pubsub.Message{Metadata: msg.Header, Body: msg.Body})
}
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,
	})
}
444651
func NewRepublisher(ns namesys.Publisher, ds ds.Datastore, self ic.PrivKey, ks keystore.Keystore) *Republisher {
	return &Republisher{
		ns:             ns,
		ds:             ds,
		self:           self,
		ks:             ks,
		Interval:       DefaultRebroadcastInterval,
		RecordLifetime: DefaultRecordLifetime,
	}
}
622893
func NewPublisher(topic string, c client.Client) Publisher {
	if c == nil {
		c = client.NewClient()
	}
	return &publisher{c, topic}
}