// BroadcastMultiple broadcasts a text message to multiple sessions given in the sessions slice.
350237
func (m *Melody) BroadcastMultiple(msg []byte, sessions []*Session) error {
	for _, sess := range sessions {
		if writeErr := sess.Write(msg); writeErr != nil {
			return writeErr
		}
	}
	return nil
}
594993
func (s *Service) SubscribeMultiple(subscriptions []packet.Subscription) SubscribeFuture {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	// save subscription
	for _, v := range subscriptions {
		s.subscriptions.Set(v.Topic, v)
	}

	// allocate future
	f := future.New()

	// queue subscribe
	s.commandQueue <- &command{
		subscribe:     true,
		future:        f,
		subscriptions: subscriptions,
	}

	return &subscribeFuture{f}
}
761344
func (api *PrivateHostAPI) FilterMultiBy(key string, value interface{}) *PrivateHostAPI {
	api.filterBy(key, value, true)
	return api
}
838603
func (m *GormAreaRepository) LoadMultiple(ctx context.Context, ids []uuid.UUID) ([]Area, error) {
	defer goa.MeasureSince([]string{"goa", "db", "Area", "getmultiple"}, time.Now())
	var objs []Area
	if len(ids) == 0 {
		return objs, nil
	}
	for i := 0; i < len(ids); i++ {
		m.db = m.db.Or("id = ?", ids[i])
	}
	tx := m.db.Find(&objs)
	if tx.Error != nil {
		return nil, errors.NewInternalError(ctx, tx.Error)
	}
	return objs, nil
}
454474
func (m *ChannelDeMultiplexer) DeMultiplex(msg interface{}) {
	m.lock.RLock()
	defer m.lock.RUnlock()
	if m.isClosed() {
		return
	}
	for _, ch := range m.channels {
		if ch.pred(msg) {
			ch.ch <- msg
		}
	}
}
442320
func (c *Client) BroadcastEnvelope(txEnv *txs.Envelope, logger *logging.Logger) (*exec.TxExecution, error) {
	err := c.dial(logger)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), c.timeout)
	defer cancel()

	return c.transactClient.BroadcastTxSync(ctx, &rpctransact.TxEnvelopeParam{Envelope: txEnv})
}
795125
func (w *Producer) MultiPublish(topic string, body [][]byte) error {
	cmd, err := MultiPublish(topic, body)
	if err != nil {
		return err
	}
	return w.sendCommand(cmd)
}
447734
func (api *PacketFilterAPI) FilterMultiBy(key string, value interface{}) *PacketFilterAPI {
	api.filterBy(key, value, true)
	return api
}
765265
func (c *ChatRPC) ChatKBFSToImpteamUpgrade(
	_ context.Context, _ chat1.ChatKBFSToImpteamUpgradeArg) error {
	return nil
}
334166
func NewMultiTimings(name string, help string, labels []string) *MultiTimings {
	t := &MultiTimings{
		Timings: Timings{
			histograms: make(map[string]*Histogram),
			help:       help,
		},
		labels: labels,
	}
	if name != "" {
		publish(name, t)
	}

	return t
}
872127
func (d *Disque) FetchMultiple(queueName string, count int, timeout time.Duration) (jobs []*Job, err error) {
	jobs = make([]*Job, 0)
	if err = d.pickClient(); err == nil {
		args := redis.Args{}.
			Add("TIMEOUT").Add(int64(timeout.Seconds() * 1000)).
			Add("COUNT").Add(count).
			Add("WITHCOUNTERS").
			Add("FROM").Add(queueName)

		if values, err := redis.Values(d.call("GETJOB", args)); err == nil {
			for _, job := range values {
				if jobValues, err := redis.Values(job, err); err == nil {
					var nacks, additionalDeliveries int64
					nacks, err = redis.Int64(jobValues[4], err)
					additionalDeliveries, err = redis.Int64(jobValues[6], err)

					j := &Job{
						QueueName:            string(jobValues[0].([]byte)),
						JobID:                string(jobValues[1].([]byte)),
						Message:              string(jobValues[2].([]byte)),
						Nacks:                nacks,
						AdditionalDeliveries: additionalDeliveries,
					}
					jobs = append(jobs, j)

					// update stats using fragment of the job-id
					statsKey := j.JobID[2:10]
					d.stats[statsKey] = d.stats[statsKey] + 1
				}
			}
		}
	}
	d.count += count
	return jobs, err
}
761596
func (c *Connector) MultiUpsert(ctx context.Context, ei *dosa.EntityInfo, multiValues []map[string]dosa.FieldValue) ([]error, error) {
	values, err := fieldValueMapsFromClientMaps(multiValues)
	if err != nil {
		return nil, err
	}

	// ttl := dosa.NoTTL().Nanoseconds()
	// if ei.TTL != nil {
	// 	ttl = ei.TTL.Nanoseconds()
	// }

	// perform the multi upsert request
	request := &dosarpc.MultiUpsertRequest{
		Ref:      entityInfoToSchemaRef(ei),
		Entities: values,
		// TTL:      &ttl, mgode@ has not yet committed origin/ttl-for-multi-upsert
	}

	response, err := c.client.MultiUpsert(ctx, request, getHeaders(c.headers)...)
	if err != nil {
		if !dosarpc.Dosa_MultiUpsert_Helper.IsException(err) {
			return nil, errors.Wrap(err, "failed to MultiUpsert due to network issue")
		}

		return nil, errors.Wrap(err, "failed to MultiUpsert")
	}

	rpcErrors := response.Errors
	results := make([]error, len(rpcErrors))
	for i, rpcError := range rpcErrors {
		if rpcError != nil {
			results[i] = wrapIDLError(rpcError)
		}
	}

	return results, nil
}
12667
func (client *Client) Multicast(to []string, messages ...SendingMessage) *MulticastCall {
	return &MulticastCall{
		c:        client,
		to:       to,
		messages: messages,
	}
}
1178329
func (s *Sender) SendMulticastNoRetry(msg *Message, registrationIds []string) (*MulticastResult, error) {
	if err := checkUnrecoverableErrors(s, "", registrationIds, msg, 0); err != nil {
		return nil, err
	}
	rawMsg := &message{Message: *msg, registrationIds: registrationIds}

	resp, err := s.sendRaw(rawMsg)
	if err != nil {
		return nil, err
	}

	result := new(MulticastResult)
	result.Success = resp.Success
	result.Failure = resp.Failure
	result.CanonicalIds = resp.CanonicalIds
	result.MulticastID = resp.MulticastID
	if resp.Results != nil {
		result.Results = make([]Result, len(resp.Results))
		for i, res := range resp.Results {
			result.Results[i] = Result{
				MessageID:               res.MessageID,
				CanonicalRegistrationID: res.RegistrationID,
				Error: res.Err,
			}
		}
	}
	return result, nil
}
162730
func (api *IconAPI) FilterMultiBy(key string, value interface{}) *IconAPI {
	api.filterBy(key, value, true)
	return api
}
1221717
func (cm *rpcConnManager) BroadcastMessage(msg wire.Message) {
	cm.server.BroadcastMessage(msg)
}
770003
func NewMultiStream(
	listen net.Listener,
	dial func(string, time.Duration) (net.Conn, error),
	advertise net.Addr,
	lg *log.Logger,
	serviceCodes ...byte,
) (map[byte]raft.StreamLayer, error) {
	// sanity checks for listener and advertised address
	if advertise == nil {
		advertise = listen.Addr()
	}
	tcpAdvertise, ok := advertise.(*net.TCPAddr)
	if !ok {
		lg.Printf("NewMultiStream Warning: advertising non TCP address %s", advertise)
	} else if tcpAdvertise.IP.IsUnspecified() {
		lg.Printf("NewMultiStream Warning: advertising unspecified IP: %s for address %s", tcpAdvertise.IP, tcpAdvertise.String())
	}
	// set up router
	r := &router{listen, make(chan net.Conn, 16), make(chan closeReq, 16), make(map[byte]*serviceStreams), lg}
	// set up a channel of conns for each unique serviceCode
	for _, b := range serviceCodes {
		_, exists := r.chans[b]
		if exists {
			lg.Printf("Warning, serviceCode %d was present more than once in NewMultiStream -- allocating a single service")
			continue
		}
		r.chans[b] = &serviceStreams{
			r:      r,
			addr:   advertise,
			conns:  make(chan net.Conn, 16),
			closed: false,
			myCode: b,
			dial:   dial,
		}
	}
	// type hack, force to map[byte]raft.StreamLayer for return type
	retMap := make(map[byte]raft.StreamLayer)
	for k, v := range r.chans {
		retMap[k] = v
	}
	go r.serve()
	return retMap, nil
}