// partitionMessages receives a predicate and a slice of gossip messages
// and returns a tuple of two slices: the messages that hold for the predicate
// and the rest
778919
func partitionMessages(pred common.MessageAcceptor, a []*emittedGossipMessage) ([]*emittedGossipMessage, []*emittedGossipMessage) {
	s1 := []*emittedGossipMessage{}
	s2 := []*emittedGossipMessage{}
	for _, m := range a {
		if pred(m) {
			s1 = append(s1, m)
		} else {
			s2 = append(s2, m)
		}
	}
	return s1, s2
}
383251
func (wc *WhisperClient) GetFilterMessages(ctx *Context, id string) (*Messages, error) {
	rawFilterMessages, err := wc.client.FilterMessages(ctx.context, id)
	if err != nil {
		return nil, err
	}
	res := make([]*whisper.Message, len(rawFilterMessages))
	copy(res, rawFilterMessages)
	return &Messages{res}, nil
}
175314
func (p *Peer) QueueMessage(msg wire.Message, doneChan chan<- struct{}) {
	p.QueueMessageWithEncoding(msg, doneChan, wire.BaseEncoding)
}
1241460
func (c *MemberlistManager) GetPartitions() []int32 {
	c.RLock()
	defer c.RUnlock()
	return c.members[c.nodeName].Partitions
}
328034
func (t *Topic) Partitions() (PartitionList, error) {
	value, _, err := t.kz.conn.Get(t.metadataPath())
	if err != nil {
		return nil, err
	}

	return t.parsePartitions(value)
}
1098087
func (s *MessageStore) Peers() (map[[33]byte]struct{}, error) {
	peers := make(map[[33]byte]struct{})
	err := s.db.View(func(tx *bbolt.Tx) error {
		messageStore := tx.Bucket(messageStoreBucket)
		if messageStore == nil {
			return ErrCorruptedMessageStore
		}

		return messageStore.ForEach(func(k, _ []byte) error {
			var pubKey [33]byte
			copy(pubKey[:], k[:33])
			peers[pubKey] = struct{}{}
			return nil
		})
	})
	if err != nil {
		return nil, err
	}

	return peers, nil
}
1218341
func (m *SingleNodeManager) GetPartitions() []int32 {
	m.RLock()
	defer m.RUnlock()
	return m.node.Partitions
}
1033751
func (s *StanServer) initPartitions() error {
	// The option says that the server should only use the pre-defined channels,
	// but none was specified. Don't see the point in continuing...
	if len(s.opts.StoreLimits.PerChannel) == 0 {
		return ErrNoChannel
	}
	nc, err := s.createNatsClientConn("pc")
	if err != nil {
		return err
	}
	p := &partitions{
		s:  s,
		nc: nc,
	}
	// Now that the connection is created, we need to set s.partitioning to cp
	// so that server shutdown can properly close this connection.
	s.partitions = p
	p.createChannelsMapAndSublist(s.opts.StoreLimits.PerChannel)
	p.sendListSubject = partitionsPrefix + "." + s.opts.ID
	// Use the partitions' own connection for channels list requests
	p.processChanSub, err = p.nc.Subscribe(p.sendListSubject, p.processChannelsListRequests)
	if err != nil {
		return fmt.Errorf("unable to subscribe: %v", err)
	}
	p.processChanSub.SetPendingLimits(-1, -1)
	p.inboxSub, err = p.nc.SubscribeSync(nats.NewInbox())
	if err != nil {
		return fmt.Errorf("unable to subscribe: %v", err)
	}
	p.Lock()
	// Set this before the first attempt so we don't miss any notification
	// of a change in topology. Since we hold the lock, and even if there
	// was a notification happening now, the callback will execute only
	// after we are done with the initial check.
	nc.SetDiscoveredServersHandler(p.topologyChanged)
	// Now send our list and check if any server is complaining
	// about having one channel in common.
	if err := p.checkChannelsUniqueInCluster(); err != nil {
		p.Unlock()
		return err
	}
	p.Unlock()
	return nil
}
1213817
func (g *gossipServiceImpl) gossipInChan(messages []*emittedGossipMessage, chanRoutingFactory channelRoutingFilterFactory) {
	if len(messages) == 0 {
		return
	}
	totalChannels := extractChannels(messages)
	var channel common.ChainID
	var messagesOfChannel []*emittedGossipMessage
	for len(totalChannels) > 0 {
		// Take first channel
		channel, totalChannels = totalChannels[0], totalChannels[1:]
		// Extract all messages of that channel
		grabMsgs := func(o interface{}) bool {
			return bytes.Equal(o.(*emittedGossipMessage).Channel, channel)
		}
		messagesOfChannel, messages = partitionMessages(grabMsgs, messages)
		if len(messagesOfChannel) == 0 {
			continue
		}
		// Grab channel object for that channel
		gc := g.chanState.getGossipChannelByChainID(channel)
		if gc == nil {
			g.logger.Warning("Channel", channel, "wasn't found")
			continue
		}
		// Select the peers to send the messages to
		// For leadership messages we will select all peers that pass routing factory - e.g. all peers in channel and org
		membership := g.disc.GetMembership()
		var peers2Send []*comm.RemotePeer
		if messagesOfChannel[0].IsLeadershipMsg() {
			peers2Send = filter.SelectPeers(len(membership), membership, chanRoutingFactory(gc))
		} else {
			peers2Send = filter.SelectPeers(g.conf.PropagatePeerNum, membership, chanRoutingFactory(gc))
		}

		// Send the messages to the remote peers
		for _, msg := range messagesOfChannel {
			filteredPeers := g.removeSelfLoop(msg, peers2Send)
			g.comm.Send(msg.SignedGossipMessage, filteredPeers...)
		}
	}
}
287774
func (g *GossipSyncer) FilterGossipMsgs(msgs ...msgWithSenders) {
	// If the peer doesn't have an update horizon set, then we won't send
	// it any new update messages.
	if g.remoteUpdateHorizon == nil {
		return
	}

	// If we've been signaled to exit, or are exiting, then we'll stop
	// short.
	select {
	case <-g.quit:
		return
	default:
	}

	// TODO(roasbeef): need to ensure that peer still online...send msg to
	// gossiper on peer termination to signal peer disconnect?

	var err error

	// Before we filter out the messages, we'll construct an index over the
	// set of channel announcements and channel updates. This will allow us
	// to quickly check if we should forward a chan ann, based on the known
	// channel updates for a channel.
	chanUpdateIndex := make(map[lnwire.ShortChannelID][]*lnwire.ChannelUpdate)
	for _, msg := range msgs {
		chanUpdate, ok := msg.msg.(*lnwire.ChannelUpdate)
		if !ok {
			continue
		}

		chanUpdateIndex[chanUpdate.ShortChannelID] = append(
			chanUpdateIndex[chanUpdate.ShortChannelID], chanUpdate,
		)
	}

	// We'll construct a helper function that we'll us below to determine
	// if a given messages passes the gossip msg filter.
	g.Lock()
	startTime := time.Unix(int64(g.remoteUpdateHorizon.FirstTimestamp), 0)
	endTime := startTime.Add(
		time.Duration(g.remoteUpdateHorizon.TimestampRange) * time.Second,
	)
	g.Unlock()

	passesFilter := func(timeStamp uint32) bool {
		t := time.Unix(int64(timeStamp), 0)
		return t.Equal(startTime) ||
			(t.After(startTime) && t.Before(endTime))
	}

	msgsToSend := make([]lnwire.Message, 0, len(msgs))
	for _, msg := range msgs {
		// If the target peer is the peer that sent us this message,
		// then we'll exit early as we don't need to filter this
		// message.
		if _, ok := msg.senders[g.cfg.peerPub]; ok {
			continue
		}

		switch msg := msg.msg.(type) {

		// For each channel announcement message, we'll only send this
		// message if the channel updates for the channel are between
		// our time range.
		case *lnwire.ChannelAnnouncement:
			// First, we'll check if the channel updates are in
			// this message batch.
			chanUpdates, ok := chanUpdateIndex[msg.ShortChannelID]
			if !ok {
				// If not, we'll attempt to query the database
				// to see if we know of the updates.
				chanUpdates, err = g.cfg.channelSeries.FetchChanUpdates(
					g.cfg.chainHash, msg.ShortChannelID,
				)
				if err != nil {
					log.Warnf("no channel updates found for "+
						"short_chan_id=%v",
						msg.ShortChannelID)
					continue
				}
			}

			for _, chanUpdate := range chanUpdates {
				if passesFilter(chanUpdate.Timestamp) {
					msgsToSend = append(msgsToSend, msg)
					break
				}
			}

			if len(chanUpdates) == 0 {
				msgsToSend = append(msgsToSend, msg)
			}

		// For each channel update, we'll only send if it the timestamp
		// is between our time range.
		case *lnwire.ChannelUpdate:
			if passesFilter(msg.Timestamp) {
				msgsToSend = append(msgsToSend, msg)
			}

		// Similarly, we only send node announcements if the update
		// timestamp ifs between our set gossip filter time range.
		case *lnwire.NodeAnnouncement:
			if passesFilter(msg.Timestamp) {
				msgsToSend = append(msgsToSend, msg)
			}
		}
	}

	log.Tracef("GossipSyncer(%x): filtered gossip msgs: set=%v, sent=%v",
		g.cfg.peerPub[:], len(msgs), len(msgsToSend))

	if len(msgsToSend) == 0 {
		return
	}

	g.cfg.sendToPeer(msgsToSend...)
}
1366119
func (f *PeerEventsFilter) SentMessages() *PeerEventsFilter {
	f.eventType = simulations.EventTypeMsg
	b := false
	f.msgReceive = &b
	return f
}
1228292
func partitionKeys(k interface{}) ([]*rsa.PublicKey, []*ecdsa.PublicKey, [][]byte) {
	var (
		rsaKeys   []*rsa.PublicKey
		ecdsaKeys []*ecdsa.PublicKey
		hmacKeys  [][]byte
	)

	switch typed := k.(type) {
	case []byte:
		hmacKeys = append(hmacKeys, typed)
	case [][]byte:
		hmacKeys = typed
	case string:
		hmacKeys = append(hmacKeys, []byte(typed))
	case []string:
		for _, s := range typed {
			hmacKeys = append(hmacKeys, []byte(s))
		}
	case *rsa.PublicKey:
		rsaKeys = append(rsaKeys, typed)
	case []*rsa.PublicKey:
		rsaKeys = typed
	case *ecdsa.PublicKey:
		ecdsaKeys = append(ecdsaKeys, typed)
	case []*ecdsa.PublicKey:
		ecdsaKeys = typed
	}

	return rsaKeys, ecdsaKeys, hmacKeys
}
109457
func (p *Murmur2HashPartitioner) Partition(message *kafka.ProducerMessage, numPartitions int32) (int32, error) {
	if message.Key == nil {
		return p.random.Partition(message, numPartitions)
	}
	key, err := message.Key.Encode()
	if err != nil {
		return -1, err
	}
	// murmur2 implementation based on https://github.com/aappleby/smhasher/blob/master/src/MurmurHash2.cpp
	m := uint32(0x5bd1e995)
	r := uint32(24)
	seed := uint32(0x9747b28c)
	dataLen := uint32(len(key))
	/* Initialize the hash to a 'random' value */

	h := seed ^ dataLen

	/* Mix 4 bytes at a time into the hash */
	data := key
	i := 0
	for {
		if dataLen < 4 {
			break
		}
		k := *(*uint32)(unsafe.Pointer(&data[i*4]))
		k *= m
		k ^= k >> r
		k *= m

		h *= m
		h ^= k
		i++
		dataLen -= 4
	}

	/* Handle the last few bytes of the input array */

	switch dataLen {
	case 3:
		h ^= uint32(data[i*4+2]) << 16
		h ^= uint32(data[i*4+1]) << 8
		h ^= uint32(data[i*4])
		h *= m
	case 2:
		h ^= uint32(data[i*4+1]) << 8
		h ^= uint32(data[i*4])
		h *= m
	case 1:
		h ^= uint32(data[i*4])
		h *= m
	default:
	}

	/* Do a few final mixes of the hash to ensure the last few
	 * bytes are well-incorporated. */

	h ^= h >> 13
	h *= m
	h ^= h >> 15

	// convert h to positive
	partition := (int32(h) & 0x7fffffff) % numPartitions

	return partition, nil

}
305242
func (m *MemLogger) Messages() []LogEntry {
	if m.lock != nil {
		m.lock.Lock()
		defer m.lock.Unlock()
	}
	if m.data == nil || len(*m.data) == 0 {
		return nil
	}
	ret := make([]LogEntry, len(*m.data))
	copy(ret, *m.data)
	return ret
}
128635
func (s *Statement) SetMessages(v []*Message) *Statement {
	s.Messages = v
	return s
}
967237
func (mw *MirrorWriter) broadcastMessage(b []byte) bool {
	var dropped bool
	for i, w := range mw.writers {
		_, err := w.Write(b)
		if err != nil {
			mw.writers[i] = nil
			dropped = true
		}
	}
	return dropped
}
1027124
func (s *DescribeFlowOutput) SetMessages(v *Messages) *DescribeFlowOutput {
	s.Messages = v
	return s
}