// Distribute provide a function that takes a React and other multiple Publishers and distribute the data from the first Publisher to others
9508
func Distribute(rs Publisher, ms ...Sender) Publisher {
	return rs.React(func(r Publisher, err error, data interface{}) {
		for _, mi := range ms {
			go func(rec Sender) {
				if err != nil {
					rec.SendError(err)
					return
				}
				rec.Send(data)
			}(mi)
		}
	}, true)
}
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
}
161561
func DistributeSignals(rs Reactor, ms ...Sender) Reactor {
	return rs.React(func(r Reactor, err error, data interface{}) {
		for _, mi := range ms {
			go func(rec Sender) {
				if err != nil {
					rec.SendError(err)
					return
				}
				rec.Send(data)
			}(mi)
		}
	}, true)
}
1159542
func (g PipelineGroup) Reduce() error {
	var deployment *DeploymentConfigRef
	for _, p := range g {
		if p.Deployment == nil || p.Deployment == deployment {
			continue
		}
		if deployment == nil {
			deployment = p.Deployment
		} else {
			deployment.Images = append(deployment.Images, p.Deployment.Images...)
			deployment.Env = NewEnvironment(deployment.Env, p.Deployment.Env)
			p.Deployment = deployment
		}
	}
	return nil
}
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
}
749485
func (c *Client) Publish(topic string, payload []byte, qos packet.QOS, retain bool) (GenericFuture, error) {
	msg := &packet.Message{
		Topic:   topic,
		Payload: payload,
		QOS:     qos,
		Retain:  retain,
	}

	return c.PublishMessage(msg)
}
101407
func (np NoopProvider) Publish(ctx context.Context, topic string, m *Msg) error {
	return nil
}
228481
func (s *Server) Publish(ctx context.Context, msg interface{}) error {
	return s.PublishWithTags(ctx, msg, query.TagMap(make(map[string]interface{})))
}
480992
func (h *Hub) Publish(e Event) {
	h.m.RLock()
	if handlers, ok := h.subscribers[e.Kind()]; ok {
		for _, h := range handlers {
			h.f(e)
		}
	}
	h.m.RUnlock()
}
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})
}
856939
func Publish(ctx context.Context, topic string, msg proto.Message) *PublishResult {
	pr := &PublishResult{Ready: make(chan struct{})}
	go func() {
		err := client.Publish(ctx, topic, msg, false)
		pr.Err = err
		close(pr.Ready)
	}()
	return pr
}
669427
func (ps *PubSub) Publish(topic string, item interface{}) error {
	ps.RLock()
	defer ps.RUnlock()
	s, subscribed := ps.subscriptions[topic]
	if !subscribed {
		return errors.New("no subscribers")
	}
	for _, sub := range s.ToArray() {
		c := sub.(*subscription).c
		// Not enough room in buffer, continue in order to not block publisher
		if len(c) == subscriptionBuffSize {
			continue
		}
		c <- item
	}
	return nil
}
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
}
1016932
func (Pipe) Publish(ctx *context.Context) error {
	client, err := client.NewGitHub(ctx)
	if err != nil {
		return err
	}
	return doRun(ctx, client)
}
292902
func (c FakeClient) Publish(id *charm.URL, channels []params.Channel, resources map[string]int) error {
	return c.repo.Publish(id, channels, resources)
}
37214
func (sc *conn) PublishAsync(subject string, data []byte, ah AckHandler) (string, error) {
	return sc.publishAsync(subject, data, ah, nil)
}
595512
func (router *Distribute) Start() error {
	for _, streamID := range router.boundStreamIDs {
		targetRouter := core.StreamRegistry.GetRouterOrFallback(streamID)
		router.routers = append(router.routers, targetRouter)
	}
	return nil
}