// Multicast method
12667
func (client *Client) Multicast(to []string, messages ...SendingMessage) *MulticastCall {
	return &MulticastCall{
		c:        client,
		to:       to,
		messages: messages,
	}
}
159893
func (s *Server) Broadcast(hash *uint64, msg *protocol.Message) error {
	alreadySentTo := make(map[uint64]bool)
	if msg.Gossip {
		for _, to := range msg.SentTo {
			alreadySentTo[to] = true
		}
	}
	sentTo := []uint64{murmur3.Sum64([]byte(s.LocalID()))}
	var toPeers []*Conn
	for _, peer := range s.Peers {
		peerHash := murmur3.Sum64([]byte(peer.Peer.Id))
		if (hash == nil || peer.Peer.GetKeyspace().Includes(*hash)) && !alreadySentTo[peerHash] {
			sentTo = append(sentTo, peerHash)
			toPeers = append(toPeers, peer)
		}
	}
	if msg.Gossip {
		msg.SentTo = append(msg.SentTo, sentTo...)
	}
	if len(toPeers) == 0 {
		return ErrNoRecipients
	}
	for _, peer := range toPeers {
		s.Printf("Broadcasting to %s", peer.Peer.Id)
		if err := peer.Send(msg); err != nil {
			return err
		}
	}
	return nil
}
193379
func (n *NetworkServerAPI) CreateMulticastGroup(ctx context.Context, req *ns.CreateMulticastGroupRequest) (*ns.CreateMulticastGroupResponse, error) {
	if req.MulticastGroup == nil {
		return nil, grpc.Errorf(codes.InvalidArgument, "multicast_group must not be nil")
	}

	mg := storage.MulticastGroup{
		FCnt:           req.MulticastGroup.FCnt,
		DR:             int(req.MulticastGroup.Dr),
		Frequency:      int(req.MulticastGroup.Frequency),
		PingSlotPeriod: int(req.MulticastGroup.PingSlotPeriod),
	}

	switch req.MulticastGroup.GroupType {
	case ns.MulticastGroupType_CLASS_B:
		mg.GroupType = storage.MulticastGroupB
	case ns.MulticastGroupType_CLASS_C:
		mg.GroupType = storage.MulticastGroupC
	default:
		return nil, grpc.Errorf(codes.InvalidArgument, "invalid group_type")
	}

	copy(mg.ID[:], req.MulticastGroup.Id)
	copy(mg.MCAddr[:], req.MulticastGroup.McAddr)
	copy(mg.MCNwkSKey[:], req.MulticastGroup.McNwkSKey)
	copy(mg.ServiceProfileID[:], req.MulticastGroup.ServiceProfileId)
	copy(mg.RoutingProfileID[:], req.MulticastGroup.RoutingProfileId)

	if err := storage.CreateMulticastGroup(storage.DB(), &mg); err != nil {
		return nil, errToRPCError(err)
	}

	return &ns.CreateMulticastGroupResponse{
		Id: mg.ID.Bytes(),
	}, nil
}
628180
func (o *MultiCastList) MultiCastChannelMaps(info *bambou.FetchingInfo) (MultiCastChannelMapsList, *bambou.Error) {

	var list MultiCastChannelMapsList
	err := bambou.CurrentSession().FetchChildren(o, MultiCastChannelMapIdentity, &list, info)
	return list, err
}
857877
func (worker *Worker) broadcast(outpack *outPack) {
	for _, v := range worker.agents {
		v.write(outpack)
	}
}
286334
func (mgr *AddrSubManager) BroadcastAddr(p peer.ID, addr ma.Multiaddr) {
	mgr.mu.RLock()
	defer mgr.mu.RUnlock()

	if subs, ok := mgr.subs[p]; ok {
		for _, sub := range subs {
			sub.pubAddr(addr)
		}
	}
}
994031
func GetMulticastGroupsForDevEUI(db sqlx.Queryer, devEUI lorawan.EUI64) ([]uuid.UUID, error) {
	var out []uuid.UUID

	err := sqlx.Select(db, &out, `
		select
			multicast_group_id
		from
			device_multicast_group
		where
			dev_eui = $1`,
		devEUI[:])
	if err != nil {
		return nil, handlePSQLError(err, "select error")
	}

	return out, nil
}
429046
func (peer *Peer) broadcast() error {
	envelopes := peer.host.Envelopes()
	bundle := make([]*Envelope, 0, len(envelopes))
	for _, envelope := range envelopes {
		if !peer.marked(envelope) && envelope.PoW() >= peer.powRequirement && peer.bloomMatch(envelope) {
			bundle = append(bundle, envelope)
		}
	}

	if len(bundle) > 0 {
		// transmit the batch of envelopes
		if err := p2p.Send(peer.ws, messagesCode, bundle); err != nil {
			return err
		}

		// mark envelopes only if they were successfully sent
		for _, e := range bundle {
			peer.mark(e)
		}

		log.Trace("broadcast", "num. messages", len(bundle))
	}
	return nil
}
8746
func CreateMulticastGroup(db sqlx.Execer, mg *MulticastGroup) error {
	now := time.Now()
	mg.CreatedAt = now
	mg.UpdatedAt = now

	if mg.ID == uuid.Nil {
		var err error
		mg.ID, err = uuid.NewV4()
		if err != nil {
			return errors.Wrap(err, "new uuid v4 error")
		}
	}

	_, err := db.Exec(`
		insert into multicast_group (
			id,
			created_at,
			updated_at,
			mc_addr,
			mc_nwk_s_key,
			f_cnt,
			group_type,
			dr,
			frequency,
			ping_slot_period,
			service_profile_id,
			routing_profile_id
		) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,
		mg.ID,
		mg.CreatedAt,
		mg.UpdatedAt,
		mg.MCAddr[:],
		mg.MCNwkSKey[:],
		mg.FCnt,
		mg.GroupType,
		mg.DR,
		mg.Frequency,
		mg.PingSlotPeriod,
		mg.ServiceProfileID,
		mg.RoutingProfileID,
	)
	if err != nil {
		return handlePSQLError(err, "insert error")
	}

	log.WithFields(log.Fields{
		"id": mg.ID,
	}).Info("multicast-group created")

	return nil
}
106472
func GetMulticastQueueItemsForMulticastGroup(db sqlx.Queryer, multicastGroupID uuid.UUID) ([]MulticastQueueItem, error) {
	var items []MulticastQueueItem

	err := sqlx.Select(db, &items, `
		select
			*
		from
			multicast_queue
		where
			multicast_group_id = $1
		order by
			id
	`, multicastGroupID)
	if err != nil {
		return nil, handlePSQLError(err, "select error")
	}

	return items, nil
}
393127
func (t *tribe) broadcast(mt msgType, msg interface{}, notify chan<- struct{}) error {
	raw, err := encodeMessage(mt, msg)
	if err != nil {
		return err
	}

	t.broadcasts.QueueBroadcast(&broadcast{
		msg:    raw,
		notify: notify,
	})
	return nil
}
430045
func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
	if !memR.config.Broadcast {
		return
	}

	peerID := memR.ids.GetForPeer(peer)
	var next *clist.CElement
	for {
		// In case of both next.NextWaitChan() and peer.Quit() are variable at the same time
		if !memR.IsRunning() || !peer.IsRunning() {
			return
		}
		// This happens because the CElement we were looking at got garbage
		// collected (removed). That is, .NextWait() returned nil. Go ahead and
		// start from the beginning.
		if next == nil {
			select {
			case <-memR.Mempool.TxsWaitChan(): // Wait until a tx is available
				if next = memR.Mempool.TxsFront(); next == nil {
					continue
				}
			case <-peer.Quit():
				return
			case <-memR.Quit():
				return
			}
		}

		memTx := next.Value.(*mempoolTx)

		// make sure the peer is up to date
		peerState, ok := peer.Get(types.PeerStateKey).(PeerState)
		if !ok {
			// Peer does not have a state yet. We set it in the consensus reactor, but
			// when we add peer in Switch, the order we call reactors#AddPeer is
			// different every time due to us using a map. Sometimes other reactors
			// will be initialized before the consensus reactor. We should wait a few
			// milliseconds and retry.
			time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
			continue
		}
		if peerState.GetHeight() < memTx.Height()-1 { // Allow for a lag of 1 block
			time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
			continue
		}

		// ensure peer hasn't already sent us this tx
		if _, ok := memTx.senders.Load(peerID); !ok {
			// send memTx
			msg := &TxMessage{Tx: memTx.tx}
			success := peer.Send(MempoolChannel, cdc.MustMarshalBinaryBare(msg))
			if !success {
				time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
				continue
			}
		}

		select {
		case <-next.NextWaitChan():
			// see the start of the for loop for nil check
			next = next.Next()
		case <-peer.Quit():
			return
		case <-memR.Quit():
			return
		}
	}
}
165917
func Cast(b []byte) Multiaddr {
	m, err := NewMultiaddrBytes(b)
	if err != nil {
		panic(fmt.Errorf("multiaddr failed to parse: %s", err))
	}
	return m
}
1203144
func (conn *BytesConnectionFields) MarkOffset(msg client.ConsumerMessage, metadata string) {
	if conn.multiplexer != nil && conn.multiplexer.Consumer != nil {
		conn.multiplexer.Consumer.MarkOffset(&msg, metadata)
	}
}
1367418
func (stc *ScatterConn) multiGo(
	ctx context.Context,
	name string,
	rss []*srvtopo.ResolvedShard,
	tabletType topodatapb.TabletType,
	action shardActionFunc,
) (allErrors *concurrency.AllErrorRecorder) {
	allErrors = new(concurrency.AllErrorRecorder)
	if len(rss) == 0 {
		return allErrors
	}

	oneShard := func(rs *srvtopo.ResolvedShard, i int) {
		var err error
		startTime, statsKey := stc.startAction(name, rs.Target)
		defer stc.endAction(startTime, allErrors, statsKey, &err, nil)
		err = action(rs, i)
	}

	if len(rss) == 1 {
		// only one shard, do it synchronously.
		oneShard(rss[0], 0)
		return allErrors
	}

	var wg sync.WaitGroup
	for i, rs := range rss {
		wg.Add(1)
		go func(rs *srvtopo.ResolvedShard, i int) {
			defer wg.Done()
			oneShard(rs, i)
		}(rs, i)
	}
	wg.Wait()
	return allErrors
}
487771
func GetMulticastGroup(db sqlx.Queryer, id uuid.UUID, forUpdate bool) (MulticastGroup, error) {
	var mg MulticastGroup
	var fu string

	if forUpdate {
		fu = " for update"
	}

	err := sqlx.Get(db, &mg, `
		select
			*
		from
			multicast_group
		where
			id = $1`+fu,
		id,
	)
	if err != nil {
		return mg, handlePSQLError(err, "select error")
	}
	return mg, nil
}
642489
func GetUnicast(o dhcp6.Options) (IP, error) {
	v, err := o.GetOne(dhcp6.OptionUnicast)
	if err != nil {
		return nil, err
	}

	var ip IP
	err = ip.UnmarshalBinary(v)
	return ip, err
}