stream.go
func (st *Stream) produceKVs(ctx context.Context) error {
	var size int
	var txn *Txn
	if st.readTs > 0 {
		txn = st.db.NewTransactionAt(st.readTs, false)
	} else {
		txn = st.db.NewTransaction(false)
	}
	defer txn.Discard()

	iterate := func(kr keyRange) error {
		iterOpts := DefaultIteratorOptions
		iterOpts.AllVersions = true
		iterOpts.Prefix = st.Prefix
		iterOpts.PrefetchValues = false
		itr := txn.NewIterator(iterOpts)
		defer itr.Close()

		outList := new(pb.KVList)
		var prevKey []byte
		for itr.Seek(kr.left); itr.Valid(); {
			// it.Valid would only return true for keys with the provided Prefix in iterOpts.
			item := itr.Item()
			if bytes.Equal(item.Key(), prevKey) {
				itr.Next()
				continue
			}
			prevKey = append(prevKey[:0], item.Key()...)

			// Check if we reached the end of the key range.
			if len(kr.right) > 0 && bytes.Compare(item.Key(), kr.right) >= 0 {
				break
			}
			// Check if we should pick this key.
			if st.ChooseKey != nil && !st.ChooseKey(item) {
				continue
			}

			// Now convert to key value.
			list, err := st.KeyToList(item.KeyCopy(nil), itr)
			if err != nil {
				return err
			}
			if list == nil || len(list.Kv) == 0 {
				continue
			}
			outList.Kv = append(outList.Kv, list.Kv...)
			size += list.Size()
			if size >= pageSize {
				st.kvChan <- outList
				outList = new(pb.KVList)
				size = 0
			}
		}
		if len(outList.Kv) > 0 {
			st.kvChan <- outList
		}
		return nil
	}

	for {
		select {
		case kr, ok := <-st.rangeCh:
			if !ok {
				// Done with the keys.
				return nil
			}
			if err := iterate(kr); err != nil {
				return err
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}
func (st *Stream) produceKVs(ctx context.Context) error {
	var size int
	var txn *Txn
	if st.readTs > 0 {
		txn = st.db.NewTransactionAt(st.readTs, false)
	} else {
		txn = st.db.NewTransaction(false)
	}
	defer txn.Discard()

	iterate := func(kr keyRange) error {
		iterOpts := DefaultIteratorOptions
		iterOpts.AllVersions = true
		iterOpts.Prefix = st.Prefix
		iterOpts.PrefetchValues = false
		itr := txn.NewIterator(iterOpts)
		defer itr.Close()

		outList := new(pb.KVList)
		var prevKey []byte
		for itr.Seek(kr.left); itr.Valid(); {
			// it.Valid would only return true for keys with the provided Prefix in iterOpts.
			item := itr.Item()
			if bytes.Equal(item.Key(), prevKey) {
				itr.Next()
				continue
			}
			prevKey = append(prevKey[:0], item.Key()...)

			// Check if we reached the end of the key range.
			if len(kr.right) > 0 && bytes.Compare(item.Key(), kr.right) >= 0 {
				break
			}
			// Check if we should pick this key.
			if st.ChooseKey != nil && !st.ChooseKey(item) {
				continue
			}

			// Now convert to key value.
			list, err := st.KeyToList(item.KeyCopy(nil), itr)
			if err != nil {
				return err
			}
			if list == nil || len(list.Kv) == 0 {
				continue
			}
			outList.Kv = append(outList.Kv, list.Kv...)
			size += list.Size()
			if size >= pageSize {
				st.kvChan <- outList
				outList = new(pb.KVList)
				size = 0
			}
		}
		if len(outList.Kv) > 0 {
			st.kvChan <- outList
		}
		return nil
	}

	for {
		select {
		case kr, ok := <-st.rangeCh:
			if !ok {
				// Done with the keys.
				return nil
			}
			if err := iterate(kr); err != nil {
				return err
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}
func ( st * Stream ) produceKVs ( ctx context . Context ) error { var size int var txn * Txn if st . readTs > 0 { txn = st . db . NewTransactionAt ( st . readTs , false ) } else { txn = st . db . NewTransaction ( false ) } defer txn . Discard ( ) iterate := func ( kr keyRange ) error { iterOpts := DefaultIteratorOptions iterOpts . AllVersions = true iterOpts . Prefix = st . Prefix iterOpts . PrefetchValues = false itr := txn . NewIterator ( iterOpts ) defer itr . Close ( ) outList := new ( pb . KVList ) var prevKey [ ] byte for itr . Seek ( kr . left ) ; itr . Valid ( ) ; { // it.Valid would only return true for keys with the provided Prefix in iterOpts. item := itr . Item ( ) if bytes . Equal ( item . Key ( ) , prevKey ) { itr . Next ( ) continue } prevKey = append ( prevKey [ : 0 ] , item . Key ( ) ... ) // Check if we reached the end of the key range. if len ( kr . right ) > 0 && bytes . Compare ( item . Key ( ) , kr . right ) >= 0 { break } // Check if we should pick this key. if st . ChooseKey != nil && ! st . ChooseKey ( item ) { continue } // Now convert to key value. list , err := st . KeyToList ( item . KeyCopy ( nil ) , itr ) if err != nil { return err } if list == nil || len ( list . Kv ) == 0 { continue } outList . Kv = append ( outList . Kv , list . Kv ... ) size += list . Size ( ) if size >= pageSize { st . kvChan <- outList outList = new ( pb . KVList ) size = 0 } } if len ( outList . Kv ) > 0 { st . kvChan <- outList } return nil } for { select { case kr , ok := <- st . rangeCh : if ! ok { // Done with the keys. return nil } if err := iterate ( kr ) ; err != nil { return err } case <- ctx . Done ( ) : return ctx . Err ( ) } } }
// produceKVs picks up ranges from rangeCh, generates KV lists and sends them to kvChan.
produceKVs picks up ranges from rangeCh generates KV lists and sends them to kvChan .