// note: when using eofSignal to detect whether a message payload
// has been read, Read might not be called for zero sized messages.
167076
func (r *eofSignal) Read(buf []byte) (int, error) {
	if r.count == 0 {
		if r.eof != nil {
			r.eof <- struct{}{}
			r.eof = nil
		}
		return 0, io.EOF
	}

	max := len(buf)
	if int(r.count) < len(buf) {
		max = int(r.count)
	}
	n, err := r.wrapped.Read(buf[:max])
	r.count -= uint32(n)
	if (err != nil || r.count == 0) && r.eof != nil {
		r.eof <- struct{}{} // tell Peer that msg has been consumed
		r.eof = nil
	}
	return n, err
}
107630
func readToEOF(r io.Reader, buf []byte) (n int, err error) {
	for err == nil && n < len(buf) {
		var nn int
		nn, err = r.Read(buf[n:])
		n += nn
	}
	switch {
	case err == io.EOF:
		return n, nil
	case n == MaxBlockSize && err == nil:
		// This is paranoic, but some readers will return
		// quickly when passed a zero-length byte slice.
		var dummy [1]byte
		_, err = r.Read(dummy[:])
		if err == nil {
			return n, io.ErrShortBuffer
		}
		if err == io.EOF {
			err = nil
		}
	}
	return n, err
}
73432
func (b *Machine) ReadMessage(r io.Reader) ([]byte, error) {
	pktLen, err := b.ReadHeader(r)
	if err != nil {
		return nil, err
	}

	buf := make([]byte, pktLen)
	return b.ReadBody(r, buf)
}
433581
func (ec *EpollConsole) signalRead() {
	ec.readc.L.Lock()
	ec.readc.Signal()
	ec.readc.L.Unlock()
}
885081
func (es *EventStream) Notify(event string, bytes []byte) {
	// TODO reader?

	lines := strings.Split(string(bytes), "\n")

	data := ""
	for _, l := range lines {
		data += event + ": " + l + "\n"
	}

	sz := len(data) + 1
	size := fmt.Sprintf("%X", sz)

	for _, er := range es.receivers {
		go er.send(size, data)
	}
}
240538
func MaxMessageSize(max int) Opt {
	return func(r *reader) {
		buf := make([]byte, max>>1)
		r.Buffer(buf, max)
		r.maxf = max
	}
}
1011682
func (d *Decoder) Decode(msg *Message) (err error) {
	var length Integer
	err = binary.Read(d.R, binary.BigEndian, &length)
	if err != nil {
		if err != io.EOF {
			err = fmt.Errorf("error reading message length: %s", err)
		}
		return
	}
	if length > d.MaxLength {
		return errors.New("message too long")
	}
	if length == 0 {
		msg.Keepalive = true
		return
	}
	msg.Keepalive = false
	r := &io.LimitedReader{R: d.R, N: int64(length)}
	// Check that all of r was utilized.
	defer func() {
		if err != nil {
			return
		}
		if r.N != 0 {
			err = fmt.Errorf("%d bytes unused in message type %d", r.N, msg.Type)
		}
	}()
	msg.Keepalive = false
	c, err := readByte(r)
	if err != nil {
		return
	}
	msg.Type = MessageType(c)
	switch msg.Type {
	case Choke, Unchoke, Interested, NotInterested, HaveAll, HaveNone:
		return
	case Have, AllowedFast, Suggest:
		err = msg.Index.Read(r)
	case Request, Cancel, Reject:
		for _, data := range []*Integer{&msg.Index, &msg.Begin, &msg.Length} {
			err = data.Read(r)
			if err != nil {
				break
			}
		}
	case Bitfield:
		b := make([]byte, length-1)
		_, err = io.ReadFull(r, b)
		msg.Bitfield = unmarshalBitfield(b)
	case Piece:
		for _, pi := range []*Integer{&msg.Index, &msg.Begin} {
			err := pi.Read(r)
			if err != nil {
				return err
			}
		}
		dataLen := r.N
		msg.Piece = (*d.Pool.Get().(*[]byte))
		if int64(cap(msg.Piece)) < dataLen {
			return errors.New("piece data longer than expected")
		}
		msg.Piece = msg.Piece[:dataLen]
		_, err := io.ReadFull(r, msg.Piece)
		if err != nil {
			return errors.Wrap(err, "reading piece data")
		}
	case Extended:
		b, err := readByte(r)
		if err != nil {
			break
		}
		msg.ExtendedID = ExtensionNumber(b)
		msg.ExtendedPayload, err = ioutil.ReadAll(r)
	case Port:
		err = binary.Read(r, binary.BigEndian, &msg.Port)
	default:
		err = fmt.Errorf("unknown message type %#v", c)
	}
	return
}
1332460
func (c *Conn) ReadMessage() (messageType int, p []byte, err error) {
	var r io.Reader
	messageType, r, err = c.NextReader()
	if err != nil {
		return messageType, nil, err
	}
	p, err = ioutil.ReadAll(r)
	return messageType, p, err
}
569489
func (c *ServerContext) DecryptMessage(msg []byte, seqno uint32) (uint32, []byte, error) {
	return decryptMessage(c.sctxt, msg, seqno)
}
1064646
func (sp *ServerPeer) OnRead(_ *peer.Peer, bytesRead int, msg wire.Message,
	err error) {

	sp.server.AddBytesReceived(uint64(bytesRead))

	// Send a message to each subscriber. Each message gets its own
	// goroutine to prevent blocking on the mutex lock.
	// TODO: Flood control.
	sp.mtxSubscribers.RLock()
	defer sp.mtxSubscribers.RUnlock()
	for subscription := range sp.recvSubscribers {
		go func(subscription spMsgSubscription) {
			select {
			case <-subscription.quitChan:
			case subscription.msgChan <- spMsg{
				msg: msg,
				sp:  sp,
			}:
			}
		}(subscription)
	}
}
981551
func (r *ByteReader) ReadByte() (byte, error) {
	if r.off >= len(r.buf) {
		return 0, io.EOF
	}
	return r.readByte(), nil
}
1223957
func (udp *udpUnidlerSocket) readFromSock(buffer []byte, svcInfo *userspace.ServiceInfo) bool {
	if !svcInfo.IsAlive() {
		// The service port was closed or replaced.
		return false
	}

	// Block until data arrives.
	// TODO: Accumulate a histogram of n or something, to fine tune the buffer size.
	_, _, err := udp.ReadFrom(buffer)
	if err != nil {
		if e, ok := err.(net.Error); ok {
			if e.Temporary() {
				klog.V(1).Infof("ReadFrom had a temporary failure: %v", err)
				return true
			}
		}
		utilruntime.HandleError(fmt.Errorf("ReadFrom failed, exiting ProxyLoop: %v", err))
		return false
	}

	return true
}
1253349
func (msg *ReceivedMessage) ValidateAndParse() bool {
	end := len(msg.Raw)
	if end < 1 {
		return false
	}

	if isMessageSigned(msg.Raw[0]) {
		end -= signatureLength
		if end <= 1 {
			return false
		}
		msg.Signature = msg.Raw[end : end+signatureLength]
		msg.Src = msg.SigToPubKey()
		if msg.Src == nil {
			return false
		}
	}

	beg := 1
	payloadSize := 0
	sizeOfPayloadSizeField := int(msg.Raw[0] & SizeMask) // number of bytes indicating the size of payload
	if sizeOfPayloadSizeField != 0 {
		payloadSize = int(bytesToUintLittleEndian(msg.Raw[beg : beg+sizeOfPayloadSizeField]))
		if payloadSize+1 > end {
			return false
		}
		beg += sizeOfPayloadSizeField
		msg.Payload = msg.Raw[beg : beg+payloadSize]
	}

	beg += payloadSize
	msg.Padding = msg.Raw[beg:end]
	return true
}
255988
func ReadServerMessage(r io.Reader, m []Message) ([]Message, error) {
	return ReadMessage(r, ws.StateClientSide, m)
}
1226949
func (cr *s3ChunkedReader) Read(buf []byte) (n int, err error) {
	for {
		switch cr.state {
		case readChunkHeader:
			cr.readS3ChunkHeader()
			// If we're at the end of a chunk.
			if cr.n == 0 && cr.err == io.EOF {
				cr.state = readChunkTrailer
				cr.lastChunk = true
				continue
			}
			if cr.err != nil {
				return 0, cr.err
			}
			cr.state = readChunk
		case readChunkTrailer:
			cr.err = readCRLF(cr.reader)
			if cr.err != nil {
				return 0, errMalformedEncoding
			}
			cr.state = verifyChunk
		case readChunk:
			// There is no more space left in the request buffer.
			if len(buf) == 0 {
				return n, nil
			}
			rbuf := buf
			// The request buffer is larger than the current chunk size.
			// Read only the current chunk from the underlying reader.
			if uint64(len(rbuf)) > cr.n {
				rbuf = rbuf[:cr.n]
			}
			var n0 int
			n0, cr.err = cr.reader.Read(rbuf)
			if cr.err != nil {
				// We have lesser than chunk size advertised in chunkHeader, this is 'unexpected'.
				if cr.err == io.EOF {
					cr.err = io.ErrUnexpectedEOF
				}
				return 0, cr.err
			}

			// Calculate sha256.
			cr.chunkSHA256Writer.Write(rbuf[:n0])
			// Update the bytes read into request buffer so far.
			n += n0
			buf = buf[n0:]
			// Update bytes to be read of the current chunk before verifying chunk's signature.
			cr.n -= uint64(n0)

			// If we're at the end of a chunk.
			if cr.n == 0 {
				cr.state = readChunkTrailer
				continue
			}
		case verifyChunk:
			// Calculate the hashed chunk.
			hashedChunk := hex.EncodeToString(cr.chunkSHA256Writer.Sum(nil))
			// Calculate the chunk signature.
			newSignature := getChunkSignature(cr.cred, cr.seedSignature, cr.region, cr.seedDate, hashedChunk)
			if !compareSignatureV4(cr.chunkSignature, newSignature) {
				// Chunk signature doesn't match we return signature does not match.
				cr.err = errSignatureMismatch
				return 0, cr.err
			}
			// Newly calculated signature becomes the seed for the next chunk
			// this follows the chaining.
			cr.seedSignature = newSignature
			cr.chunkSHA256Writer.Reset()
			if cr.lastChunk {
				cr.state = eofChunk
			} else {
				cr.state = readChunkHeader
			}
		case eofChunk:
			return n, io.EOF
		}
	}
}
727321
func (l *tomlLexer) read() rune {
	r := l.peek()
	if r == '\n' {
		l.endbufferLine++
		l.endbufferCol = 1
	} else {
		l.endbufferCol++
	}
	l.inputIdx++
	return r
}
631112
func ReadMessage(c *messagebird.Client, messageID string) (*Message, error) {
	message := &Message{}
	if err := request(c, message, http.MethodGet, messagesPath+"/"+messageID, nil); err != nil {
		return nil, err
	}

	return message, nil
}