// GetStreamResult implements AggregationFunction interface.
760366
func (sf *sumFunction) GetStreamResult() (d types.Datum) {
	if sf.streamCtx == nil {
		return
	}
	d = sf.streamCtx.Value
	sf.streamCtx = nil
	return
}
835739
func (d *ResourceDiff) get(addr []string, source string) getResult {
	result, err := d.multiReader.ReadFieldMerge(addr, source)
	if err != nil {
		panic(err)
	}

	return d.finalizeResult(addr, result)
}
1259514
func (reader PluginConfigReaderWithError) GetStreamArray(key string, defaultValue []MessageStreamID) ([]MessageStreamID, error) {
	key = reader.config.registerKey(key)
	if reader.HasValue(key) {
		values, err := reader.GetStringArray(key, []string{})
		if err != nil {
			return nil, err
		}
		streamArray := []MessageStreamID{}
		for _, streamName := range values {
			streamArray = append(streamArray, GetStreamID(streamName))
		}
		return streamArray, nil
	}

	return defaultValue, nil
}
91229
func (af *avgFunction) GetResult(evalCtx *AggEvaluateContext) (d types.Datum) {
	switch evalCtx.Value.Kind() {
	case types.KindFloat64:
		sum := evalCtx.Value.GetFloat64()
		d.SetFloat64(sum / float64(evalCtx.Count))
		return
	case types.KindMysqlDecimal:
		x := evalCtx.Value.GetMysqlDecimal()
		y := types.NewDecFromInt(evalCtx.Count)
		to := new(types.MyDecimal)
		err := types.DecimalDiv(x, y, to, types.DivFracIncr)
		terror.Log(err)
		frac := af.RetTp.Decimal
		if frac == -1 {
			frac = mysql.MaxDecimalScale
		}
		err = to.Round(to, mathutil.Min(frac, mysql.MaxDecimalScale), types.ModeHalfEven)
		terror.Log(err)
		d.SetMysqlDecimal(to)
	}
	return
}
205791
func (a *API) GetResult(domain, result int) (*Result, error) {
	dst := new(Result)
	url := a.BuildURL("/v2/domain/%d/result/%d", domain, result)
	err := a.Get(url, Params{}, dst)
	return dst, err
}
83802
func (h *Handler) HandleGetQueryResult(msg *pb.ChaincodeMessage, txContext *TransactionContext) (*pb.ChaincodeMessage, error) {
	iterID := h.UUIDGenerator.New()

	getQueryResult := &pb.GetQueryResult{}
	err := proto.Unmarshal(msg.Payload, getQueryResult)
	if err != nil {
		return nil, errors.Wrap(err, "unmarshal failed")
	}

	metadata, err := getQueryMetadataFromBytes(getQueryResult.Metadata)
	if err != nil {
		return nil, err
	}

	totalReturnLimit := calculateTotalReturnLimit(metadata)
	isPaginated := false

	var executeIter commonledger.ResultsIterator
	var paginationInfo map[string]interface{}

	chaincodeName := h.ChaincodeName()
	collection := getQueryResult.Collection
	if isCollectionSet(collection) {
		if txContext.IsInitTransaction {
			return nil, errors.New("private data APIs are not allowed in chaincode Init()")
		}
		if err := errorIfCreatorHasNoReadAccess(chaincodeName, collection, txContext); err != nil {
			return nil, err
		}
		executeIter, err = txContext.TXSimulator.ExecuteQueryOnPrivateData(chaincodeName, collection, getQueryResult.Query)
	} else if isMetadataSetForPagination(metadata) {
		paginationInfo, err = createPaginationInfoFromMetadata(metadata, totalReturnLimit, pb.ChaincodeMessage_GET_QUERY_RESULT)
		if err != nil {
			return nil, err
		}
		isPaginated = true
		executeIter, err = txContext.TXSimulator.ExecuteQueryWithMetadata(chaincodeName,
			getQueryResult.Query, paginationInfo)

	} else {
		executeIter, err = txContext.TXSimulator.ExecuteQuery(chaincodeName, getQueryResult.Query)
	}
	if err != nil {
		return nil, errors.WithStack(err)
	}

	txContext.InitializeQueryContext(iterID, executeIter)

	payload, err := h.QueryResponseBuilder.BuildQueryResponse(txContext, executeIter, iterID, isPaginated, totalReturnLimit)
	if err != nil {
		txContext.CleanupQueryContext(iterID)
		return nil, errors.WithStack(err)
	}

	payloadBytes, err := proto.Marshal(payload)
	if err != nil {
		txContext.CleanupQueryContext(iterID)
		return nil, errors.Wrap(err, "marshal failed")
	}

	chaincodeLogger.Debugf("Got keys and values. Sending %s", pb.ChaincodeMessage_RESPONSE)
	return &pb.ChaincodeMessage{Type: pb.ChaincodeMessage_RESPONSE, Payload: payloadBytes, Txid: msg.Txid, ChannelId: msg.ChannelId}, nil
}
209628
func (c *cmd) streamResults(doneCh chan struct{}, ackCh chan rExecAck, heartCh chan rExecHeart,
	outputCh chan rExecOutput, exitCh chan rExecExit, errCh chan struct{}) {
	kv := c.apiclient.KV()
	opts := api.QueryOptions{WaitTime: c.conf.wait}
	dir := path.Join(c.conf.prefix, c.sessionID) + "/"
	seen := make(map[string]struct{})

	for {
		// Check if we've been signaled to exit
		select {
		case <-doneCh:
			return
		default:
		}

		// Block on waiting for new keys
		keys, qm, err := kv.Keys(dir, "", &opts)
		if err != nil {
			c.UI.Error(fmt.Sprintf("Failed to read results: %s", err))
			goto ERR_EXIT
		}

		// Fast-path the no-change case
		if qm.LastIndex == opts.WaitIndex {
			continue
		}
		opts.WaitIndex = qm.LastIndex

		// Handle each key
		for _, key := range keys {
			// Ignore if we've seen it
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}

			// Trim the directory
			full := key
			key = strings.TrimPrefix(key, dir)

			// Handle the key type
			switch {
			case key == rExecFileName:
				continue
			case strings.HasSuffix(key, rExecAckSuffix):
				ackCh <- rExecAck{Node: strings.TrimSuffix(key, rExecAckSuffix)}

			case strings.HasSuffix(key, rExecExitSuffix):
				pair, _, err := kv.Get(full, nil)
				if err != nil || pair == nil {
					c.UI.Error(fmt.Sprintf("Failed to read key '%s': %v", full, err))
					continue
				}
				code, err := strconv.ParseInt(string(pair.Value), 10, 32)
				if err != nil {
					c.UI.Error(fmt.Sprintf("Failed to parse exit code '%s': %v", pair.Value, err))
					continue
				}
				exitCh <- rExecExit{
					Node: strings.TrimSuffix(key, rExecExitSuffix),
					Code: int(code),
				}

			case strings.LastIndex(key, rExecOutputDivider) != -1:
				pair, _, err := kv.Get(full, nil)
				if err != nil || pair == nil {
					c.UI.Error(fmt.Sprintf("Failed to read key '%s': %v", full, err))
					continue
				}
				idx := strings.LastIndex(key, rExecOutputDivider)
				node := key[:idx]
				if len(pair.Value) == 0 {
					heartCh <- rExecHeart{Node: node}
				} else {
					outputCh <- rExecOutput{Node: node, Output: pair.Value}
				}

			default:
				c.UI.Error(fmt.Sprintf("Unknown key '%s', ignoring.", key))
			}
		}
	}

ERR_EXIT:
	select {
	case errCh <- struct{}{}:
	default:
	}
}
46943
func (f *FakeLoggerEventStreamingClient) StreamResult(addr string, args []string) (logutil.EventStream, error) {
	f.mu.Lock()
	defer f.mu.Unlock()

	k := generateKey(args)
	result, ok := f.results[k]
	if !ok {
		return nil, fmt.Errorf("no response was registered for args: %v", args)
	}
	if result.addr != "" && addr != result.addr {
		return nil, fmt.Errorf("client sent request to wrong server address. got: %v want: %v", addr, result.addr)
	}
	result.count--
	if result.count == 0 {
		delete(f.results, k)
	}

	return &streamResultAdapter{
		lines: strings.Split(result.output, "\n"),
		index: 0,
		err:   result.err,
	}, nil
}
520776
func (reader PluginConfigReaderWithError) GetStreamMap(key string, defaultValue string) (map[MessageStreamID]string, error) {
	key = reader.config.registerKey(key)
	streamMap := make(map[MessageStreamID]string)
	if defaultValue != "" {
		streamMap[WildcardStreamID] = defaultValue
	}

	if reader.HasValue(key) {
		value, err := reader.config.Settings.StringMap(key)
		if err != nil {
			return nil, err
		}

		for streamName, target := range value {
			streamMap[GetStreamID(streamName)] = target
		}
	}

	return streamMap, nil
}
938477
func (c *Client) GetTaskResult(id int) []string {
	l, _ := c.GetTaskOutput(id, "result")
	return l
}
987535
func (h *httpStreamHandler) getStreamPair(requestID string) (*httpStreamPair, bool) {
	h.streamPairsLock.Lock()
	defer h.streamPairsLock.Unlock()

	if p, ok := h.streamPairs[requestID]; ok {
		klog.V(5).Infof("(conn=%p, request=%s) found existing stream pair", h.conn, requestID)
		return p, false
	}

	klog.V(5).Infof("(conn=%p, request=%s) creating new stream pair", h.conn, requestID)

	p := newPortForwardPair(requestID)
	h.streamPairs[requestID] = p

	return p, true
}
33413
func (cf *countFunction) GetStreamResult() (d types.Datum) {
	if cf.streamCtx == nil {
		return types.NewDatum(0)
	}
	d.SetInt64(cf.streamCtx.Count)
	cf.streamCtx = nil
	return
}
1337347
func getStreamLength(dataLen, chunkSize int64) int64 {
	if dataLen <= 0 {
		return 0
	}

	chunksCount := int64(dataLen / chunkSize)
	remainingBytes := int64(dataLen % chunkSize)
	streamLen := int64(0)
	streamLen += chunksCount * getSignedChunkLength(chunkSize)
	if remainingBytes > 0 {
		streamLen += getSignedChunkLength(remainingBytes)
	}
	streamLen += getSignedChunkLength(0)
	return streamLen
}
1212762
func (sf *sumFunction) GetGroupResult(groupKey []byte) (d types.Datum) {
	return sf.getContext(groupKey).Value
}
773601
func (e *SelectStatement) AggregateResult(output Record) error {
	for i, expr := range e.selectAST.Expression.Expressions {
		v, err := expr.evalNode(nil)
		if err != nil {
			return err
		}
		output.Set(fmt.Sprintf("_%d", i+1), v)
	}
	return nil
}
180515
func (cli *Client) StreamrouterStreamprocsGet(appname string) (*pb.StreamrouterStreamprocsReply, error) {
	req := &pb.StreamrouterStreamprocsRequest{Appname: appname}
	ctx, cancel := context.WithTimeout(context.Background(), cli.timeout)
	defer cancel()
	rpl, err := cli.streamrouterStreamprocsClient.Get(ctx, req)
	return rpl, err
}
1255777
func (cf *concatFunction) GetStreamResult() (d types.Datum) {
	if cf.streamCtx == nil {
		return
	}
	if cf.streamCtx.Buffer != nil {
		d.SetString(cf.streamCtx.Buffer.String())
	} else {
		d.SetNull()
	}
	cf.streamCtx = nil
	return
}