// processUnowned either reverse maps or validates the values for an unowned column.
7918
func (ins *Insert) processUnowned(vcursor VCursor, vindexColumnsKeys [][]sqltypes.Value, colVindex *vindexes.ColumnVindex, bv map[string]*querypb.BindVariable, ksids [][]byte) error {
	var reverseIndexes []int
	var reverseKsids [][]byte
	var verifyIndexes []int
	var verifyKeys []sqltypes.Value
	var verifyKsids [][]byte

	for rowNum, rowColumnKeys := range vindexColumnsKeys {
		// Right now, we only validate against the first column of a colvindex.
		// TODO(sougou): address this when we add multicolumn Map support.
		vindexKey := rowColumnKeys[0]
		if ksids[rowNum] == nil {
			continue
		}
		if vindexKey.IsNull() {
			reverseIndexes = append(reverseIndexes, rowNum)
			reverseKsids = append(reverseKsids, ksids[rowNum])
		} else {
			verifyIndexes = append(verifyIndexes, rowNum)
			verifyKeys = append(verifyKeys, vindexKey)
			verifyKsids = append(verifyKsids, ksids[rowNum])
		}
	}

	// For cases where a value was not supplied, we reverse map it
	// from the keyspace id, if possible.
	if reverseKsids != nil {
		reversible, ok := colVindex.Vindex.(vindexes.Reversible)
		if !ok {
			return fmt.Errorf("value must be supplied for column %v", colVindex.Columns)
		}
		reverseKeys, err := reversible.ReverseMap(vcursor, reverseKsids)
		if err != nil {
			return err
		}
		for i, reverseKey := range reverseKeys {
			rowNum := reverseIndexes[i]
			for colIdx, col := range colVindex.Columns {
				if colIdx == 0 {
					// Fill the first column with the reverse-mapped value.
					bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(reverseKey)
				} else {
					// Fill other columns with supplied values.
					bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexColumnsKeys[rowNum][colIdx])
				}
			}
		}
	}

	if verifyKsids != nil {
		// If values were supplied, we validate against keyspace id.
		verified, err := colVindex.Vindex.Verify(vcursor, verifyKeys, verifyKsids)
		if err != nil {
			return err
		}
		for i, v := range verified {
			rowNum := verifyIndexes[i]
			if !v {
				if ins.Opcode != InsertShardedIgnore {
					return fmt.Errorf("values %v for column %v does not map to keyspace ids", vindexColumnsKeys, colVindex.Columns)
				}
				// InsertShardedIgnore: skip the row.
				ksids[rowNum] = nil
				continue
			}
			for colIdx, col := range colVindex.Columns {
				bv[insertVarName(col, rowNum)] = sqltypes.ValueBindVariable(vindexColumnsKeys[rowNum][colIdx])
			}
		}
	}
	return nil
}
469307
func (e *Environment) ensurePrepared(c context.Context) {
	e.prepareOnce.Do(func() {
		e.processCacheData = caching.NewProcessCacheData()
		e.globalSettings = settings.New(gaesettings.Storage{})
		if e.Prepare != nil {
			e.Prepare(c)
		}
	})
}
181850
func (t *MailTarget) Process(e *Entry) {
	if t.Allow(e) {
		select {
		case t.entries <- e:
		default:
		}
	}
}
137558
func (deallocator *Deallocator) processService(ctx context.Context, service *api.Service) (bool, error) {
	if !service.PendingDelete {
		return false, nil
	}

	var (
		tasks []*api.Task
		err   error
	)

	deallocator.store.View(func(tx store.ReadTx) {
		tasks, err = store.FindTasks(tx, store.ByServiceID(service.ID))
	})

	if err != nil {
		log.G(ctx).WithError(err).Errorf("failed to retrieve the list of tasks for service %v", service.ID)
		// if in doubt, let's proceed to clean up the service anyway
		// better to clean up resources that shouldn't be cleaned up yet
		// than ending up with a service and some resources lost in limbo forever
		return true, deallocator.deallocateService(ctx, service)
	} else if len(tasks) == 0 {
		// no tasks remaining for this service, we can clean it up
		return true, deallocator.deallocateService(ctx, service)
	}
	deallocator.services[service.ID] = &serviceWithTaskCounts{service: service, taskCount: len(tasks)}
	return false, nil
}
439978
func (b *rescanBatch) done(err error) {
	for _, c := range b.errChans {
		c <- err
	}
}
833548
func (q *Type) Done(item interface{}) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	q.processing.delete(item)
	if q.dirty.has(item) {
		q.queue = append(q.queue, item)
		q.cond.Signal()
	}
}
1078780
func (q *NestedQuery) IgnoreUnmapped(value bool) *NestedQuery {
	q.ignoreUnmapped = &value
	return q
}
454389
func (f *FlagLoader) processField(fieldName string, field *structs.Field) error {
	if f.CamelCase {
		fieldName = strings.Join(camelcase.Split(fieldName), "-")
		fieldName = strings.Replace(fieldName, "---", "-", -1)
	}

	switch field.Kind() {
	case reflect.Struct:
		for _, ff := range field.Fields() {
			flagName := field.Name() + "-" + ff.Name()

			if f.Flatten {
				// first check if it's set or not, because if we have duplicate
				// we don't want to break the flag. Panic by giving a readable
				// output
				f.flagSet.VisitAll(func(fl *flag.Flag) {
					if strings.ToLower(ff.Name()) == fl.Name {
						// already defined
						panic(fmt.Sprintf("flag '%s' is already defined in outer struct", fl.Name))
					}
				})

				flagName = ff.Name()
			}

			if err := f.processField(flagName, ff); err != nil {
				return err
			}
		}
	default:
		// Add custom prefix to the flag if it's set
		if f.Prefix != "" {
			fieldName = f.Prefix + "-" + fieldName
		}

		// we only can get the value from expored fields, unexported fields panics
		if field.IsExported() {
			f.flagSet.Var(newFieldValue(field), flagName(fieldName), f.flagUsage(fieldName, field))
		}
	}

	return nil
}
124079
func (ev *ChangeEvent) Done(err error) {
	//TODO publish response to the topic
	if err != nil {
		logrus.DefaultLogger().Error(err)
	}
}
1368770
func (st *bundleSyncTask) processDeleted() (externalError bool, retriableError bool, e error) {
	if hasDeleteResourcesFinalizer(st.bundle) {
		if !resources.HasFinalizer(st.bundle, meta_v1.FinalizerDeleteDependents) {
			// If "foregroundDeletion" finalizer was not set, perform manual cascade deletion
			retrieable, err := st.deleteAllResources()
			if err != nil {
				return false, retrieable, err
			}
		}

		// If the "foregroundDeletion" finalizer is set, or the manual deletion
		// of resources has succeeded, remove the "deleteResources" finalizer
		st.newFinalizers = removeDeleteResourcesFinalizer(st.bundle.GetFinalizers())
	}
	return false, false, nil
}
1295306
func (op *RemoveUnitOperation) Done(err error) error {
	if err != nil {
		if !op.Force {
			return errors.Annotatef(err, "cannot remove unit %q", op.unit)
		}
		op.AddError(errors.Errorf("force removing unit %q proceeded despite encountering ERROR %v", op.unit, err))
	}
	return nil
}
177501
func ProcessBatchedData(socketRead io.ReadCloser, socketWrite io.Writer,
	makeProcessBatch func() (func([]byte) []byte, func([]byte, []byte)),
	batchSize int, workItemSize int) {
	readChan := make(chan []byte, 2)
	writeChan := make(chan []byte, 2)
	go readBatch(readChan, socketRead, batchSize, workItemSize)
	go writeBatch(writeChan, socketWrite)
	processBatch, prefetchBatch := makeProcessBatch()
	for buf := range readChan {
		res := processBatch(buf)
		writeChan <- res
		prefetchBatch(buf, res)
	}
	close(writeChan)
}
771826
func (nc *Conn) processPong() {
	var ch chan struct{}

	nc.mu.Lock()
	if len(nc.pongs) > 0 {
		ch = nc.pongs[0]
		nc.pongs = nc.pongs[1:]
	}
	nc.pout = 0
	nc.mu.Unlock()
	if ch != nil {
		ch <- struct{}{}
	}
}
1069660
func (b Backend) ProcessEvents(events ...*auditinternal.Event) bool {
	for _, event := range events {
		if event == nil {
			continue
		}
		attr, err := ev.NewAttributes(event)
		if err != nil {
			audit.HandlePluginError(PluginName, err, event)
			continue
		}
		level, stages := b.policyChecker.LevelAndStages(attr)
		if level == auditinternal.LevelNone {
			continue
		}
		// make shallow copy before modifying to satisfy interface definition
		ev := *event
		e, err := policy.EnforcePolicy(&ev, level, stages)
		if err != nil {
			audit.HandlePluginError(PluginName, err, event)
			continue
		}
		if e == nil {
			continue
		}
		b.delegateBackend.ProcessEvents(e)
	}
	// Returning true regardless of results, since dynamic audit backends
	// can never cause apiserver request to fail.
	return true
}
71602
func doneProcessing(w *responseWriter, ctx *RequestContext) bool {
	return w.Written() || ctx.err != nil || ctx.skip
}
161606
func (tpc *TwoPC) ReadAbandoned(ctx context.Context, abandonTime time.Time) (map[string]time.Time, error) {
	conn, err := tpc.readPool.Get(ctx)
	if err != nil {
		return nil, err
	}
	defer conn.Recycle()

	bindVars := map[string]*querypb.BindVariable{
		"time_created": sqltypes.Int64BindVariable(abandonTime.UnixNano()),
	}
	qr, err := tpc.read(ctx, conn, tpc.readAbandoned, bindVars)
	if err != nil {
		return nil, err
	}
	txs := make(map[string]time.Time, len(qr.Rows))
	for _, row := range qr.Rows {
		t, err := sqltypes.ToInt64(row[1])
		if err != nil {
			return nil, err
		}
		txs[row[0].ToString()] = time.Unix(0, t)
	}
	return txs, nil
}
640607
func (ctx *compctx) ProcessSassError(bs []byte) error {

	if len(bs) == 0 {
		return nil
	}

	err := json.Unmarshal(bs, &ctx.err)
	if err != nil {
		return err
	}

	errors := ctx.err
	ctx.errorString = fmt.Sprintf("Error > %s:%d\n%s",
		errors.File, errors.Line, errors.Message)
	return nil
}