mirror of
https://github.com/ipfs/kubo.git
synced 2026-02-21 10:27:46 +08:00
133 lines
3.3 KiB
Go
133 lines
3.3 KiB
Go
package coreapi
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
|
|
caopts "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
|
|
|
|
cid "gx/ipfs/QmPSQnBKM9g7BaUcZCvswUJVscQ1ipjmwxN5PXCjkp9EQ7/go-cid"
|
|
cidutil "gx/ipfs/QmQJSeE3CX4zos9qeaG8EhecEK9zvrTEfTG84J8C5NVRwt/go-cidutil"
|
|
peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
|
|
dag "gx/ipfs/QmVB15p7qNVQQGC5RQcAQAovDFaQpRNqorbRwpvdNjHmVC/go-merkledag"
|
|
blockservice "gx/ipfs/QmWAU1Etv448cx9GLohtr27vnVC87amqE7fN4Hf4hcLDQY/go-blockservice"
|
|
routing "gx/ipfs/QmaM261ArNTmKMybV4LKy68JTZrf5CkCbRGDxsZwYHgqDS/go-libp2p-routing"
|
|
offline "gx/ipfs/QmcRC35JF2pJQneAxa5LdQBQRumWggccWErogSrCkS1h8T/go-ipfs-exchange-offline"
|
|
pstore "gx/ipfs/Qmda4cPRvSRyox3SqgJN6DfSZGU5TtHufPTp9uXjFj71X6/go-libp2p-peerstore"
|
|
blockstore "gx/ipfs/QmegPGspn3RpTMQ23Fd3GVVMopo1zsEMurudbFMZ5UXBLH/go-ipfs-blockstore"
|
|
)
|
|
|
|
type DhtAPI CoreAPI
|
|
|
|
func (api *DhtAPI) FindPeer(ctx context.Context, p peer.ID) (pstore.PeerInfo, error) {
|
|
pi, err := api.node.Routing.FindPeer(ctx, peer.ID(p))
|
|
if err != nil {
|
|
return pstore.PeerInfo{}, err
|
|
}
|
|
|
|
return pi, nil
|
|
}
|
|
|
|
func (api *DhtAPI) FindProviders(ctx context.Context, p coreiface.Path, opts ...caopts.DhtFindProvidersOption) (<-chan pstore.PeerInfo, error) {
|
|
settings, err := caopts.DhtFindProvidersOptions(opts...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rp, err := api.core().ResolvePath(ctx, p)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
numProviders := settings.NumProviders
|
|
if numProviders < 1 {
|
|
return nil, fmt.Errorf("number of providers must be greater than 0")
|
|
}
|
|
|
|
pchan := api.node.Routing.FindProvidersAsync(ctx, rp.Cid(), numProviders)
|
|
return pchan, nil
|
|
}
|
|
|
|
func (api *DhtAPI) Provide(ctx context.Context, path coreiface.Path, opts ...caopts.DhtProvideOption) error {
|
|
settings, err := caopts.DhtProvideOptions(opts...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if api.node.Routing == nil {
|
|
return errors.New("cannot provide in offline mode")
|
|
}
|
|
|
|
rp, err := api.core().ResolvePath(ctx, path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
c := rp.Cid()
|
|
|
|
has, err := api.node.Blockstore.Has(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if !has {
|
|
return fmt.Errorf("block %s not found locally, cannot provide", c)
|
|
}
|
|
|
|
if settings.Recursive {
|
|
err = provideKeysRec(ctx, api.node.Routing, api.node.Blockstore, []cid.Cid{c})
|
|
} else {
|
|
err = provideKeys(ctx, api.node.Routing, []cid.Cid{c})
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func provideKeys(ctx context.Context, r routing.IpfsRouting, cids []cid.Cid) error {
|
|
for _, c := range cids {
|
|
err := r.Provide(ctx, c, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func provideKeysRec(ctx context.Context, r routing.IpfsRouting, bs blockstore.Blockstore, cids []cid.Cid) error {
|
|
provided := cidutil.NewStreamingSet()
|
|
|
|
errCh := make(chan error)
|
|
go func() {
|
|
dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
|
|
for _, c := range cids {
|
|
err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, provided.Visitor(ctx))
|
|
if err != nil {
|
|
errCh <- err
|
|
}
|
|
}
|
|
}()
|
|
|
|
for {
|
|
select {
|
|
case k := <-provided.New:
|
|
err := r.Provide(ctx, k, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
case err := <-errCh:
|
|
return err
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (api *DhtAPI) core() coreiface.CoreAPI {
|
|
return (*CoreAPI)(api)
|
|
}
|