// readPump pumps messages from the websocket connection to the hub.
191959
func (c *wsConn) readPump() {
	defer func() {
		c.sh.wsHub.unregister <- c
		c.ws.Close()
	}()
	c.ws.SetReadLimit(maxMessageSize)
	c.ws.SetReadDeadline(time.Now().Add(pongWait))
	c.ws.SetPongHandler(func(string) error { c.ws.SetReadDeadline(time.Now().Add(pongWait)); return nil })
	for {
		_, message, err := c.ws.ReadMessage()
		if err != nil {
			break
		}
		if debug {
			log.Printf("websocket: got message %#q", message)
		}
		cm := new(wsClientMessage)
		if err := json.Unmarshal(message, cm); err != nil {
			log.Printf("Ignoring bogus websocket message. Err: %v", err)
			continue
		}
		c.sh.wsHub.watchReq <- watchReq{
			conn: c,
			tag:  cm.Tag,
			q:    cm.Query,
		}
	}
}
747262
func (a *AsyncConfig) ReadFrom(r io.Reader) (int64, error) {
	return encoding.ReadFrom(r, &a.PacketInMask,
		&a.PortStatusMask, &a.FlowRemovedMask)
}
378313
func (fc *Flow) Read(s Sourcer) (ret *Dataset) {
	return s.Generate(fc)
}
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
}
791542
func (m *MultipartReply) ReadFrom(r io.Reader) (int64, error) {
	return encoding.ReadFrom(r, &m.Type, &m.Flags, &defaultPad4)
}
268807
func ReadReply(p protocol.Protocol, r io.ReaderAt) (_ wire.Value, seqID int32, _ error) {
	envelope, err := p.DecodeEnveloped(r)
	if err != nil {
		return wire.Value{}, 0, err
	}

	switch {
	case envelope.Type == wire.Reply:
		return envelope.Value, envelope.SeqID, nil
	case envelope.Type != wire.Exception:
		return envelope.Value, envelope.SeqID, fmt.Errorf("unknown envelope type for reply, got %v", envelope.Type)
	}

	// Decode the exception payload.
	ex := &exception.TApplicationException{}
	if err := ex.FromWire(envelope.Value); err != nil {
		return envelope.Value, envelope.SeqID, fmt.Errorf("failed to decode exception: %v", err)
	}

	return envelope.Value, envelope.SeqID, ex
}
608669
func (s *Server) ReadSSFPacketSocket(serverConn net.PacketConn, packetPool *sync.Pool) {
	// TODO This is duplicated from ReadMetricSocket and feels like it could be it's
	// own function?
	p := packetPool.Get().([]byte)
	if len(p) == 0 {
		log.WithField("len", len(p)).Fatal(
			"packetPool making empty slices: trace_max_length_bytes must be >= 0")
	}
	packetPool.Put(p)

	for {
		buf := packetPool.Get().([]byte)
		n, _, err := serverConn.ReadFrom(buf)
		if err != nil {
			// In tests, the probably-best way to
			// terminate this reader is to issue a shutdown and close the listening
			// socket, which returns an error, so let's handle it here:
			select {
			case <-s.shutdown:
				log.WithError(err).Info("Ignoring ReadFrom error while shutting down")
				return
			default:
				log.WithError(err).Error("Error reading from UDP trace socket")
				continue
			}
		}

		s.HandleTracePacket(buf[:n])
		packetPool.Put(buf)
	}
}
842856
func (upc *UDPPacketConn) ReadFrom(b []byte) (int, net.Addr, error) {
	ipLen := IPv4MaximumHeaderSize
	udpLen := UDPMinimumSize

	for {
		pkt := make([]byte, ipLen+udpLen+len(b))
		n, _, err := upc.PacketConn.ReadFrom(pkt)
		if err != nil {
			return 0, nil, err
		}
		pkt = pkt[:n]
		buf := uio.NewBigEndianBuffer(pkt)

		// To read the header length, access data directly.
		ipHdr := IPv4(buf.Data())
		ipHdr = IPv4(buf.Consume(int(ipHdr.HeaderLength())))

		if ipHdr.TransportProtocol() != UDPProtocolNumber {
			continue
		}
		udpHdr := UDP(buf.Consume(udpLen))

		addr := &net.UDPAddr{
			IP:   net.IP(ipHdr.DestinationAddress()),
			Port: int(udpHdr.DestinationPort()),
		}
		if !udpMatch(addr, upc.boundAddr) {
			continue
		}
		return copy(b, buf.ReadAll()), addr, nil
	}
}
829848
func newReadPump(cnx redis.Conn) *readPump {
	return &readPump{
		cnx:    cnx,
		data:   make(chan interface{}),
		errs:   make(chan error),
		closer: make(chan struct{}),
	}
}
129402
func (a *ActionPushMPLS) ReadFrom(r io.Reader) (int64, error) {
	return encoding.ReadFrom(r, &defaultPad4, &a.EtherType, &defaultPad2)
}
680404
func (rwops *RWops) Read(buf []byte) (n int, err error) {
	return rwops.Read2(buf, 1, uint(len(buf)))
}
1233834
func ReadMessage(r io.Reader, msg proto.Message) error {
	return readProtoMsg(r, msg, maxMsgSize)
}
448593
func (c *Client) readListener() {
	for {
		select {
		case <-c.done:
			// Delete our connection
			c.hub.Unregister(c)

			// Tell writer to stop
			c.done <- true

			//exit processing loop
			return
		default:
			// Read from websocket
			var msg Message
			err := websocket.JSON.Receive(c.ws, &msg)

			switch {
			case err == io.EOF:
				// website connection closed
				c.done <- true
				return
			case err != nil:
				// Error, abort!
				c.done <- true
				return
			default:
				if msg.Event == "close" {
					c.done <- true
					return
				}
			}
		}
	}
}
787525
func readBulk(reader *bufio.Reader, head string) ([]byte, error) {
    var err error
    var data []byte

    if head == "" {
        head, err = reader.ReadString('\n')
        if err != nil {
            return nil, err
        }
    }
    switch head[0] {
    case ':':
        data = []byte(strings.TrimSpace(head[1:]))

    case '$':
        size, err := strconv.Atoi(strings.TrimSpace(head[1:]))
        if err != nil {
            return nil, err
        }
        if size == -1 {
            return nil, doesNotExist
        }
        lr := io.LimitReader(reader, int64(size))
        data, err = ioutil.ReadAll(lr)
        if err == nil {
            // read end of line
            _, err = reader.ReadString('\n')
        }
    default:
        return nil, RedisError("Expecting Prefix '$' or ':'")
    }

    return data, err
}
1290039
func (s *StreamingReader) Read(buf []byte) (int, error) {
	switch {
	// After the last chunk is read from underlying reader, we
	// never re-fill s.buf.
	case s.done:

	// s.buf will be (re-)filled with next chunk when has lesser
	// bytes than asked for.
	case s.buf.Len() < len(buf):
		s.chunkBufLen = 0
		for {
			n1, err := s.baseReadCloser.Read(s.chunkBuf[s.chunkBufLen:])
			// Usually we validate `err` first, but in this case
			// we are validating n > 0 for the following reasons.
			//
			// 1. n > 0, err is one of io.EOF, nil (near end of stream)
			// A Reader returning a non-zero number of bytes at the end
			// of the input stream may return either err == EOF or err == nil
			//
			// 2. n == 0, err is io.EOF (actual end of stream)
			//
			// Callers should always process the n > 0 bytes returned
			// before considering the error err.
			if n1 > 0 {
				s.chunkBufLen += n1
				s.bytesRead += int64(n1)

				if s.chunkBufLen == payloadChunkSize ||
					(s.chunkNum == s.totalChunks-1 &&
						s.chunkBufLen == s.lastChunkSize) {
					// Sign the chunk and write it to s.buf.
					s.signChunk(s.chunkBufLen)
					break
				}
			}
			if err != nil {
				if err == io.EOF {
					// No more data left in baseReader - last chunk.
					// Done reading the last chunk from baseReader.
					s.done = true

					// bytes read from baseReader different than
					// content length provided.
					if s.bytesRead != s.contentLen {
						return 0, io.ErrUnexpectedEOF
					}

					// Sign the chunk and write it to s.buf.
					s.signChunk(0)
					break
				}
				return 0, err
			}

		}
	}
	return s.buf.Read(buf)
}
950380
func readBuffer(copyTo chan<- []byte, socketRead io.ReadCloser, batchSize int, workSize int) {
	defer close(copyTo)
	for {
		batch := make([]byte, batchSize)
		size, err := socketRead.Read(batch)
		if err == nil && workSize != 0 && size%workSize != 0 {
			var lsize int
			lsize, err = io.ReadFull(
				socketRead,
				batch[size:size+workSize-(size%workSize)])
			size += lsize
		}
		if size > 0 {
			if err != nil && workSize != 0 {
				size -= (size % workSize)
			}
			copyTo <- batch[:size]
		}
		if err != nil {
			if err != io.EOF {
				log.Print("Error encountered in readBuffer:", err)
			}
			return
		}
	}
}
228037
func (t *Topic) messagePump() {
	var msg *Message
	var buf []byte
	var err error
	var chans []*Channel
	var memoryMsgChan chan *Message
	var backendChan chan []byte

	// do not pass messages before Start(), but avoid blocking Pause() or GetChannel()
	for {
		select {
		case <-t.channelUpdateChan:
			continue
		case <-t.pauseChan:
			continue
		case <-t.exitChan:
			goto exit
		case <-t.startChan:
		}
		break
	}
	t.RLock()
	for _, c := range t.channelMap {
		chans = append(chans, c)
	}
	t.RUnlock()
	if len(chans) > 0 && !t.IsPaused() {
		memoryMsgChan = t.memoryMsgChan
		backendChan = t.backend.ReadChan()
	}

	// main message loop
	for {
		select {
		case msg = <-memoryMsgChan:
		case buf = <-backendChan:
			msg, err = decodeMessage(buf)
			if err != nil {
				t.ctx.nsqd.logf(LOG_ERROR, "failed to decode message - %s", err)
				continue
			}
		case <-t.channelUpdateChan:
			chans = chans[:0]
			t.RLock()
			for _, c := range t.channelMap {
				chans = append(chans, c)
			}
			t.RUnlock()
			if len(chans) == 0 || t.IsPaused() {
				memoryMsgChan = nil
				backendChan = nil
			} else {
				memoryMsgChan = t.memoryMsgChan
				backendChan = t.backend.ReadChan()
			}
			continue
		case <-t.pauseChan:
			if len(chans) == 0 || t.IsPaused() {
				memoryMsgChan = nil
				backendChan = nil
			} else {
				memoryMsgChan = t.memoryMsgChan
				backendChan = t.backend.ReadChan()
			}
			continue
		case <-t.exitChan:
			goto exit
		}

		for i, channel := range chans {
			chanMsg := msg
			// copy the message because each channel
			// needs a unique instance but...
			// fastpath to avoid copy if its the first channel
			// (the topic already created the first copy)
			if i > 0 {
				chanMsg = NewMessage(msg.ID, msg.Body)
				chanMsg.Timestamp = msg.Timestamp
				chanMsg.deferred = msg.deferred
			}
			if chanMsg.deferred != 0 {
				channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
				continue
			}
			err := channel.PutMessage(chanMsg)
			if err != nil {
				t.ctx.nsqd.logf(LOG_ERROR,
					"TOPIC(%s) ERROR: failed to put msg(%s) to channel(%s) - %s",
					t.name, msg.ID, channel.name, err)
			}
		}
	}

exit:
	t.ctx.nsqd.logf(LOG_INFO, "TOPIC(%s): closing ... messagePump", t.name)
}