func (ab *dsAddrBook) ConsumePeerRecord(recordEnvelope *record.Envelope, ttl time.Duration) (bool, error) {
	r, err := recordEnvelope.Record()
	if err != nil {
		return false, err
	}
	rec, ok := r.(*peer.PeerRecord)
	if !ok {
		return false, fmt.Errorf("envelope did not contain PeerRecord")
	}
	if !rec.PeerID.MatchesPublicKey(recordEnvelope.PublicKey) {
		return false, fmt.Errorf("signing key does not match PeerID in PeerRecord")
	}
	if ab.latestPeerRecordSeq(rec.PeerID) > rec.Seq {
		return false, nil
	}
	addrs := cleanAddrs(rec.Addrs)
	err = ab.setAddrs(rec.PeerID, addrs, ttl, ttlExtend, true)
	if err != nil {
		return false, err
	}
	err = ab.storeSignedPeerRecord(rec.PeerID, recordEnvelope, rec)
	if err != nil {
		return false, err
	}
	return true, nil
}
ConsumePeerRecord adds addresses from a signed peer.PeerRecord (contained in
a record.Envelope), which will expire after the given TTL.
See https://godoc.org/github.com/libp2p/go-libp2p-core/peerstore#CertifiedAddrBook for more details.
package pstoreds
import (
	"context"
	"fmt"
	"sort"
	"sync"
	"time"
	ds "github.com/ipfs/go-datastore"
	"github.com/ipfs/go-datastore/query"
	logging "github.com/ipfs/go-log"
	"github.com/libp2p/go-libp2p-core/peer"
	pstore "github.com/libp2p/go-libp2p-core/peerstore"
	"github.com/libp2p/go-libp2p-core/record"
	pb "github.com/libp2p/go-libp2p-peerstore/pb"
	"github.com/libp2p/go-libp2p-peerstore/pstoremem"
	lru "github.com/hashicorp/golang-lru"
	b32 "github.com/multiformats/go-base32"
	ma "github.com/multiformats/go-multiaddr"
)
type ttlWriteMode int
const (
	ttlOverride ttlWriteMode = iota
	ttlExtend
)
var (
	log = logging.Logger("peerstore/ds")
	addrBookBase = ds.NewKey("/peers/addrs")
)
type addrsRecord struct {
	sync.RWMutex
	*pb.AddrBookRecord
	dirty bool
}
func (r *addrsRecord) flush(write ds.Write) (err error) {
	key := addrBookBase.ChildString(b32.RawStdEncoding.EncodeToString([]byte(r.Id.ID)))
	if len(r.Addrs) == 0 {
		if err = write.Delete(context.TODO(), key); err == nil {
			r.dirty = false
		}
		return err
	}
	data, err := r.Marshal()
	if err != nil {
		return err
	}
	if err = write.Put(context.TODO(), key, data); err != nil {
		return err
	}
	r.dirty = false
	return nil
}
func (r *addrsRecord) clean() (chgd bool) {
	now := time.Now().Unix()
	addrsLen := len(r.Addrs)
	if !r.dirty && !r.hasExpiredAddrs(now) {
		return false
	}
	if addrsLen == 0 {
		return true
	}
	if r.dirty && addrsLen > 1 {
		sort.Slice(r.Addrs, func(i, j int) bool {
			return r.Addrs[i].Expiry < r.Addrs[j].Expiry
		})
	}
	r.Addrs = removeExpired(r.Addrs, now)
	return r.dirty || len(r.Addrs) != addrsLen
}
func (r *addrsRecord) hasExpiredAddrs(now int64) bool {
	if len(r.Addrs) > 0 && r.Addrs[0].Expiry <= now {
		return true
	}
	return false
}
func removeExpired(entries []*pb.AddrBookRecord_AddrEntry, now int64) []*pb.AddrBookRecord_AddrEntry {
	pivot := -1
	for i, addr := range entries {
		if addr.Expiry > now {
			break
		}
		pivot = i
	}
	return entries[pivot+1:]
}
type dsAddrBook struct {
	ctx  context.Context
	opts Options
	cache       cache
	ds          ds.Batching
	gc          *dsAddrBookGc
	subsManager *pstoremem.AddrSubManager
	childrenDone sync.WaitGroup
	cancelFn     func()
}
var _ pstore.AddrBook = (*dsAddrBook)(nil)
var _ pstore.CertifiedAddrBook = (*dsAddrBook)(nil)
func NewAddrBook(ctx context.Context, store ds.Batching, opts Options) (ab *dsAddrBook, err error) {
	ctx, cancelFn := context.WithCancel(ctx)
	ab = &dsAddrBook{
		ctx:         ctx,
		ds:          store,
		opts:        opts,
		cancelFn:    cancelFn,
		subsManager: pstoremem.NewAddrSubManager(),
	}
	if opts.CacheSize > 0 {
		if ab.cache, err = lru.NewARC(int(opts.CacheSize)); err != nil {
			return nil, err
		}
	} else {
		ab.cache = new(noopCache)
	}
	if ab.gc, err = newAddressBookGc(ctx, ab); err != nil {
		return nil, err
	}
	return ab, nil
}
func (ab *dsAddrBook) Close() error {
	ab.cancelFn()
	ab.childrenDone.Wait()
	return nil
}
func (ab *dsAddrBook) loadRecord(id peer.ID, cache bool, update bool) (pr *addrsRecord, err error) {
	if err := id.Validate(); err != nil {
		return nil, err
	}
	if e, ok := ab.cache.Get(id); ok {
		pr = e.(*addrsRecord)
		pr.Lock()
		defer pr.Unlock()
		if pr.clean() && update {
			err = pr.flush(ab.ds)
		}
		return pr, err
	}
	pr = &addrsRecord{AddrBookRecord: &pb.AddrBookRecord{}}
	key := addrBookBase.ChildString(b32.RawStdEncoding.EncodeToString([]byte(id)))
	data, err := ab.ds.Get(context.TODO(), key)
	switch err {
	case ds.ErrNotFound:
		err = nil
		pr.Id = &pb.ProtoPeerID{ID: id}
	case nil:
		if err = pr.Unmarshal(data); err != nil {
			return nil, err
		}
		if pr.clean() && update {
			err = pr.flush(ab.ds)
		}
	default:
		return nil, err
	}
	if cache {
		ab.cache.Add(id, pr)
	}
	return pr, err
}
func (ab *dsAddrBook) AddAddr(p peer.ID, addr ma.Multiaddr, ttl time.Duration) {
	ab.AddAddrs(p, []ma.Multiaddr{addr}, ttl)
}
func (ab *dsAddrBook) AddAddrs(p peer.ID, addrs []ma.Multiaddr, ttl time.Duration) {
	if ttl <= 0 {
		return
	}
	addrs = cleanAddrs(addrs)
	ab.setAddrs(p, addrs, ttl, ttlExtend, false)
}