-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Showing
5 changed files
with
358 additions
and
126 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/ipfs/boxo/ipns" | ||
| "github.com/ipfs/boxo/routing/http/server" | ||
| "github.com/ipfs/boxo/routing/http/types" | ||
| "github.com/ipfs/boxo/routing/http/types/iter" | ||
| "github.com/ipfs/go-cid" | ||
| "github.com/libp2p/go-libp2p/core/peer" | ||
| "github.com/libp2p/go-libp2p/core/routing" | ||
| ) | ||
|
|
||
| var _ server.ContentRouter = composableRouter{} | ||
|
|
||
| type composableRouter struct { | ||
| providers server.ContentRouter | ||
| peers server.ContentRouter | ||
| ipns server.ContentRouter | ||
| } | ||
|
|
||
| func (r composableRouter) FindProviders(ctx context.Context, key cid.Cid, limit int) (iter.ResultIter[types.Record], error) { | ||
| return r.providers.FindProviders(ctx, key, limit) | ||
| } | ||
|
|
||
| //lint:ignore SA1019 // ignore staticcheck | ||
| func (r composableRouter) ProvideBitswap(ctx context.Context, req *server.BitswapWriteProvideRequest) (time.Duration, error) { | ||
| return 0, routing.ErrNotSupported | ||
| } | ||
|
|
||
| func (r composableRouter) FindPeers(ctx context.Context, pid peer.ID, limit int) (iter.ResultIter[*types.PeerRecord], error) { | ||
| return r.providers.FindPeers(ctx, pid, limit) | ||
| } | ||
|
|
||
| func (r composableRouter) GetIPNS(ctx context.Context, name ipns.Name) (*ipns.Record, error) { | ||
| return r.ipns.GetIPNS(ctx, name) | ||
| } | ||
|
|
||
| func (r composableRouter) PutIPNS(ctx context.Context, name ipns.Name, record *ipns.Record) error { | ||
| return r.ipns.PutIPNS(ctx, name, record) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/ipfs/boxo/ipns" | ||
| "github.com/ipfs/go-cid" | ||
| dht "github.com/libp2p/go-libp2p-kad-dht" | ||
| "github.com/libp2p/go-libp2p-kad-dht/fullrt" | ||
| record "github.com/libp2p/go-libp2p-record" | ||
| "github.com/libp2p/go-libp2p/core/host" | ||
| "github.com/libp2p/go-libp2p/core/peer" | ||
| "github.com/libp2p/go-libp2p/core/routing" | ||
| ) | ||
|
|
||
| type bundledDHT struct { | ||
| standard *dht.IpfsDHT | ||
| fullRT *fullrt.FullRT | ||
| } | ||
|
|
||
| func newBundledDHT(ctx context.Context, h host.Host) (routing.Routing, error) { | ||
| standardDHT, err := dht.New(ctx, h, dht.Mode(dht.ModeClient), dht.BootstrapPeers(dht.GetDefaultBootstrapPeerAddrInfos()...)) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| fullRT, err := fullrt.NewFullRT(h, "/ipfs", | ||
| fullrt.DHTOption( | ||
| dht.BucketSize(20), | ||
| dht.Validator(record.NamespacedValidator{ | ||
| "pk": record.PublicKeyValidator{}, | ||
| "ipns": ipns.Validator{}, | ||
| }), | ||
| dht.BootstrapPeers(dht.GetDefaultBootstrapPeerAddrInfos()...), | ||
| dht.Mode(dht.ModeClient), | ||
| )) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &bundledDHT{ | ||
| standard: standardDHT, | ||
| fullRT: fullRT, | ||
| }, nil | ||
| } | ||
|
|
||
| func (b *bundledDHT) getDHT() routing.Routing { | ||
| if b.fullRT.Ready() { | ||
| return b.fullRT | ||
| } | ||
| return b.standard | ||
| } | ||
|
|
||
| func (b *bundledDHT) Provide(ctx context.Context, c cid.Cid, brdcst bool) error { | ||
| return b.getDHT().Provide(ctx, c, brdcst) | ||
| } | ||
|
|
||
| func (b *bundledDHT) FindProvidersAsync(ctx context.Context, c cid.Cid, i int) <-chan peer.AddrInfo { | ||
| return b.getDHT().FindProvidersAsync(ctx, c, i) | ||
| } | ||
|
|
||
| func (b *bundledDHT) FindPeer(ctx context.Context, id peer.ID) (peer.AddrInfo, error) { | ||
| return b.getDHT().FindPeer(ctx, id) | ||
| } | ||
|
|
||
| func (b *bundledDHT) PutValue(ctx context.Context, k string, v []byte, option ...routing.Option) error { | ||
| return b.getDHT().PutValue(ctx, k, v, option...) | ||
| } | ||
|
|
||
| func (b *bundledDHT) GetValue(ctx context.Context, s string, option ...routing.Option) ([]byte, error) { | ||
| return b.getDHT().GetValue(ctx, s, option...) | ||
| } | ||
|
|
||
| func (b *bundledDHT) SearchValue(ctx context.Context, s string, option ...routing.Option) (<-chan []byte, error) { | ||
| return b.getDHT().SearchValue(ctx, s, option...) | ||
| } | ||
|
|
||
| func (b *bundledDHT) Bootstrap(ctx context.Context) error { | ||
| return b.standard.Bootstrap(ctx) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,197 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/ipfs/boxo/ipns" | ||
| "github.com/ipfs/boxo/routing/http/server" | ||
| "github.com/ipfs/boxo/routing/http/types" | ||
| "github.com/ipfs/boxo/routing/http/types/iter" | ||
| "github.com/ipfs/go-cid" | ||
| "github.com/libp2p/go-libp2p/core/peer" | ||
| "github.com/libp2p/go-libp2p/core/routing" | ||
| ) | ||
|
|
||
| var _ server.ContentRouter = parallelRouter{} | ||
|
|
||
| type parallelRouter struct { | ||
| routers []server.ContentRouter | ||
| } | ||
|
|
||
| func (r parallelRouter) FindProviders(ctx context.Context, key cid.Cid, limit int) (iter.ResultIter[types.Record], error) { | ||
| switch len(r.routers) { | ||
| case 0: | ||
| return iter.ToResultIter(iter.FromSlice([]types.Record{})), nil | ||
| case 1: | ||
| return r.routers[0].FindProviders(ctx, key, limit) | ||
| } | ||
|
|
||
| its := make([]iter.ResultIter[types.Record], len(r.routers)) | ||
| for i, ri := range r.routers { | ||
| var err error | ||
| its[i], err = ri.FindProviders(ctx, key, limit) | ||
| if err != nil { | ||
| for _, it := range its { | ||
| if it != nil { | ||
| _ = it.Close() | ||
| } | ||
| } | ||
| return nil, err | ||
| } | ||
| } | ||
| return &manyIter[types.Record]{it: its}, nil | ||
| } | ||
|
|
||
| type manyIter[T any] struct { | ||
| it []iter.ResultIter[T] | ||
| next int | ||
| } | ||
|
|
||
| func (mi *manyIter[T]) Next() bool { | ||
| for i, it := range mi.it { | ||
| if it.Next() { | ||
| mi.next = i | ||
| return true | ||
| } | ||
| } | ||
|
|
||
| mi.next = -1 | ||
| return false | ||
| } | ||
|
|
||
| func (mi *manyIter[T]) Val() iter.Result[T] { | ||
| if mi.next == -1 { | ||
| return iter.Result[T]{Err: errors.New("no next value")} | ||
| } | ||
| return mi.it[mi.next].Val() | ||
| } | ||
|
|
||
| func (mi *manyIter[T]) Close() error { | ||
| var err error | ||
| for _, it := range mi.it { | ||
| err = errors.Join(err, it.Close()) | ||
| } | ||
| return err | ||
| } | ||
|
|
||
| //lint:ignore SA1019 // ignore staticcheck | ||
| func (r parallelRouter) ProvideBitswap(ctx context.Context, req *server.BitswapWriteProvideRequest) (time.Duration, error) { | ||
| return 0, routing.ErrNotSupported | ||
| } | ||
|
|
||
| func (r parallelRouter) FindPeers(ctx context.Context, pid peer.ID, limit int) (iter.ResultIter[*types.PeerRecord], error) { | ||
| switch len(r.routers) { | ||
| case 0: | ||
| return iter.ToResultIter(iter.FromSlice([]*types.PeerRecord{})), nil | ||
| case 1: | ||
| return r.routers[0].FindPeers(ctx, pid, limit) | ||
| } | ||
|
|
||
| its := make([]iter.ResultIter[*types.PeerRecord], len(r.routers)) | ||
| for i, ri := range r.routers { | ||
| var err error | ||
| its[i], err = ri.FindPeers(ctx, pid, limit) | ||
| if err != nil { | ||
| for _, it := range its { | ||
| if it != nil { | ||
| _ = it.Close() | ||
| } | ||
| } | ||
| return nil, err | ||
| } | ||
| } | ||
| return &manyIter[*types.PeerRecord]{it: its}, nil | ||
| } | ||
|
|
||
| func (r parallelRouter) GetIPNS(ctx context.Context, name ipns.Name) (*ipns.Record, error) { | ||
| switch len(r.routers) { | ||
| case 0: | ||
| return nil, routing.ErrNotFound | ||
| case 1: | ||
| return r.routers[0].GetIPNS(ctx, name) | ||
| } | ||
|
|
||
| ctx, cancel := context.WithCancel(ctx) | ||
| defer cancel() | ||
|
|
||
| results := make(chan struct { | ||
| val *ipns.Record | ||
| err error | ||
| }) | ||
| for _, ri := range r.routers { | ||
| go func(ri server.ContentRouter) { | ||
| value, err := ri.GetIPNS(ctx, name) | ||
| select { | ||
| case results <- struct { | ||
| val *ipns.Record | ||
| err error | ||
| }{ | ||
| val: value, | ||
| err: err, | ||
| }: | ||
| case <-ctx.Done(): | ||
| } | ||
| }(ri) | ||
| } | ||
|
|
||
| var errs error | ||
|
|
||
| for range r.routers { | ||
| select { | ||
| case res := <-results: | ||
| switch res.err { | ||
| case nil: | ||
| return res.val, nil | ||
| case routing.ErrNotFound, routing.ErrNotSupported: | ||
| continue | ||
| } | ||
| // If the context has expired, just return that error | ||
| // and ignore the other errors. | ||
| if ctx.Err() != nil { | ||
| return nil, ctx.Err() | ||
| } | ||
|
|
||
| errs = errors.Join(errs, res.err) | ||
| case <-ctx.Done(): | ||
| return nil, ctx.Err() | ||
| } | ||
| } | ||
|
|
||
| if errs == nil { | ||
| return nil, routing.ErrNotFound | ||
| } | ||
|
|
||
| return nil, errs | ||
| } | ||
|
|
||
| func (r parallelRouter) PutIPNS(ctx context.Context, name ipns.Name, record *ipns.Record) error { | ||
| switch len(r.routers) { | ||
| case 0: | ||
| return nil | ||
| case 1: | ||
| return r.routers[0].PutIPNS(ctx, name, record) | ||
| } | ||
|
|
||
| ctx, cancel := context.WithCancel(ctx) | ||
| defer cancel() | ||
|
|
||
| var wg sync.WaitGroup | ||
| results := make([]error, len(r.routers)) | ||
| wg.Add(len(r.routers)) | ||
| for i, ri := range r.routers { | ||
| go func(ri server.ContentRouter, i int) { | ||
| results[i] = ri.PutIPNS(ctx, name, record) | ||
| wg.Done() | ||
| }(ri, i) | ||
| } | ||
| wg.Wait() | ||
|
|
||
| var errs error | ||
| for _, err := range results { | ||
| errs = errors.Join(errs, err) | ||
| } | ||
| return errs | ||
| } |