// recv decodes and handles RPC responses. Responses to RPC requests
// are forwarded to the pending call, if any. RPC Notifications are
// forwarded by calling notify, synchronously.
173250
func (c *Conn) recv(notify func(string, []byte), done func(error)) {
	var resp Response
	var err error
	for {
		resp.reset()
		if err = c.codec.ReadResponse(&resp); err != nil {
			done(err)
			return
		}

		// Check if this is an RPC notification from the server.
		if resp.Method != "" {
			// Method represents the event that was triggered over the
			// Chrome DevTools Protocol. We do not expect to receive
			// RPC requests, if this was one, the ID field would be set.
			notify(resp.Method, resp.Args)
			continue
		}

		c.mu.Lock()
		call := c.pending[resp.ID]
		delete(c.pending, resp.ID)
		c.mu.Unlock()

		switch {
		case call == nil:
			// No pending call, this could mean there was an error during
			// send or the server sent an unexpected response.
			if enableDebug {
				log.Println("rpcc: no pending call: " + resp.String())
			}
		case resp.Error != nil:
			call.done(resp.Error)
		default:
			var err error
			if call.Reply != nil {
				if err = json.Unmarshal(resp.Result, call.Reply); err != nil {
					err = fmt.Errorf("rpcc: decoding %s: %s", call.Method, err.Error())
				}
			}
			call.done(err)
		}
	}
}
1085550
func (c *Chan) RecvChan() <-chan interface{} {
	c.mutx.Lock()
	defer c.mutx.Unlock()
	if c.rcvch == nil {
		c.rcvch = make(chan interface{})
		go c.receiveLoop()
	}
	return c.rcvch
}
889893
func Receive(conn *Conn) (r *Reply, err error) {
	conn.Lock()
	defer func() {
		if err != nil {
			conn.state = connStateNotConnected
			conn.Unlock()
			conn.fail()
		} else {
			conn.Unlock()
		}
	}()
	if conn.RequestTimeout != 0 {
		conn.tcpConn.SetReadDeadline(time.Now().Add(conn.RequestTimeout))
	}
	return readReply(conn)
}
1114904
func (iface *Intface) Recv() (msg []string, err error) {
	msg, err = iface.pipe.RecvMessage(0)
	return
}
45208
func (a *Agent) RPC(method string, args interface{}, reply interface{}) error {
	if a.server != nil {
		return a.server.RPC(method, args, reply)
	}
	return a.client.RPC(method, args, reply)
}
1261468
func (resp *CopStreamResponse) Recv() (*coprocessor.Response, error) {
	deadline := time.Now().Add(resp.Timeout).UnixNano()
	atomic.StoreInt64(&resp.Lease.deadline, deadline)

	ret, err := resp.Tikv_CoprocessorStreamClient.Recv()

	atomic.StoreInt64(&resp.Lease.deadline, 0) // Stop the lease check.
	return ret, errors.Trace(err)
}
362226
func (s *gRPCBrokerServer) Recv() (*plugin.ConnInfo, error) {
	select {
	case <-s.quit:
		return nil, errors.New("broker closed")
	case i := <-s.recv:
		return i, nil
	}
}
1170138
func (_m *StepStream) Recv() (*orderer.StepRequest, error) {
	ret := _m.Called()

	var r0 *orderer.StepRequest
	if rf, ok := ret.Get(0).(func() *orderer.StepRequest); ok {
		r0 = rf()
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*orderer.StepRequest)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func() error); ok {
		r1 = rf()
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}
1210246
func (f *v3KeyRangeFilter) Recv() (*sqltypes.Result, error) {
	r, err := f.input.Recv()
	if err != nil {
		return nil, err
	}

	rows := make([][]sqltypes.Value, 0, len(r.Rows))
	for _, row := range r.Rows {
		ksid, err := f.resolver.keyspaceID(row)
		if err != nil {
			return nil, err
		}

		if key.KeyRangeContains(f.keyRange, ksid) {
			rows = append(rows, row)
		}
	}
	r.Rows = rows
	return r, nil
}
685524
func (t *TcpTransport) SendRecv(message *proto.Msg) (*proto.Msg, error) {
	response_ch := make(chan response)
	t.requestQueue <- request{message, response_ch}
	r := <-response_ch
	return r.message, r.err
}
306281
func (cmd *Command) RPC(httpc *http.Client, ctx *Context) error {
	addr := "http://rpc/" + ctx.Command().pathWithSep("/")
	method := "POST"
	if cmd.HTTPMethods != nil && len(cmd.HTTPMethods) > 0 {
		method = cmd.HTTPMethods[0]
	}
	var body io.Reader
	if values := ctx.FormValues(); values != nil {
		body = strings.NewReader(values.Encode())
	}
	req, err := http.NewRequest(method, addr, body)
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	req.Header.Set("User-Agent", "cli-RPC")
	resp, err := httpc.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	_, err = io.Copy(ctx, resp.Body)
	return err
}
1158091
func (r *rpcHandler) forward(method string, info structs.RPCInfo, args interface{}, reply interface{}) (bool, error) {
	var firstCheck time.Time

	region := info.RequestRegion()
	if region == "" {
		return true, fmt.Errorf("missing target RPC")
	}

	// Handle region forwarding
	if region != r.config.Region {
		// Mark that we are forwarding the RPC
		info.SetForwarded()
		err := r.forwardRegion(region, method, args, reply)
		return true, err
	}

	// Check if we can allow a stale read
	if info.IsRead() && info.AllowStaleRead() {
		return false, nil
	}

CHECK_LEADER:
	// Find the leader
	isLeader, remoteServer := r.getLeader()

	// Handle the case we are the leader
	if isLeader {
		return false, nil
	}

	// Handle the case of a known leader
	if remoteServer != nil {
		// Mark that we are forwarding the RPC
		info.SetForwarded()
		err := r.forwardLeader(remoteServer, method, args, reply)
		return true, err
	}

	// Gate the request until there is a leader
	if firstCheck.IsZero() {
		firstCheck = time.Now()
	}
	if time.Now().Sub(firstCheck) < r.config.RPCHoldTimeout {
		jitter := lib.RandomStagger(r.config.RPCHoldTimeout / structs.JitterFraction)
		select {
		case <-time.After(jitter):
			goto CHECK_LEADER
		case <-r.shutdownCh:
		}
	}

	// No leader found and hold time exceeded
	return true, structs.ErrNoLeader
}
772102
func (s *Stream) RecvCompress() string {
	if err := s.waitOnHeader(); err != nil {
		return ""
	}
	return s.recvCompress
}
102562
func (soc *Socket) Recv(flags Flag) (string, error) {
	b, _, err := soc.RecvBytesWithOpts(flags)
	return string(b), err
}
313311
func (soc *Socket) RecvBytesWithOpts(flags Flag, options ...interface{}) ([]byte, []interface{}, error) {
	opts := make([]interface{}, len(options))

	if !soc.opened {
		return []byte{}, opts, ErrorSocketClosed
	}
	var msg C.zmq_msg_t
	if i, err := C.zmq_msg_init(&msg); i != 0 {
		return []byte{}, opts, errget(err)
	}
	defer C.zmq_msg_close(&msg)

	size, err := C.zmq_msg_recv(&msg, soc.soc, C.int(flags))
	if size < 0 {
		return []byte{}, opts, errget(err)
	}
	if size == 0 {
		return []byte{}, opts, nil
	}
	data := make([]byte, int(size))
	C.zmq4_memcpy(unsafe.Pointer(&data[0]), C.zmq_msg_data(&msg), C.size_t(size))

	for i, option := range options {
		switch option.(type) {
		case OptRoutingId:
			if !has42draft {
				return []byte{}, opts, ErrorNotImplemented42draft
			}
			opts[i] = OptRoutingId(uint32(C.zmq_msg_routing_id(&msg)))
		case OptGroup:
			if !has42draft {
				return []byte{}, opts, ErrorNotImplemented42draft
			}
			opts[i] = OptGroup(C.GoString(C.zmq_msg_group(&msg)))
		default:
			return []byte{}, opts, ErrorNotImplemented42
		}
	}

	return data, opts, nil
}
1160067
func (cn *conn) recvMessage(r *readBuf) (byte, error) {
	// workaround for a QueryRow bug, see exec
	if cn.saveMessageType != 0 {
		t := cn.saveMessageType
		*r = cn.saveMessageBuffer
		cn.saveMessageType = 0
		cn.saveMessageBuffer = nil
		return t, nil
	}

	x := cn.scratch[:5]
	_, err := io.ReadFull(cn.buf, x)
	if err != nil {
		return 0, err
	}

	// read the type and length of the message that follows
	t := x[0]
	n := int(binary.BigEndian.Uint32(x[1:])) - 4
	var y []byte
	if n <= len(cn.scratch) {
		y = cn.scratch[:n]
	} else {
		y = make([]byte, n)
	}
	_, err = io.ReadFull(cn.buf, y)
	if err != nil {
		return 0, err
	}
	*r = y
	return t, nil
}
802205
func (rw *RetryWatcher) receive() {
	defer close(rw.doneChan)
	defer close(rw.resultChan)

	klog.V(4).Info("Starting RetryWatcher.")
	defer klog.V(4).Info("Stopping RetryWatcher.")

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		select {
		case <-rw.stopChan:
			cancel()
			return
		case <-ctx.Done():
			return
		}
	}()

	// We use non sliding until so we don't introduce delays on happy path when WATCH call
	// timeouts or gets closed and we need to reestablish it while also avoiding hot loops.
	wait.NonSlidingUntilWithContext(ctx, func(ctx context.Context) {
		done, retryAfter := rw.doReceive()
		if done {
			cancel()
			return
		}

		time.Sleep(retryAfter)

		klog.V(4).Infof("Restarting RetryWatcher at RV=%q", rw.lastResourceVersion)
	}, rw.minRestartDelay)
}