// ReqMultiResp is the top level interface for the client for the few
// operations which have to hit the server multiple times to guarantee
// a complete response: List keys, Map Reduce, etc.
974701
func (c *Client) ReqMultiResp(reqstruct interface{}, structname string) (response interface{}, err error) {
	node, err := c.SelectNode()
	if err != nil {
		return nil, err
	}
	return node.ReqMultiResp(reqstruct, structname)
}
159883
func (rp *ReverseProxy) ServeHTTP(rw http.ResponseWriter, outreq *http.Request, respUpdateFn respUpdateFn) error {
	transport := rp.Transport
	if requestIsWebsocket(outreq) {
		transport = newConnHijackerTransport(transport)
	} else if transport == nil {
		transport = &http.Transport{
			Dial: rp.dialer.Dial,
		}
	}

	rp.Director(outreq)

	// Original incoming server request may be canceled by the
	// user or by std lib(e.g. too many idle connections).
	// Now we issue the new outgoing client request which
	// doesn't depend on the original one. (issue 1345)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	outreq = outreq.WithContext(ctx)

	res, err := transport.RoundTrip(outreq)
	if err != nil {
		return err
	}

	isWebsocket := res.StatusCode == http.StatusSwitchingProtocols && strings.ToLower(res.Header.Get("Upgrade")) == "websocket"

	// Remove hop-by-hop headers listed in the
	// "Connection" header of the response.
	if c := res.Header.Get("Connection"); c != "" {
		for _, f := range strings.Split(c, ",") {
			if f = strings.TrimSpace(f); f != "" {
				res.Header.Del(f)
			}
		}
	}

	for _, h := range hopHeaders {
		res.Header.Del(h)
	}

	if respUpdateFn != nil {
		respUpdateFn(res)
	}

	if isWebsocket {
		res.Body.Close()
		hj, ok := rw.(http.Hijacker)
		if !ok {
			panic(httpserver.NonHijackerError{Underlying: rw})
		}

		conn, brw, err := hj.Hijack()
		if err != nil {
			return err
		}
		defer conn.Close()

		var backendConn net.Conn
		if hj, ok := transport.(*connHijackerTransport); ok {
			backendConn = hj.Conn
			if _, err := conn.Write(hj.Replay); err != nil {
				return err
			}
			bufferPool.Put(hj.Replay)
		} else {
			backendConn, err = net.DialTimeout("tcp", outreq.URL.Host, rp.dialer.Timeout)
			if err != nil {
				return err
			}
			outreq.Write(backendConn)
		}
		defer backendConn.Close()

		// Proxy backend -> frontend.
		go pooledIoCopy(conn, backendConn)

		// Proxy frontend -> backend.
		//
		// NOTE: Hijack() sometimes returns buffered up bytes in brw which
		// would be lost if we didn't read them out manually below.
		if brw != nil {
			if n := brw.Reader.Buffered(); n > 0 {
				rbuf, err := brw.Reader.Peek(n)
				if err != nil {
					return err
				}
				backendConn.Write(rbuf)
			}
		}
		pooledIoCopy(backendConn, conn)
	} else {
		copyHeader(rw.Header(), res.Header)

		// The "Trailer" header isn't included in the Transport's response,
		// at least for *http.Transport. Build it up from Trailer.
		if len(res.Trailer) > 0 {
			trailerKeys := make([]string, 0, len(res.Trailer))
			for k := range res.Trailer {
				trailerKeys = append(trailerKeys, k)
			}
			rw.Header().Add("Trailer", strings.Join(trailerKeys, ", "))
		}

		rw.WriteHeader(res.StatusCode)
		if len(res.Trailer) > 0 {
			// Force chunking if we saw a response trailer.
			// This prevents net/http from calculating the length for short
			// bodies and adding a Content-Length.
			if fl, ok := rw.(http.Flusher); ok {
				fl.Flush()
			}
		}
		rp.copyResponse(rw, res.Body)
		res.Body.Close() // close now, instead of defer, to populate res.Trailer
		copyHeader(rw.Header(), res.Trailer)
	}

	return nil
}
928343
func (r *Reply) List() ([]string, error) {
	// Doing all this in two places instead of just calling ListBytes() so we don't have
	// to iterate twice
	if r.Type == ErrorReply {
		return nil, r.Err
	}
	if r.Type != MultiReply {
		return nil, errors.New("reply type is not MultiReply")
	}

	strings := make([]string, len(r.Elems))
	for i, v := range r.Elems {
		if v.Type == BulkReply {
			strings[i] = string(v.buf)
		} else if v.Type == NilReply {
			strings[i] = ""
		} else {
			return nil, errors.New("element type is not BulkReply or NilReply")
		}
	}

	return strings, nil
}
1205455
func (req *SrvReq) RespondRflush() {
	err := PackRflush(req.Rc)
	if err != nil {
		req.RespondError(err)
	} else {
		req.Respond()
	}
}
1312096
func (s *Server) globalRPC(method string, args interface{},
	reply structs.CompoundResponse) error {

	// Make a new request into each datacenter
	dcs := s.router.GetDatacenters()

	replies, total := 0, len(dcs)
	errorCh := make(chan error, total)
	respCh := make(chan interface{}, total)

	for _, dc := range dcs {
		go func(dc string) {
			rr := reply.New()
			if err := s.forwardDC(method, dc, args, &rr); err != nil {
				errorCh <- err
				return
			}
			respCh <- rr
		}(dc)
	}

	for replies < total {
		select {
		case err := <-errorCh:
			return err
		case rr := <-respCh:
			reply.Add(rr)
			replies++
		}
	}
	return nil
}
1315834
func (rm *retrieveManager) sendReq(reqID uint64, req *distReq, val validatorFunc) *sentReq {
	r := &sentReq{
		rm:       rm,
		req:      req,
		id:       reqID,
		sentTo:   make(map[distPeer]sentReqToPeer),
		stopCh:   make(chan struct{}),
		eventsCh: make(chan reqPeerEvent, 10),
		validate: val,
	}

	canSend := req.canSend
	req.canSend = func(p distPeer) bool {
		// add an extra check to canSend: the request has not been sent to the same peer before
		r.lock.RLock()
		_, sent := r.sentTo[p]
		r.lock.RUnlock()
		return !sent && canSend(p)
	}

	request := req.request
	req.request = func(p distPeer) func() {
		// before actually sending the request, put an entry into the sentTo map
		r.lock.Lock()
		r.sentTo[p] = sentReqToPeer{false, make(chan bool, 1)}
		r.lock.Unlock()
		return request(p)
	}
	rm.lock.Lock()
	rm.sentReqs[reqID] = r
	rm.lock.Unlock()

	go r.retrieveLoop()
	return r
}
346792
func NotMultipleOf(name, in string, multiple interface{}) *Validation {
	var msg string
	if in == "" {
		msg = fmt.Sprintf(multipleOfFailNoIn, name, multiple)
	} else {
		msg = fmt.Sprintf(multipleOfFail, name, in, multiple)
	}
	return &Validation{
		code:    MultipleOfFailCode,
		Name:    name,
		In:      in,
		Value:   multiple,
		message: msg,
	}
}
481391
func (consumer *Consumer) MultiGet(length int) ([]*Package, error) {
	var collection []*Package
	if consumer.HasUnacked() {
		return nil, fmt.Errorf("unacked Packages found")
	}

	// TODO maybe use transactions for rollback in case of errors?
	reqs, err := consumer.Queue.redisClient.Pipelined(func(c *redis.Pipeline) error {
		c.BRPopLPush(
			queueInputKey(consumer.Queue.Name),
			consumerWorkingQueueKey(consumer.Queue.Name, consumer.Name),
			0,
		)
		for i := 1; i < length; i++ {
			c.RPopLPush(
				queueInputKey(consumer.Queue.Name),
				consumerWorkingQueueKey(consumer.Queue.Name, consumer.Name),
			)
		}
		return nil
	})
	if err != nil && err != redis.Nil {
		return nil, err
	}

	for _, answer := range reqs {
		switch answer := answer.(type) {
		case *redis.StringCmd:
			if answer.Val() == "" {
				continue
			}
			p, err := consumer.parseRedisAnswer(answer)
			if err != nil {
				return nil, err
			}
			p.Collection = &collection
			collection = append(collection, p)
		default:
			return nil, err
		}
	}
	consumer.Queue.incrRate(
		consumerWorkingRateKey(consumer.Queue.Name, consumer.Name),
		int64(length),
	)

	return collection, nil
}
909804
func (c *Connector) MultiRemove(ctx context.Context, ei *dosa.EntityInfo, multiKeys []map[string]dosa.FieldValue) ([]error, error) {
	keyValues, err := multiKeyValuesToRPCValues(multiKeys)
	if err != nil {
		return nil, err
	}

	// perform the multi remove request
	request := &dosarpc.MultiRemoveRequest{
		Ref:       entityInfoToSchemaRef(ei),
		KeyValues: keyValues,
	}

	response, err := c.client.MultiRemove(ctx, request, getHeaders(c.headers)...)
	if err != nil {
		if !dosarpc.Dosa_MultiRemove_Helper.IsException(err) {
			return nil, errors.Wrap(err, "failed to MultiRemove due to network issue")
		}

		return nil, errors.Wrap(err, "failed to MultiRemove")
	}

	rpcErrors := response.Errors
	results := make([]error, len(rpcErrors))
	for i, rpcError := range rpcErrors {
		if rpcError != nil {
			results[i] = wrapIDLError(rpcError)
		}
	}

	return results, nil
}
1144016
func (c *RawBinaryClient) DeleteMulti(keys []string) []MutateResponse {
	if keys == nil {
		return nil
	}

	responses := make([]MutateResponse, len(keys), len(keys))

	c.mutex.Lock()
	defer c.mutex.Unlock()

	for i, key := range keys {
		responses[i] = c.sendDeleteRequest(key)
	}

	for i, key := range keys {
		if responses[i] != nil { // error occurred while sending
			continue
		}
		responses[i] = c.receiveMutateResponse(opDelete, key)
	}

	return responses
}
622847
func ParseMultiple(headers []string) Links {
	links := make(Links, 0)
	for _, header := range headers {
		links = append(links, Parse(header)...)
	}
	return links
}
574822
func (s *OrderService) CreateMulti(orders []SubmitOrder) (MultipleOrderResponse, error) {
	ordersMap := make([]interface{}, 0)
	for _, order := range orders {
		var side string
		if order.Amount < 0 {
			order.Amount = math.Abs(order.Amount)
			side = "sell"
		} else {
			side = "buy"
		}
		ordersMap = append(ordersMap, map[string]interface{}{
			"symbol":   order.Symbol,
			"amount":   strconv.FormatFloat(order.Amount, 'f', -1, 32),
			"price":    strconv.FormatFloat(order.Price, 'f', -1, 32),
			"exchange": "bitfinex",
			"side":     side,
			"type":     order.Type,
		})
	}

	payload := map[string]interface{}{
		"orders": ordersMap,
	}

	req, err := s.client.newAuthenticatedRequest("POST", "order/new/multi", payload)
	if err != nil {
		return MultipleOrderResponse{}, err
	}

	response := new(MultipleOrderResponse)
	_, err = s.client.do(req, response)

	return *response, err

}
932927
func (limit ResourceLimit) ListRequest() (ListCommand, error) {
	req := &ListResourceLimits{
		ResourceType:     limit.ResourceType,
		ResourceTypeName: limit.ResourceTypeName,
	}

	return req, nil
}
357634
func (c *Context) MultipartForm() (*multipart.Form, error) {
	err := c.Request.ParseMultipartForm(c.engine.MaxMultipartMemory)
	return c.Request.MultipartForm, err
}
989419
func RPC(ctx context.Context, routeStr string, reply proto.Message, arg proto.Message) error {
	return doSendRPC(ctx, "", routeStr, reply, arg)
}
761344
func (api *PrivateHostAPI) FilterMultiBy(key string, value interface{}) *PrivateHostAPI {
	api.filterBy(key, value, true)
	return api
}
788304
func hasMultipleServers(config clientcmdapi.Config) bool {
	server := ""
	for _, cluster := range config.Clusters {
		if len(server) == 0 {
			server = cluster.Server
		}
		if server != cluster.Server {
			return true
		}
	}
	return false
}