// MaxReceiveMessages indicates how many messages a receive operation should pull
// during any single call
181421
func MaxReceiveMessages(max int64) broker.SubscribeOption {
	return func(o *broker.SubscribeOptions) {
		if o.Context == nil {
			o.Context = context.Background()
		}
		o.Context = context.WithValue(o.Context, maxMessagesKey{}, max)
	}
}
32806
func (el *ChannelEventListener) RegReceivedMessages() *ChannelEventListener {
	el.ReceivedMessage = make(chan bundle.Bundle)
	el.ids = append(el.ids, el.em.RegisterListener(func(data interface{}) {
		el.ReceivedMessage <- data.(bundle.Bundle)
	}, poller.EventReceivedMessage))
	return el
}
584149
func (client *Client) GetNumberReplyMessages(date string) *GetNumberMessagesCall {
	return &GetNumberMessagesCall{
		c:            client,
		date:         date,
		deliveryType: DeliveryTypeReply,
	}
}
253497
func (ch *Channel) receiveReplyInternal(msg api.Message, expSeqNum uint16) (lastReplyReceived bool, err error) {
	if msg == nil {
		return false, errors.New("nil message passed in")
	}

	var ignore bool

	if vppReply := ch.delayedReply; vppReply != nil {
		// try the delayed reply
		ch.delayedReply = nil
		ignore, lastReplyReceived, err = ch.processReply(vppReply, expSeqNum, msg)
		if !ignore {
			return lastReplyReceived, err
		}
	}

	timer := time.NewTimer(ch.replyTimeout)
	for {
		select {
		// blocks until a reply comes to ReplyChan or until timeout expires
		case vppReply := <-ch.replyChan:
			ignore, lastReplyReceived, err = ch.processReply(vppReply, expSeqNum, msg)
			if ignore {
				logrus.Warnf("ignoring reply: %+v", vppReply)
				continue
			}
			return lastReplyReceived, err

		case <-timer.C:
			err = fmt.Errorf("no reply received within the timeout period %s", ch.replyTimeout)
			return false, err
		}
	}
	return
}
460588
func NewMessagesReceivedTotalStatsHandler(sub Subsystem, vec *prometheus.CounterVec) *MessagesReceivedTotalStatsHandler {
	return &MessagesReceivedTotalStatsHandler{
		baseStatsHandler: baseStatsHandler{
			subsystem: sub,
			collector: vec,
		},
		vec: vec,
	}
}
1173601
func (wc *WhisperClient) SetMaxMessageSize(ctx *Context, size int32) error {
	return wc.client.SetMaxMessageSize(ctx.context, uint32(size))
}
182208
func (v *Validator) Messages() []string {
	msgs := []string{}
	for _, vr := range v.results {
		msgs = append(msgs, vr.message)
	}
	return msgs
}
716214
func (bot *TgBot) ProcessMessages(messages []MessageWithUpdateID) {
	for _, msg := range messages {
		// if int64(msg.UpdateID) > bot.LastUpdateID {
		// 	// Add lock
		// 	bot.LastUpdateID = int64(msg.UpdateID)
		// }
		atomic.CompareAndSwapInt64(&bot.LastUpdateID, bot.LastUpdateID, int64(msg.UpdateID))
		if bot.MainListener != nil {
			bot.MainListener <- msg
		}
	}
}
1088025
func (q *queue) CancelReceipts(request *fetchRequest) {
	q.cancel(request, q.receiptTaskQueue, q.receiptPendPool)
}
270892
func ignoreReceives(ws *websocket.Conn, timeout time.Duration) {
	defer util.HandleCrash()
	var data []byte
	for {
		resetTimeout(ws, timeout)
		if err := websocket.Message.Receive(ws, &data); err != nil {
			return
		}
	}
}
327070
func (s *ChainService) AddBytesReceived(bytesReceived uint64) {
	atomic.AddUint64(&s.bytesReceived, bytesReceived)
}
906269
func (s *StanServer) sendAvailableMessagesToQueue(c *channel, qs *queueState) {
	if c == nil || qs == nil {
		return
	}

	qs.Lock()
	// Short circuit if no active members
	if len(qs.subs) == 0 {
		qs.Unlock()
		return
	}
	// If redelivery at startup in progress, don't attempt to deliver new messages
	if qs.newOnHold {
		qs.Unlock()
		return
	}
	for nextSeq := qs.lastSent + 1; qs.stalledSubCount < len(qs.subs); nextSeq++ {
		nextMsg := s.getNextMsg(c, &nextSeq, &qs.lastSent)
		if nextMsg == nil {
			break
		}
		if _, sent, sendMore := s.sendMsgToQueueGroup(qs, nextMsg, honorMaxInFlight); !sent || !sendMore {
			break
		}
	}
	qs.Unlock()
}
841335
func (c *Conversation) receiveUnit(m ValidMessage, forgetFragments bool) (plain MessagePlaintext, toSend []ValidMessage, err error) {
	message := makeCopy(m)
	defer wipeBytes(message)

	if !c.Policies.isOTREnabled() {
		return c.receiveWithoutOTR(message)
	}

	msgType := guessMessageType(message)
	var messagesToSend []messageWithHeader
	shouldForgetFragment := true
	switch msgType {
	case msgGuessError:
		return c.withInjectionsPlain(c.receiveErrorMessage(message))
	case msgGuessQuery:
		messagesToSend, err = c.receiveQueryMessage(message)
	case msgGuessTaggedPlaintext:
		plain, messagesToSend, err = c.receiveTaggedPlaintext(message)
	case msgGuessNotOTR:
		plain, messagesToSend, err = c.receivePlaintext(message)
	case msgGuessV1KeyExch:
		return nil, nil, errUnsupportedOTRVersion
	case msgGuessFragment:
		shouldForgetFragment = false
		c.fragmentationContext, err = c.receiveFragment(c.fragmentationContext, message)
		if fragmentsFinished(c.fragmentationContext) {
			return c.withInjectionsPlain(c.receiveUnit(c.fragmentationContext.frag, false))
		}
	case msgGuessUnknown:
		c.messageEvent(MessageEventReceivedMessageUnrecognized)
	case msgGuessDHCommit, msgGuessDHKey, msgGuessRevealSig, msgGuessSignature, msgGuessData:
		plain, messagesToSend, err = c.receiveEncoded(encodedMessage(message))
	}

	if shouldForgetFragment && forgetFragments {
		c.fragmentationContext = forgetFragment()
	}

	return c.withInjectionsPlain(c.toSendEncoded(plain, messagesToSend, err))
}
954828
func (handler *Handler) sendReceive(msg *pb.ChaincodeMessage, c chan pb.ChaincodeMessage) (pb.ChaincodeMessage, error) {
	errc := make(chan error, 1)
	handler.serialSendAsync(msg, errc)

	//the serialsend above will send an err or nil
	//the select filters that first error(or nil)
	//and continues to wait for the response
	//it is possible that the response triggers first
	//in which case the errc obviously worked and is
	//ignored
	for {
		select {
		case err := <-errc:
			if err == nil {
				continue
			}
			//would have been logged, return false
			return pb.ChaincodeMessage{}, err
		case outmsg, val := <-c:
			if !val {
				return pb.ChaincodeMessage{}, errors.New("unexpected failure on receive")
			}
			return outmsg, nil
		}
	}
}
466139
func (el *ChannelEventListener) RegReceivedDeposits() *ChannelEventListener {
	el.ReceivedDeposit = make(chan bundle.Bundle)
	el.ids = append(el.ids, el.em.RegisterListener(func(data interface{}) {
		el.ReceivedDeposit <- data.(bundle.Bundle)
	}, poller.EventReceivedDeposit))
	return el
}
819839
func (s *SampleChannelDataInput) SetMaxMessages(v int64) *SampleChannelDataInput {
	s.MaxMessages = &v
	return s
}
1174030
func MaxRecvMsgSize(s int) client.Option {
	return func(o *client.Options) {
		if o.Context == nil {
			o.Context = context.Background()
		}
		o.Context = context.WithValue(o.Context, maxRecvMsgSizeKey{}, s)
	}
}