// enumerateQueuedBlobs yields blobs from the on-disk sorted.KeyValue store.
// This differs from enumeratePendingBlobs, which sends from the in-memory pending list.
4275
func (sh *SyncHandler) enumerateQueuedBlobs(dst chan<- blob.SizedRef, intr <-chan struct{}) error {
	defer close(dst)
	it := sh.queue.Find("", "")
	for it.Next() {
		br, ok := blob.Parse(it.Key())
		size, err := strconv.ParseUint(it.Value(), 10, 32)
		if !ok || err != nil {
			sh.logf("ERROR: bogus sync queue entry: %q => %q", it.Key(), it.Value())
			continue
		}
		select {
		case dst <- blob.SizedRef{Ref: br, Size: uint32(size)}:
		case <-intr:
			return it.Close()
		}
	}
	return it.Close()
}
1203035
func (s *VolumeServer) EnumerateWithFilters(
	ctx context.Context,
	req *api.SdkVolumeEnumerateWithFiltersRequest,
) (*api.SdkVolumeEnumerateWithFiltersResponse, error) {
	if s.cluster() == nil || s.driver(ctx) == nil {
		return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
	}

	var locator *api.VolumeLocator
	if len(req.GetName()) != 0 ||
		len(req.GetLabels()) != 0 ||
		req.GetOwnership() != nil {

		locator = &api.VolumeLocator{
			Name:         req.GetName(),
			VolumeLabels: req.GetLabels(),
			Ownership:    req.GetOwnership(),
		}
	}

	vols, err := s.driver(ctx).Enumerate(locator, nil)
	if err != nil {
		return nil, status.Errorf(
			codes.Internal,
			"Failed to enumerate volumes: %v",
			err.Error())
	}

	ids := make([]string, 0)
	for _, vol := range vols {
		// Check access
		if vol.IsPermitted(ctx, api.Ownership_Read) {
			ids = append(ids, vol.GetId())
		}
	}

	return &api.SdkVolumeEnumerateWithFiltersResponse{
		VolumeIds: ids,
	}, nil
}
294645
func (db *BytesConnectionRedis) ListKeys(match string) (keyval.BytesKeyIterator, error) {
	if db.closed {
		return nil, fmt.Errorf("ListKeys(%s) called on a closed connection", match)
	}
	return listKeys(db, match, nil, nil)
}
934984
func (c *asyncClient) Flushdb() (stat FutureBool, err Error) {
	resp, err := c.conn.QueueRequest(&FLUSHDB, [][]byte{})
	if err == nil {
		stat = resp.future.(FutureBool)
	}

	return
}
1166548
func (s *Set) Enumerate() []interface{} {
	// Lock set for read
	s.mutex.RLock()
	defer s.mutex.RUnlock()

	// Gather all values into a slice
	values := make([]interface{}, 0)
	for k := range s.m {
		values = append(values, k)
	}

	return values
}
485155
func (pq *PriorityQueue) Enqueue(priority uint8, value []byte) (*PriorityItem, error) {
	pq.Lock()
	defer pq.Unlock()

	// Check if queue is closed.
	if !pq.isOpen {
		return nil, ErrDBClosed
	}

	// Get the priorityLevel.
	level := pq.levels[priority]

	// Create new PriorityItem.
	item := &PriorityItem{
		ID:       level.tail + 1,
		Priority: priority,
		Key:      pq.generateKey(priority, level.tail+1),
		Value:    value,
	}

	// Add it to the priority queue.
	if err := pq.db.Put(item.Key, item.Value, nil); err != nil {
		return nil, err
	}

	// Increment tail position.
	level.tail++

	// If this priority level is more important than the curLevel.
	if pq.cmpAsc(priority) || pq.cmpDesc(priority) {
		pq.curLevel = priority
	}

	return item, nil
}
1198248
func (c *Corpus) EnumeratePermanodesLastModified(fn func(camtypes.BlobMeta) bool) {
	c.enumeratePermanodes(fn, c.permanodesByModtime.sorted(true))
}
803223
func (s *VolumeServer) SnapshotEnumerate(
	ctx context.Context,
	req *api.SdkVolumeSnapshotEnumerateRequest,
) (*api.SdkVolumeSnapshotEnumerateResponse, error) {
	if s.cluster() == nil || s.driver(ctx) == nil {
		return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
	}

	resp, err := s.SnapshotEnumerateWithFilters(
		ctx,
		&api.SdkVolumeSnapshotEnumerateWithFiltersRequest{
			VolumeId: req.GetVolumeId(),
		},
	)
	if err != nil {
		return nil, err
	}

	return &api.SdkVolumeSnapshotEnumerateResponse{
		VolumeSnapshotIds: resp.GetVolumeSnapshotIds(),
	}, nil

}
222299
func (r *SdkRoleManager) Enumerate(
	ctx context.Context,
	req *api.SdkRoleEnumerateRequest,
) (*api.SdkRoleEnumerateResponse, error) {
	keys, err := r.kv.Keys(rolePrefix, "/")
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to access roles from database: %v", err)
	}

	names := make([]string, len(keys))
	for i, key := range keys {
		names[i] = strings.TrimPrefix(key, rolePrefix+"/")
	}

	return &api.SdkRoleEnumerateResponse{
		Names: names,
	}, nil
}
157213
func (m *MockVolumeDriver) CloudBackupSchedEnumerate() (*api.CloudBackupSchedEnumerateResponse, error) {
	ret := m.ctrl.Call(m, "CloudBackupSchedEnumerate")
	ret0, _ := ret[0].(*api.CloudBackupSchedEnumerateResponse)
	ret1, _ := ret[1].(error)
	return ret0, ret1
}
212975
func (c *Client) EnumerateBlobsOpts(ctx context.Context, ch chan<- blob.SizedRef, opts EnumerateOpts) error {
	defer close(ch)
	if opts.After != "" && opts.MaxWait != 0 {
		return errors.New("client error: it's invalid to use enumerate After and MaxWaitSec together")
	}
	pfx, err := c.prefix()
	if err != nil {
		return err
	}

	error := func(msg string, e error) error {
		err := fmt.Errorf("client enumerate error: %s: %v", msg, e)
		c.printf("%v", err)
		return err
	}

	nSent := 0
	keepGoing := true
	after := opts.After
	for keepGoing {
		waitSec := 0
		if after == "" {
			if opts.MaxWait > 0 {
				waitSec = int(opts.MaxWait.Seconds())
				if waitSec == 0 {
					waitSec = 1
				}
			}
		}
		url_ := fmt.Sprintf("%s/camli/enumerate-blobs?after=%s&limit=%d&maxwaitsec=%d",
			pfx, url.QueryEscape(after), enumerateBatchSize, waitSec)
		req := c.newRequest(ctx, "GET", url_)
		resp, err := c.httpClient.Do(req)
		if err != nil {
			return error("http request", err)
		}

		json, err := c.responseJSONMap("enumerate-blobs", resp)
		if err != nil {
			return error("stat json parse error", err)
		}

		blobs, ok := getJSONMapArray(json, "blobs")
		if !ok {
			return error("response JSON didn't contain 'blobs' array", nil)
		}
		for _, v := range blobs {
			itemJSON, ok := v.(map[string]interface{})
			if !ok {
				return error("item in 'blobs' was malformed", nil)
			}
			blobrefStr, ok := getJSONMapString(itemJSON, "blobRef")
			if !ok {
				return error("item in 'blobs' was missing string 'blobRef'", nil)
			}
			size, ok := getJSONMapUint32(itemJSON, "size")
			if !ok {
				return error("item in 'blobs' was missing numeric 'size'", nil)
			}
			br, ok := blob.Parse(blobrefStr)
			if !ok {
				return error("item in 'blobs' had invalid blobref.", nil)
			}
			select {
			case ch <- blob.SizedRef{Ref: br, Size: uint32(size)}:
			case <-ctx.Done():
				return ctx.Err()
			}
			nSent++
			if opts.Limit == nSent {
				// nSent can't be zero at this point, so opts.Limit being 0
				// is okay.
				return nil
			}
		}

		after, keepGoing = getJSONMapString(json, "continueAfter")
	}
	return nil
}
1168887
func (q *ListQueue) Dequeue() (interface{}, error) {
	if q.IsEmpty() {
		q.last = nil
		return nil, errors.New("unable to dequeue element, queue is empty")
	}

	item := q.first.item
	q.first = q.first.next
	q.size--
	return item, nil
}
21355
func (m *MockCluster) Enumerate() (api.Cluster, error) {
	ret := m.ctrl.Call(m, "Enumerate")
	ret0, _ := ret[0].(api.Cluster)
	ret1, _ := ret[1].(error)
	return ret0, ret1
}
1158881
func EnumerateAll(ctx context.Context, src BlobEnumerator, fn func(blob.SizedRef) error) error {
	return EnumerateAllFrom(ctx, src, "", fn)
}
576501
func (ctrl *PersistentVolumeController) enqueueWork(queue workqueue.Interface, obj interface{}) {
	// Beware of "xxx deleted" events
	if unknown, ok := obj.(cache.DeletedFinalStateUnknown); ok && unknown.Obj != nil {
		obj = unknown.Obj
	}
	objName, err := controller.KeyFunc(obj)
	if err != nil {
		klog.Errorf("failed to get key from object: %v", err)
		return
	}
	klog.V(5).Infof("enqueued %q for sync", objName)
	queue.Add(objName)
}
362296
func (mr *MockClusterMockRecorder) EnumerateNodeConf() *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EnumerateNodeConf", reflect.TypeOf((*MockCluster)(nil).EnumerateNodeConf))
}
1244289
func (reg *registry) Enumerate(ctx context.Context, ingester func(string) error) error {
	root, err := pathFor(repositoriesRootPathSpec{})
	if err != nil {
		return err
	}

	err = reg.blobStore.driver.Walk(ctx, root, func(fileInfo driver.FileInfo) error {
		return handleRepository(fileInfo, root, "", ingester)
	})

	return err
}