/
contentrouter.go
151 lines (130 loc) · 3.86 KB
/
contentrouter.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
package contentrouter
import (
"context"
"reflect"
"time"
"github.com/sevenrats/boxo/routing/http/internal"
"github.com/sevenrats/boxo/routing/http/types"
"github.com/sevenrats/boxo/routing/http/types/iter"
"github.com/ipfs/go-cid"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-libp2p/core/routing"
"github.com/multiformats/go-multiaddr"
"github.com/multiformats/go-multihash"
)
var logger = logging.Logger("service/contentrouting")
const ttl = 24 * time.Hour
type Client interface {
ProvideBitswap(ctx context.Context, keys []cid.Cid, ttl time.Duration) (time.Duration, error)
FindProviders(ctx context.Context, key cid.Cid) (iter.ResultIter[types.ProviderResponse], error)
}
type contentRouter struct {
client Client
maxProvideConcurrency int
maxProvideBatchSize int
}
var _ routing.ContentRouting = (*contentRouter)(nil)
type option func(c *contentRouter)
func WithMaxProvideConcurrency(max int) option {
return func(c *contentRouter) {
c.maxProvideConcurrency = max
}
}
func WithMaxProvideBatchSize(max int) option {
return func(c *contentRouter) {
c.maxProvideBatchSize = max
}
}
func NewContentRoutingClient(c Client, opts ...option) *contentRouter {
cr := &contentRouter{
client: c,
maxProvideConcurrency: 5,
maxProvideBatchSize: 100,
}
for _, opt := range opts {
opt(cr)
}
return cr
}
func (c *contentRouter) Provide(ctx context.Context, key cid.Cid, announce bool) error {
// If 'true' is
// passed, it also announces it, otherwise it is just kept in the local
// accounting of which objects are being provided.
if !announce {
return nil
}
_, err := c.client.ProvideBitswap(ctx, []cid.Cid{key}, ttl)
return err
}
// ProvideMany provides a set of keys to the remote delegate.
// Large sets of keys are chunked into multiple requests and sent concurrently, according to the concurrency configuration.
// TODO: implement retries through transient errors
func (c *contentRouter) ProvideMany(ctx context.Context, mhKeys []multihash.Multihash) error {
keys := make([]cid.Cid, 0, len(mhKeys))
for _, m := range mhKeys {
keys = append(keys, cid.NewCidV1(cid.Raw, m))
}
if len(keys) <= c.maxProvideBatchSize {
_, err := c.client.ProvideBitswap(ctx, keys, ttl)
return err
}
return internal.DoBatch(
ctx,
c.maxProvideBatchSize,
c.maxProvideConcurrency,
keys,
func(ctx context.Context, batch []cid.Cid) error {
_, err := c.client.ProvideBitswap(ctx, batch, ttl)
return err
},
)
}
// Ready is part of the existing `ProvideMany` interface.
func (c *contentRouter) Ready() bool {
return true
}
// readProviderResponses reads bitswap records from the iterator into the given channel, dropping non-bitswap records.
func readProviderResponses(iter iter.ResultIter[types.ProviderResponse], ch chan<- peer.AddrInfo) {
defer close(ch)
defer iter.Close()
for iter.Next() {
res := iter.Val()
if res.Err != nil {
logger.Warnw("error iterating provider responses: %s", res.Err)
continue
}
v := res.Val
if v.GetSchema() == types.SchemaBitswap {
result, ok := v.(*types.ReadBitswapProviderRecord)
if !ok {
logger.Errorw(
"problem casting find providers result",
"Schema", v.GetSchema(),
"Type", reflect.TypeOf(v).String(),
)
continue
}
var addrs []multiaddr.Multiaddr
for _, a := range result.Addrs {
addrs = append(addrs, a.Multiaddr)
}
ch <- peer.AddrInfo{
ID: *result.ID,
Addrs: addrs,
}
}
}
}
func (c *contentRouter) FindProvidersAsync(ctx context.Context, key cid.Cid, numResults int) <-chan peer.AddrInfo {
resultsIter, err := c.client.FindProviders(ctx, key)
if err != nil {
logger.Warnw("error finding providers", "CID", key, "Error", err)
ch := make(chan peer.AddrInfo)
close(ch)
return ch
}
ch := make(chan peer.AddrInfo)
go readProviderResponses(resultsIter, ch)
return ch
}