// MessagesFilter will return a slice of all Messages that match this lambda
// function, potentially including this message and messages contained within
// this message.
// This method DOES recurse into sub-messages and sub-parts.
337759
func (m *Message) MessagesFilter(filter func(*Message) bool) []*Message {

	messages := make([]*Message, 0, 1)
	if filter(m) {
		messages = append(messages, m)
	}

	if m.HasSubMessage() {
		return append(messages, m.SubMessage.MessagesFilter(filter)...)
	}

	if m.HasParts() {
		for _, part := range m.Parts {
			messages = append(messages, part.MessagesFilter(filter)...)
		}
	}
	return messages
}
540356
func (whisper *Whisper) GetFilter(id string) *Filter {
	return whisper.filters.Get(id)
}
155375
func (sc *Client) FilterMessages(ctx context.Context, id string) ([]*whisper.Message, error) {
	var messages []*whisper.Message
	return messages, sc.c.CallContext(ctx, &messages, "shh_getFilterMessages", id)
}
893699
func (f *DedupSQSFilter) Filter(msg *sqs.Message) bool {
	msgBody := aws.StringValue(msg.Body)
	if !f.cache.Set(msgBody, "") {
		logger.WithField("message", msgBody).Info("Removing a duplicate")
		return false
	}
	return true
}
296227
func (client *QueryClient) MessageStream(name string, callback func(*sqltypes.Result) error) (err error) {
	return client.server.MessageStream(client.ctx, &client.target, name, callback)
}
139325
func GetFilter(identifier string) *FilterInterface {
	filter, ok := filters[identifier]
	if ok {
		return filter
	}

	return nil
}
865441
func (filter *Stream) ApplyFilter(msg *core.Message) (core.FilterResult, error) {
	for _, blockedID := range filter.blacklist {
		if msg.GetStreamID() == blockedID {
			return filter.GetFilterResultMessageReject(), nil // ### return, explicitly blocked ###
		}
	}

	for _, allowedID := range filter.whitelist {
		if msg.GetStreamID() == allowedID {
			return core.FilterResultMessageAccept, nil // ### return, explicitly allowed ###
		}
	}

	// Return true if no whitlist is given, false otherwise (must fulfill whitelist)
	if len(filter.whitelist) > 0 {
		return filter.GetFilterResultMessageReject(), nil
	}

	return core.FilterResultMessageAccept, nil
}
686508
func Filters(f ...Filter) Modifier {
	return func(r *http.Request) error {
		for _, filter := range f {
			filter(r)
		}
		return nil // Or should filter return err?
	}
}
4161
func FilterByType(msgs []chat1.MessageUnboxed, query *chat1.GetThreadQuery, includeAllErrors bool) (res []chat1.MessageUnboxed) {
	useTypeFilter := (query != nil && len(query.MessageTypes) > 0)

	typmap := make(map[chat1.MessageType]bool)
	if useTypeFilter {
		for _, mt := range query.MessageTypes {
			typmap[mt] = true
		}
	}

	for _, msg := range msgs {
		state, err := msg.State()
		if err != nil {
			if includeAllErrors {
				res = append(res, msg)
			}
			continue
		}
		switch state {
		case chat1.MessageUnboxedState_ERROR:
			if includeAllErrors {
				res = append(res, msg)
			}
		case chat1.MessageUnboxedState_PLACEHOLDER:
			// We don't know what the type is for these, so just include them
			res = append(res, msg)
		default:
			_, match := typmap[msg.GetMessageType()]
			if !useTypeFilter || match {
				res = append(res, msg)
			}
		}
	}
	return res
}
520465
func (filter *Rate) ApplyFilter(msg *core.Message) (core.FilterResult, error) {
	filter.stateGuard.RLock()
	state, known := filter.state[msg.GetStreamID()]
	filter.stateGuard.RUnlock()

	// Add stream if necessary
	if !known {
		filter.stateGuard.Lock()
		state = &rateState{
			count:       new(int64),
			ignore:      false,
			lastReset:   time.Now(),
			metricLimit: metrics.NewCounter(),
		}
		streamName := core.StreamRegistry.GetStreamName(msg.GetStreamID())
		filter.state[msg.GetStreamID()] = state
		filter.metricsRegistry.Register(streamName, state.metricLimit)
		filter.stateGuard.Unlock()
	}

	// Ignore if set
	if state.ignore {
		return core.FilterResultMessageAccept, nil // ### return, do not limit ###
	}

	// Reset if necessary
	if time.Since(state.lastReset) > time.Second {
		state.lastReset = time.Now()
		atomic.SwapInt64(state.count, 0)
	}

	// Check if to be filtered
	if atomic.AddInt64(state.count, 1) <= filter.rateLimit {
		return core.FilterResultMessageAccept, nil // ### return, do not limit ###
	}

	state.metricLimit.Inc(1)
	return filter.GetFilterResultMessageReject(), nil
}
482763
func (api *PublicWhisperAPI) NewMessageFilter(req Criteria) (string, error) {
	var (
		src     *ecdsa.PublicKey
		keySym  []byte
		keyAsym *ecdsa.PrivateKey
		topics  [][]byte

		symKeyGiven  = len(req.SymKeyID) > 0
		asymKeyGiven = len(req.PrivateKeyID) > 0

		err error
	)

	// user must specify either a symmetric or an asymmetric key
	if (symKeyGiven && asymKeyGiven) || (!symKeyGiven && !asymKeyGiven) {
		return "", ErrSymAsym
	}

	if len(req.Sig) > 0 {
		if src, err = crypto.UnmarshalPubkey(req.Sig); err != nil {
			return "", ErrInvalidSigningPubKey
		}
	}

	if symKeyGiven {
		if keySym, err = api.w.GetSymKey(req.SymKeyID); err != nil {
			return "", err
		}
		if !validateDataIntegrity(keySym, aesKeyLength) {
			return "", ErrInvalidSymmetricKey
		}
	}

	if asymKeyGiven {
		if keyAsym, err = api.w.GetPrivateKey(req.PrivateKeyID); err != nil {
			return "", err
		}
	}

	if len(req.Topics) > 0 {
		topics = make([][]byte, len(req.Topics))
		for i, topic := range req.Topics {
			topics[i] = make([]byte, TopicLength)
			copy(topics[i], topic[:])
		}
	}

	f := &Filter{
		Src:      src,
		KeySym:   keySym,
		KeyAsym:  keyAsym,
		PoW:      req.MinPow,
		AllowP2P: req.AllowP2P,
		Topics:   topics,
		Messages: make(map[common.Hash]*ReceivedMessage),
	}

	id, err := api.w.Subscribe(f)
	if err != nil {
		return "", err
	}

	api.mu.Lock()
	api.lastUsed[id] = time.Now()
	api.mu.Unlock()

	return id, nil
}
1239953
func (msb *MessageBlock) Messages() []*MessageBlock {
	if msb.Msg.Set != nil {
		return msb.Msg.Set.Messages
	}
	return []*MessageBlock{msb}
}
464872
func (cli *Client) CreateFilter(filter json.RawMessage) (resp *RespCreateFilter, err error) {
	urlPath := cli.BuildURL("user", cli.UserID, "filter")
	_, err = cli.MakeRequest("POST", urlPath, &filter, &resp)
	return
}
635764
func (n *Notifier) AddFilter(fn func(*Notice) *Notice) {
	n.filters = append(n.filters, fn)
}
109050
func LifeFilter(err error) error {
	cause := errors.Cause(err)
	switch cause {
	case lifeflag.ErrNotFound:
		return ErrRemoved
	case lifeflag.ErrValueChanged:
		return dependency.ErrBounce
	}
	return err
}
960578
func NewMsgCFilter(filterType FilterType, blockHash *chainhash.Hash,
	data []byte) *MsgCFilter {
	return &MsgCFilter{
		FilterType: filterType,
		BlockHash:  *blockHash,
		Data:       data,
	}
}
1004797
func (n *Notifications) Filter(oldStatus, newStatus buildbucketpb.Status) Notifications {
	notifications := n.GetNotifications()
	filtered := make([]*Notification, 0, len(notifications))
	for _, notification := range notifications {
		if notification.ShouldNotify(oldStatus, newStatus) {
			filtered = append(filtered, notification)
		}
	}
	return Notifications{Notifications: filtered}
}