// ScanShard loops over records on a specific shard, calls the callback func
// for each record and checkpoints the progress of scan.
773723
func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn ScanFunc) error {
	// get last seq number from checkpoint
	lastSeqNum, err := c.checkpoint.Get(c.streamName, shardID)
	if err != nil {
		return fmt.Errorf("get checkpoint error: %v", err)
	}

	// get shard iterator
	shardIterator, err := c.getShardIterator(c.streamName, shardID, lastSeqNum)
	if err != nil {
		return fmt.Errorf("get shard iterator error: %v", err)
	}

	c.logger.Log("[START]\t", shardID, lastSeqNum)
	defer func() {
		c.logger.Log("[STOP]\t", shardID)
	}()

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			resp, err := c.client.GetRecords(&kinesis.GetRecordsInput{
				ShardIterator: shardIterator,
			})

			// attempt to recover from GetRecords error by getting new shard iterator
			if err != nil {
				shardIterator, err = c.getShardIterator(c.streamName, shardID, lastSeqNum)
				if err != nil {
					return fmt.Errorf("get shard iterator error: %v", err)
				}
				continue
			}

			// loop over records, call callback func
			for _, r := range resp.Records {
				select {
				case <-ctx.Done():
					return nil
				default:
					err := fn(r)
					if err != nil && err != SkipCheckpoint {
						return err
					}

					if err != SkipCheckpoint {
						if err := c.checkpoint.Set(c.streamName, shardID, *r.SequenceNumber); err != nil {
							return err
						}
					}

					c.counter.Add("records", 1)
					lastSeqNum = *r.SequenceNumber
				}
			}

			if isShardClosed(resp.NextShardIterator, shardIterator) {
				c.logger.Log("[CLOSED]\t", shardID)
				return nil
			}

			shardIterator = resp.NextShardIterator
		}
	}
}
825392
func (m SessionMock) ScanIterator(statement string, arguments ...interface{}) Iterator {
	return m.Called(statement, arguments).Get(0).(Iterator)
}
1067554
func (c *Conn) GetCatShards() (shards CatShards) {
	shards = make(CatShards, 0)
	//force it to only respond with the columns we know about and in a forced order
	args := map[string]interface{}{"bytes": "b", "h": "index,shard,prirep,state,docs,store,ip,node"}
	s, err := c.DoCommand("GET", "/_cat/shards", args, nil)
	if err == nil {
		catShardLines := strings.Split(string(s[:]), "\n")
		for _, shardLine := range catShardLines {
			shard, _ := NewCatShardInfo(shardLine)
			if nil != shard {
				shards = append(shards, *shard)
			}
		}
	}
	return shards
}
766694
func (ts *Server) GetShardReplication(ctx context.Context, cell, keyspace, shard string) (*ShardReplicationInfo, error) {
	conn, err := ts.ConnForCell(ctx, cell)
	if err != nil {
		return nil, err
	}

	nodePath := path.Join(KeyspacesPath, keyspace, ShardsPath, shard, ShardReplicationFile)
	data, _, err := conn.Get(ctx, nodePath)
	if err != nil {
		return nil, err
	}

	sr := &topodatapb.ShardReplication{}
	if err = proto.Unmarshal(data, sr); err != nil {
		return nil, vterrors.Wrap(err, "bad ShardReplication data")
	}

	return NewShardReplicationInfo(sr, cell, keyspace, shard), nil
}
107370
func (v *RPC) GetShardKey() (o string) {
	if v != nil && v.ShardKey != nil {
		return *v.ShardKey
	}

	return
}
870841
func (r reedSolomon) Verify(shards [][]byte) (bool, error) {
	if len(shards) != r.Shards {
		return false, ErrTooFewShards
	}
	err := checkShards(shards, false)
	if err != nil {
		return false, err
	}

	// Slice of buffers being checked.
	toCheck := shards[r.DataShards:]

	// Do the checking.
	return r.checkSomeShards(r.parity, shards[0:r.DataShards], toCheck, r.ParityShards, len(shards[0])), nil
}
402617
func (c *cluster) MoveShard(ctx context.Context, col Collection, shard ShardID, fromServer, toServer ServerID) error {
	req, err := c.conn.NewRequest("POST", "_admin/cluster/moveShard")
	if err != nil {
		return WithStack(err)
	}
	input := moveShardRequest{
		Database:   col.Database().Name(),
		Collection: col.Name(),
		Shard:      shard,
		FromServer: fromServer,
		ToServer:   toServer,
	}
	if _, err := req.SetBody(input); err != nil {
		return WithStack(err)
	}
	applyContextSettings(ctx, req)
	resp, err := c.conn.Do(ctx, req)
	if err != nil {
		return WithStack(err)
	}
	if err := resp.CheckStatus(202); err != nil {
		return WithStack(err)
	}
	return nil
}
674161
func (m *BaseShardManager) getShardsForSentinelsLocked(
	keys []string) map[int]*ShardMapping {

	numShards := len(m.shardStates)
	results := make(map[int]*ShardMapping)

	for _, key := range keys {
		shardId := m.getShardId(key, numShards)

		entry, inMap := results[shardId]
		if !inMap {
			entry = &ShardMapping{}
			if shardId != -1 {
				state := m.shardStates[shardId]
				if state.State == ActiveServer ||
					state.State == WriteOnlyServer ||
					state.State == WarmUpServer {

					m.fillEntryWithConnection(state.Address, entry)

					// During WARM_UP state, we do try to write sentinels to
					// memcache but any failures are ignored. We run memcache
					// server in this mode for sometime to prime our memcache
					// and warm up memcache server.
					if state.State == WarmUpServer {
						entry.WarmingUp = true
					}
				} else {
					connSkippedByAddr.Add(state.Address, 1)
				}
			}
			entry.Keys = make([]string, 0, 1)
			results[shardId] = entry
		}
		entry.Keys = append(entry.Keys, key)
	}

	return results
}
1331547
func ScanTokens(data []byte, it func([]byte) bool) bool {
	lexer := &Scanner{data: data}

	var ok bool
	for lexer.Next() {
		switch lexer.Type() {
		case ItemToken:
			ok = true
			if !it(lexer.Bytes()) {
				return true
			}
		case ItemSeparator:
			if !isComma(lexer.Bytes()) {
				return false
			}
		default:
			return false
		}
	}

	return ok && !lexer.err
}
1189744
func (s *GitScanner) ScanAll(cb GitScannerFoundPointer) error {
	callback, err := firstGitScannerCallback(cb, s.FoundPointer)
	if err != nil {
		return err
	}

	opts := s.opts(ScanAllMode)
	opts.SkipDeletedBlobs = false
	return scanLeftRightToChan(s, callback, "", "", opts)
}
841733
func (m *BaseShardManager) GetShard(
	key string) (
	shardId int,
	conn net2.ManagedConn,
	err error) {

	m.rwMutex.RLock()
	defer m.rwMutex.RUnlock()

	shardId = m.getShardId(key, len(m.shardStates))
	if shardId == -1 {
		return
	}

	state := m.shardStates[shardId]
	if state.State != ActiveServer {
		m.logInfo("Memcache shard ", shardId, " is not in active state.")
		connSkippedByAddr.Add(state.Address, 1)
		return
	}

	entry := &ShardMapping{}
	m.fillEntryWithConnection(state.Address, entry)
	conn, err = entry.Connection, entry.ConnErr

	return
}
671644
func ParseShard(blob []byte) (Shard, error) {
	var sorted []string
	dec := gob.NewDecoder(bytes.NewReader(blob))
	if err := dec.Decode(&sorted); err != nil {
		return nil, err
	}
	shard := make(Shard, len(sorted))
	for _, blob := range sorted {
		shard[blob] = struct{}{}
	}
	return shard, nil
}
743710
func (m *BaseShardManager) GetShardsForKeys(
	keys []string) map[int]*ShardMapping {

	m.rwMutex.RLock()
	defer m.rwMutex.RUnlock()

	numShards := len(m.shardStates)
	results := make(map[int]*ShardMapping)

	for _, key := range keys {
		shardId := m.getShardId(key, numShards)

		entry, inMap := results[shardId]
		if !inMap {
			entry = &ShardMapping{}
			if shardId != -1 {
				state := m.shardStates[shardId]
				if state.State == ActiveServer {
					m.fillEntryWithConnection(state.Address, entry)
				} else {
					connSkippedByAddr.Add(state.Address, 1)
				}
			}
			entry.Keys = make([]string, 0, 1)
			results[shardId] = entry
		}
		entry.Keys = append(entry.Keys, key)
	}

	return results
}
589432
func (s *scanner) scanRbrack(r rune) bool {
	if s.mode&scanRbrack == 0 {
		return false
	}
	return r == '}'
}
1163187
func (d *Database) scanRow(data *structData, rows *sql.Rows, dst interface{}, columns []string) error {
	// check if there is data waiting
	if !rows.Next() {
		if err := rows.Err(); err != nil {
			return err
		}
		return sql.ErrNoRows
	}

	// get a list of targets
	targets, err := d.Targets(dst, columns)
	if err != nil {
		return err
	}

	// perform the scan
	if err := rows.Scan(targets...); err != nil {
		return err
	}

	// post-process and copy the target values into the struct
	if err := d.WriteTargets(dst, columns, targets); err != nil {
		return err
	}

	return rows.Err()
}
1311514
func (m *BaseShardManager) GetShardsForSentinelsFromKeys(
	keys []string) map[int]*ShardMapping {

	m.rwMutex.RLock()
	defer m.rwMutex.RUnlock()

	return m.getShardsForSentinelsLocked(keys)
}
823322
func (m *BaseShardManager) GetShardsForSentinelsFromItems(
	items []*Item) map[int]*ShardMapping {

	keys := make([]string, len(items))
	for i, item := range items {
		keys[i] = item.Key
	}

	m.rwMutex.RLock()
	defer m.rwMutex.RUnlock()

	numShards := len(m.shardStates)
	results := m.getShardsForSentinelsLocked(keys)

	// Now fill in all entries with items.
	for _, item := range items {
		shardId := m.getShardId(item.Key, numShards)

		entry := results[shardId]
		if len(entry.Items) == 0 {
			entry.Items = make([]*Item, 0, len(entry.Keys))
		}
		entry.Items = append(entry.Items, item)
	}

	return results
}