// streamResult is a helper function to stream results that possibly
// contain non-fatal errors.  The helper function is allowed to panic
// on internal errors.
976308
func streamResult(procVal func(interface{}, io.Writer) nonFatalError) func(cmds.Response, cmds.ResponseEmitter) error {
	return func(res cmds.Response, re cmds.ResponseEmitter) (err error) {
		defer func() {
			if r := recover(); r != nil {
				err = fmt.Errorf("internal error: %v", r)
			}
			re.Close()
		}()

		var errors bool
		for {
			v, err := res.Next()
			if err != nil {
				if err == io.EOF {
					break
				}
				return err
			}

			errorMsg := procVal(v, os.Stdout)

			if errorMsg != "" {
				errors = true
				fmt.Fprintf(os.Stderr, "%s\n", errorMsg)
			}
		}

		if errors {
			return fmt.Errorf("errors while displaying some entries")
		}
		return nil
	}
}
537386
func (s *TasksService) FindResult(t TaskID) (*http.Response, error) {
	return s.client.GetResultByURI(t.ResultURI())
}
1341296
func StreamResponse(w http.ResponseWriter, data io.ReadCloser) {
	w.Header().Set("Content-Type", DefaultContentTypeV1_1)
	if _, err := copyBuf(w, data); err != nil {
		fmt.Printf("ERROR in stream: %v\n", err)
	}
	data.Close()
}
352543
func (c *ClientupdateCountCall) EventResult(eventResult string) *ClientupdateCountCall {
	c.urlParams_.Set("eventResult", eventResult)
	return c
}
1076936
func (adaptor DummyAdapter) SendResult(result api.RunnerResult) {
	str, jsonerr := jsonFromResult(result)
	if jsonerr != nil {
		fmt.Println(jsonerr)
		return
	}
	fmt.Println("\n" + str)
}
1257501
func encodeResult(call Call, result interface{}, wr io.Writer, indent bool) error {
	reply := Reply{
		Result: result,
	}
	return encodeReply(call, reply, wr, indent)
}
794216
func (qre *QueryExecutor) Stream(callback func(*sqltypes.Result) error) error {
	qre.logStats.OriginalSQL = qre.query
	qre.logStats.PlanType = qre.plan.PlanID.String()

	defer func(start time.Time) {
		tabletenv.QueryStats.Record(qre.plan.PlanID.String(), start)
		tabletenv.RecordUserQuery(qre.ctx, qre.plan.TableName(), "Stream", int64(time.Since(start)))
	}(time.Now())

	if err := qre.checkPermissions(); err != nil {
		return err
	}

	// if we have a transaction id, let's use the txPool for this query
	var conn *connpool.DBConn
	if qre.transactionID != 0 {
		txConn, err := qre.tsv.te.txPool.Get(qre.transactionID, "for streaming query")
		if err != nil {
			return err
		}
		defer txConn.Recycle()
		conn = txConn.DBConn
	} else {
		dbConn, err := qre.getStreamConn()
		if err != nil {
			return err
		}
		defer dbConn.Recycle()
		conn = dbConn
	}

	qd := NewQueryDetail(qre.logStats.Ctx, conn)
	qre.tsv.qe.streamQList.Add(qd)
	defer qre.tsv.qe.streamQList.Remove(qd)

	return qre.streamFetch(conn, qre.plan.FullQuery, qre.bindVars, "", callback)
}
875343
func (d *downloadTransfer) result() (layer.Layer, error) {
	return d.layer, d.err
}
352247
func (vtg *VTGate) StreamExecute(ctx context.Context, session *vtgatepb.Session, sql string, bindVariables map[string]*querypb.BindVariable, callback func(*sqltypes.Result) error) error {
	// In this context, we don't care if we can't fully parse destination
	destKeyspace, destTabletType, dest, _ := vtg.executor.ParseDestinationTarget(session.TargetString)
	statsKey := []string{"StreamExecute", destKeyspace, topoproto.TabletTypeLString(destTabletType)}

	defer vtg.timings.Record(statsKey, time.Now())

	var err error
	if bvErr := sqltypes.ValidateBindVariables(bindVariables); bvErr != nil {
		err = vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%v", bvErr)
		goto handleError
	}

	// TODO: This could be simplified to have a StreamExecute that takes
	// a destTarget without explicit destination.
	switch dest.(type) {
	case key.DestinationShard:
		err = vtg.resolver.StreamExecute(
			ctx,
			sql,
			bindVariables,
			destKeyspace,
			destTabletType,
			dest,
			session.Options,
			func(reply *sqltypes.Result) error {
				vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
				return callback(reply)
			})
	default:
		err = vtg.executor.StreamExecute(
			ctx,
			"StreamExecute",
			NewSafeSession(session),
			sql,
			bindVariables,
			querypb.Target{
				Keyspace:   destKeyspace,
				TabletType: destTabletType,
			},
			func(reply *sqltypes.Result) error {
				vtg.rowsReturned.Add(statsKey, int64(len(reply.Rows)))
				return callback(reply)
			})
	}
handleError:
	if err != nil {
		query := map[string]interface{}{
			"Sql":           sql,
			"BindVariables": bindVariables,
			"Session":       session,
		}
		return recordAndAnnotateError(err, statsKey, query, vtg.logStreamExecute)
	}
	return nil
}
565581
func (e *StreamAggExec) appendResult2Chunk(chk *chunk.Chunk) error {
	for i, aggFunc := range e.aggFuncs {
		err := aggFunc.AppendFinalResult2Chunk(e.ctx, e.partialResults[i], chk)
		if err != nil {
			return err
		}
		aggFunc.ResetPartialResult(e.partialResults[i])
	}
	if len(e.aggFuncs) == 0 {
		chk.SetNumVirtualRows(chk.NumRows() + 1)
	}
	return nil
}
48176
func Result(listener func(*vecty.Event)) *vecty.EventListener {
	return &vecty.EventListener{Name: "result", Listener: listener}
}
588253
func (r *ViewResults) Close() error {
	if r.streamResult == nil || r.streamResult.Closed() {
		return r.makeError()
	}

	err := r.streamResult.Close()
	if r.strace != nil {
		r.strace.Finish()
	}
	if r.ctx.Err() == context.DeadlineExceeded {
		return timeoutError{}
	}
	if vErr := r.makeError(); vErr != nil {
		return vErr
	}
	return err
}
1186875
func (dr *StreamReporter) Report(key, value string, resultType ValidationResultType) error {
	var c color
	switch resultType {
	case good:
		c = green
	case bad:
		c = red
	case warn:
		c = yellow
	default:
		c = white
	}
	if dr.WriteStream == nil {
		return errors.New("WriteStream has to be defined for this reporter")
	}

	fmt.Fprintf(dr.WriteStream, "%s: %s\n", colorize(key, white), colorize(value, c))
	return nil
}
282191
func NewResult(operation Operation, result Result, errMsg ...string) *OperationResult {
	msg := ""
	if len(errMsg) > 0 {
		msg = errMsg[0]
	}
	return &OperationResult{
		Operation:  operation,
		Result:     result,
		ErrMessage: msg,
	}
}
1326444
func handleProbeResult(d *Daemon, c *container.Container, result *types.HealthcheckResult, done chan struct{}) {
	c.Lock()
	defer c.Unlock()

	// probe may have been cancelled while waiting on lock. Ignore result then
	select {
	case <-done:
		return
	default:
	}

	retries := c.Config.Healthcheck.Retries
	if retries <= 0 {
		retries = defaultProbeRetries
	}

	h := c.State.Health
	oldStatus := h.Status()

	if len(h.Log) >= maxLogEntries {
		h.Log = append(h.Log[len(h.Log)+1-maxLogEntries:], result)
	} else {
		h.Log = append(h.Log, result)
	}

	if result.ExitCode == exitStatusHealthy {
		h.FailingStreak = 0
		h.SetStatus(types.Healthy)
	} else { // Failure (including invalid exit code)
		shouldIncrementStreak := true

		// If the container is starting (i.e. we never had a successful health check)
		// then we check if we are within the start period of the container in which
		// case we do not increment the failure streak.
		if h.Status() == types.Starting {
			startPeriod := timeoutWithDefault(c.Config.Healthcheck.StartPeriod, defaultStartPeriod)
			timeSinceStart := result.Start.Sub(c.State.StartedAt)

			// If still within the start period, then don't increment failing streak.
			if timeSinceStart < startPeriod {
				shouldIncrementStreak = false
			}
		}

		if shouldIncrementStreak {
			h.FailingStreak++

			if h.FailingStreak >= retries {
				h.SetStatus(types.Unhealthy)
			}
		}
		// Else we're starting or healthy. Stay in that state.
	}

	// replicate Health status changes
	if err := c.CheckpointTo(d.containersReplica); err != nil {
		// queries will be inconsistent until the next probe runs or other state mutations
		// checkpoint the container
		logrus.Errorf("Error replicating health state for container %s: %v", c.ID, err)
	}

	current := h.Status()
	if oldStatus != current {
		d.LogContainerEvent(c, "health_status: "+current)
	}
}
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
}
306453
func (jst *Tracer) GetResult() (json.RawMessage, error) {
	// Transform the context into a JavaScript object and inject into the state
	obj := jst.vm.PushObject()

	for key, val := range jst.ctx {
		switch val := val.(type) {
		case uint64:
			jst.vm.PushUint(uint(val))

		case string:
			jst.vm.PushString(val)

		case []byte:
			ptr := jst.vm.PushFixedBuffer(len(val))
			copy(makeSlice(ptr, uint(len(val))), val)

		case common.Address:
			ptr := jst.vm.PushFixedBuffer(20)
			copy(makeSlice(ptr, 20), val[:])

		case *big.Int:
			pushBigInt(val, jst.vm)

		default:
			panic(fmt.Sprintf("unsupported type: %T", val))
		}
		jst.vm.PutPropString(obj, key)
	}
	jst.vm.PutPropString(jst.stateObject, "ctx")

	// Finalize the trace and return the results
	result, err := jst.call("result", "ctx", "db")
	if err != nil {
		jst.err = wrapError("result", err)
	}
	// Clean up the JavaScript environment
	jst.vm.DestroyHeap()
	jst.vm.Destroy()

	return result, jst.err
}