// MessageSetBySequenceNumber returns a slice of messages given a set of
// sequence number ranges
760199
func (m *DummyMailbox) MessageSetBySequenceNumber(set types.SequenceSet) []Message {
	var msgs []Message

	// If the mailbox is empty, return empty array
	if m.Messages() == 0 {
		return msgs
	}

	// For each sequence range in the sequence set
	for _, msgRange := range set {
		// If Min is "*", meaning the last message in the mailbox, Max should
		// always be Nil
		if msgRange.Min.Last() {
			// Return the last message in the mailbox
			msgs = append(msgs, m.MessageBySequenceNumber(m.Messages()))
			continue
		}

		start, err := msgRange.Min.Value()
		if err != nil {
			fmt.Printf("Error: %s\n", err.Error())
			return msgs
		}

		// If no Max is specified, the sequence number must be either a fixed
		// sequence number or
		if msgRange.Max.Nil() {
			var sequenceNo uint32
			// Fetch specific message by sequence number
			sequenceNo, err = msgRange.Min.Value()
			if err != nil {
				fmt.Printf("Error: %s\n", err.Error())
				return msgs
			}
			msg := m.MessageBySequenceNumber(sequenceNo)
			if msg != nil {
				msgs = append(msgs, msg)
			}
			continue
		}

		var end uint32
		if msgRange.Max.Last() {
			end = uint32(len(m.messages))
		} else {
			end, err = msgRange.Max.Value()
		}

		// Note this is very inefficient when
		// the message array is large. A proper
		// storage system using eg SQL might
		// instead perform a query here using
		// the range values instead.
		for seqNo := start; seqNo <= end; seqNo++ {
			msgs = append(msgs, m.MessageBySequenceNumber(seqNo))
		}
	}
	return msgs

}
1321402
func RandAlphaNumSeq(n int) string {
	b := make([]rune, n)
	for i := range b {
		b[i] = alphaNum[rand.Intn(len(alphaNum))]
	}
	return string(b)
}
252199
func findMessageByEnumName(s string) (*dotaMessage, bool) {
	for _, t := range messageTypes {
		if s == t.enumName {
			return t, true
		}
	}
	return nil, false
}
485783
func (s *BatchPutMessageInput) SetMessages(v []*Message) *BatchPutMessageInput {
	s.Messages = v
	return s
}
253359
func (l *Conn) nextMessageID() int64 {
	if l.chanMessageID != nil {
		if messageID, ok := <-l.chanMessageID; ok {
			return messageID
		}
	}
	return 0
}
629190
func lastSequenceIndex(str string) int {
	if len(str) == 0 {
		return -1
	}

	if len(str) == 1 {
		return 0
	}

	sequenceChar := str[0:1]
	lastPos := 0
	for i := 1; i < len(str); i++ {
		if str[i:i+1] != sequenceChar {
			break
		}

		lastPos = i
	}

	return lastPos
}
782163
func (chain *chainImpl) processMessagesToBlocks() ([]uint64, error) {
	counts := make([]uint64, 11) // For metrics and tests
	msg := new(ab.KafkaMessage)

	defer func() {
		// notify that we are not processing messages to blocks
		close(chain.doneProcessingMessagesToBlocks)
	}()

	defer func() { // When Halt() is called
		select {
		case <-chain.errorChan: // If already closed, don't do anything
		default:
			close(chain.errorChan)
		}
	}()

	subscription := fmt.Sprintf("added subscription to %s/%d", chain.channel.topic(), chain.channel.partition())
	var topicPartitionSubscriptionResumed <-chan string
	var deliverSessionTimer *time.Timer
	var deliverSessionTimedOut <-chan time.Time

	for {
		select {
		case <-chain.haltChan:
			logger.Warningf("[channel: %s] Consenter for channel exiting", chain.ChainID())
			counts[indexExitChanPass]++
			return counts, nil
		case kafkaErr := <-chain.channelConsumer.Errors():
			logger.Errorf("[channel: %s] Error during consumption: %s", chain.ChainID(), kafkaErr)
			counts[indexRecvError]++
			select {
			case <-chain.errorChan: // If already closed, don't do anything
			default:

				switch kafkaErr.Err {
				case sarama.ErrOffsetOutOfRange:
					// the kafka consumer will auto retry for all errors except for ErrOffsetOutOfRange
					logger.Errorf("[channel: %s] Unrecoverable error during consumption: %s", chain.ChainID(), kafkaErr)
					close(chain.errorChan)
				default:
					if topicPartitionSubscriptionResumed == nil {
						// register listener
						topicPartitionSubscriptionResumed = saramaLogger.NewListener(subscription)
						// start session timout timer
						deliverSessionTimer = time.NewTimer(chain.consenter.retryOptions().NetworkTimeouts.ReadTimeout)
						deliverSessionTimedOut = deliverSessionTimer.C
					}
				}
			}
			select {
			case <-chain.errorChan: // we are not ignoring the error
				logger.Warningf("[channel: %s] Closed the errorChan", chain.ChainID())
				// This covers the edge case where (1) a consumption error has
				// closed the errorChan and thus rendered the chain unavailable to
				// deliver clients, (2) we're already at the newest offset, and (3)
				// there are no new Broadcast requests coming in. In this case,
				// there is no trigger that can recreate the errorChan again and
				// mark the chain as available, so we have to force that trigger via
				// the emission of a CONNECT message. TODO Consider rate limiting
				go sendConnectMessage(chain.consenter.retryOptions(), chain.haltChan, chain.producer, chain.channel)
			default: // we are ignoring the error
				logger.Warningf("[channel: %s] Deliver sessions will be dropped if consumption errors continue.", chain.ChainID())
			}
		case <-topicPartitionSubscriptionResumed:
			// stop listening for subscription message
			saramaLogger.RemoveListener(subscription, topicPartitionSubscriptionResumed)
			// disable subscription event chan
			topicPartitionSubscriptionResumed = nil

			// stop timeout timer
			if !deliverSessionTimer.Stop() {
				<-deliverSessionTimer.C
			}
			logger.Warningf("[channel: %s] Consumption will resume.", chain.ChainID())

		case <-deliverSessionTimedOut:
			// stop listening for subscription message
			saramaLogger.RemoveListener(subscription, topicPartitionSubscriptionResumed)
			// disable subscription event chan
			topicPartitionSubscriptionResumed = nil

			close(chain.errorChan)
			logger.Warningf("[channel: %s] Closed the errorChan", chain.ChainID())

			// make chain available again via CONNECT message trigger
			go sendConnectMessage(chain.consenter.retryOptions(), chain.haltChan, chain.producer, chain.channel)

		case in, ok := <-chain.channelConsumer.Messages():
			if !ok {
				logger.Criticalf("[channel: %s] Kafka consumer closed.", chain.ChainID())
				return counts, nil
			}

			// catch the possibility that we missed a topic subscription event before
			// we registered the event listener
			if topicPartitionSubscriptionResumed != nil {
				// stop listening for subscription message
				saramaLogger.RemoveListener(subscription, topicPartitionSubscriptionResumed)
				// disable subscription event chan
				topicPartitionSubscriptionResumed = nil
				// stop timeout timer
				if !deliverSessionTimer.Stop() {
					<-deliverSessionTimer.C
				}
			}

			select {
			case <-chain.errorChan: // If this channel was closed...
				chain.errorChan = make(chan struct{}) // ...make a new one.
				logger.Infof("[channel: %s] Marked consenter as available again", chain.ChainID())
			default:
			}
			if err := proto.Unmarshal(in.Value, msg); err != nil {
				// This shouldn't happen, it should be filtered at ingress
				logger.Criticalf("[channel: %s] Unable to unmarshal consumed message = %s", chain.ChainID(), err)
				counts[indexUnmarshalError]++
				continue
			} else {
				logger.Debugf("[channel: %s] Successfully unmarshalled consumed message, offset is %d. Inspecting type...", chain.ChainID(), in.Offset)
				counts[indexRecvPass]++
			}
			switch msg.Type.(type) {
			case *ab.KafkaMessage_Connect:
				_ = chain.processConnect(chain.ChainID())
				counts[indexProcessConnectPass]++
			case *ab.KafkaMessage_TimeToCut:
				if err := chain.processTimeToCut(msg.GetTimeToCut(), in.Offset); err != nil {
					logger.Warningf("[channel: %s] %s", chain.ChainID(), err)
					logger.Criticalf("[channel: %s] Consenter for channel exiting", chain.ChainID())
					counts[indexProcessTimeToCutError]++
					return counts, err // TODO Revisit whether we should indeed stop processing the chain at this point
				}
				counts[indexProcessTimeToCutPass]++
			case *ab.KafkaMessage_Regular:
				if err := chain.processRegular(msg.GetRegular(), in.Offset); err != nil {
					logger.Warningf("[channel: %s] Error when processing incoming message of type REGULAR = %s", chain.ChainID(), err)
					counts[indexProcessRegularError]++
				} else {
					counts[indexProcessRegularPass]++
				}
			}
		case <-chain.timer:
			if err := sendTimeToCut(chain.producer, chain.channel, chain.lastCutBlockNumber+1, &chain.timer); err != nil {
				logger.Errorf("[channel: %s] cannot post time-to-cut message = %s", chain.ChainID(), err)
				// Do not return though
				counts[indexSendTimeToCutError]++
			} else {
				counts[indexSendTimeToCutPass]++
			}
		}
	}
}
543892
func (s *RateIncreaseCriteria) SetNumberOfNotifiedThings(v int64) *RateIncreaseCriteria {
	s.NumberOfNotifiedThings = &v
	return s
}
1303194
func (s *StreamRecord) SetSequenceNumber(v string) *StreamRecord {
	s.SequenceNumber = &v
	return s
}
1059192
func (lm *logMatcher) FailureMessage(actual interface{}) (message string) {
	return fmt.Sprintf(
		"Expected\n\t%s\nto contain log sequence \n\t%s",
		format.Object(lm.actual, 0),
		format.Object(lm.expected, 0),
	)
}
1125569
func (db *DB) minSeq() uint64 {
	db.snapsMu.Lock()
	defer db.snapsMu.Unlock()

	if e := db.snapsList.Front(); e != nil {
		return e.Value.(*snapshotElement).seq
	}

	return db.getSeq()
}
386832
func parseSeq(v string) (s Seq, err error) {
	if sep := strings.IndexRune(v, ':'); sep < 0 {
		s.Start, err = parseSeqNumber(v)
		s.Stop = s.Start
		return
	} else if s.Start, err = parseSeqNumber(v[:sep]); err == nil {
		if s.Stop, err = parseSeqNumber(v[sep+1:]); err == nil {
			if (s.Stop < s.Start && s.Stop != 0) || s.Start == 0 {
				s.Start, s.Stop = s.Stop, s.Start
			}
			return
		}
	}
	return s, ErrBadSeqSet(v)
}
328354
func (s *sequence) fromByteArray(data []byte) error {
	l := len(data)
	if l%12 != 0 {
		return fmt.Errorf("cannot deserialize byte sequence of length %d (%v)", l, data)
	}

	p := s
	i := 0
	for {
		p.block = binary.BigEndian.Uint32(data[i : i+4])
		p.count = binary.BigEndian.Uint64(data[i+4 : i+12])
		i += 12
		if i == l {
			break
		}
		p.next = &sequence{}
		p = p.next
	}

	return nil
}
958200
func (client *Client) GetNumberPushMessages(date string) *GetNumberMessagesCall {
	return &GetNumberMessagesCall{
		c:            client,
		date:         date,
		deliveryType: DeliveryTypePush,
	}
}
931511
func (s *SequenceNumberRange) SetStartingSequenceNumber(v string) *SequenceNumberRange {
	s.StartingSequenceNumber = &v
	return s
}
730513
func (ms *FileMsgStore) GetSequenceFromTimestamp(timestamp int64) (uint64, error) {
	ms.RLock()
	defer ms.RUnlock()

	// No message ever stored
	if ms.first == 0 {
		return 0, nil
	}
	// All messages have expired
	if ms.first > ms.last {
		return ms.last + 1, nil
	}
	// If we have some state, try to quickly get the sequence
	if ms.firstMsg != nil && timestamp <= ms.firstMsg.Timestamp {
		return ms.first, nil
	}
	if ms.lastMsg != nil {
		if timestamp == ms.lastMsg.Timestamp {
			return ms.last, nil
		}
		if timestamp > ms.lastMsg.Timestamp {
			return ms.last + 1, nil
		}
	}
	// This will require disk access.
	for i := ms.firstFSlSeq; i <= ms.lastFSlSeq; i++ {
		// support possible missing slices
		slice := ms.files[i]
		if slice == nil {
			continue
		}
		if err := ms.lockIndexFile(slice); err != nil {
			return 0, err
		}
		seq := slice.firstSeq
		if firstMsgInSlice, err := ms.getMsgIndex(slice, seq); err != nil {
			ms.unlockIndexFile(slice)
			return 0, err
		} else if timestamp > firstMsgInSlice.timestamp {
			seq = slice.lastSeq
			lastMsgInSlice, err := ms.getMsgIndex(slice, seq)
			if err != nil {
				ms.unlockIndexFile(slice)
				return 0, err
			}
			if timestamp > lastMsgInSlice.timestamp {
				// Not there, move to the next slice.
				ms.unlockIndexFile(slice)
				continue
			}
			// It may be equal, so search only if strictly lower
			if timestamp < lastMsgInSlice.timestamp {
				// We know that the timestamp is somewhere in this slice.

				// Could do binary search, but will be probably more efficient
				// to do sequential disk reads. The index records are small,
				// so read of a record will probably bring many consecutive ones
				// in the system's disk cache, resulting in memory-only access
				// for the following indexes...
				for seq = slice.firstSeq + 1; seq <= slice.lastSeq-1; seq++ {
					mindex, err := ms.getMsgIndex(slice, seq)
					if err != nil {
						ms.unlockIndexFile(slice)
						return 0, err
					}
					if mindex.timestamp >= timestamp {
						break
					}
				}
			}
		}
		ms.unlockIndexFile(slice)
		// We are here if the timestamp is smaller than the first
		// message in the first slice, or we have found the first
		// sequence that is >= timestamp.
		return seq, nil
	}
	return ms.last + 1, nil
}
1268173
func MessageFromBytes(msgBytes []byte) (msg RFC2822Message, err error) {
	// The header and body are separated by a double new-line
	splitMessage := bytes.SplitN(msgBytes, []byte("\r\n\r\n"), 2)

	// Read the headers
	headerReader := textproto.NewReader(bufio.NewReader(bytes.NewReader(splitMessage[0])))
	msg.Headers, err = headerReader.ReadMIMEHeader()

	if err != nil && err != io.EOF {
		return msg, err
	}

	if len(splitMessage) == 2 {
		msg.Body = string(splitMessage[1])
	}

	return msg, nil
}