@cryptotaxi247 / kubo / commits / 94000e649

replace nodebuilder with a nicer interface

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> use NewNode instead of NewIPFSNode in most of the codebase License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> make mocknet work with node constructor better License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> finish cleanup of old construction method License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> blockservice.New doesnt return an error anymore License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> break up node construction into separate function License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com> add error case to default filling on node constructor License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Aug 13, 2015 at 11:44 UTC 94000e64907f78406f9ed80e091b728bc608e284
21 files changed +253 -401
cmd/ipfs/daemon.go
+8 -5
@@ -192,9 +192,11 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
192 return
193 }
194
195 - // Start assembling corebuilder
196 - nb := core.NewNodeBuilder().Online()
197 - nb.SetRepo(repo)
195 + // Start assembling node config
196 + ncfg := &core.BuildCfg{
197 + Online: true,
198 + Repo: repo,
199 + }
200
201 routingOption, _, err := req.Option(routingOptionKwd).String()
202 if err != nil {
@@ -215,10 +217,11 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
217 Addrs: []ma.Multiaddr{addr.Transport()},
218 })
219 }
218 - nb.SetRouting(corerouting.SupernodeClient(infos...))
220 +
221 + ncfg.Routing = corerouting.SupernodeClient(infos...)
222 }
223
221 - node, err := nb.Build(req.Context())
224 + node, err := core.NewNode(req.Context(), ncfg)
225 if err != nil {
226 log.Error("error from node construction: ", err)
227 res.SetError(err, cmds.ErrNormal)
cmd/ipfs/init.go
+2 -2
@@ -157,7 +157,7 @@ func addDefaultAssets(out io.Writer, repoRoot string) error {
157 return err
158 }
159
160 - nd, err := core.NewIPFSNode(ctx, core.Offline(r))
160 + nd, err := core.NewNode(ctx, &core.BuildCfg{Repo: r})
161 if err != nil {
162 return err
163 }
@@ -191,7 +191,7 @@ func initializeIpnsKeyspace(repoRoot string) error {
191 return err
192 }
193
194 - nd, err := core.NewIPFSNode(ctx, core.Offline(r))
194 + nd, err := core.NewNode(ctx, &core.BuildCfg{Repo: r})
195 if err != nil {
196 return err
197 }
cmd/ipfs/main.go
+4 -1
@@ -204,7 +204,10 @@ func (i *cmdInvocation) constructNodeFunc(ctx context.Context) func() (*core.Ipf
204
205 // ok everything is good. set it on the invocation (for ownership)
206 // and return it.
207 - n, err := core.NewIPFSNode(ctx, core.Standard(r, cmdctx.Online))
207 + n, err := core.NewNode(ctx, &core.BuildCfg{
208 + Online: cmdctx.Online,
209 + Repo: r,
210 + })
211 if err != nil {
212 return nil, err
213 }
cmd/ipfswatch/main.go
+5 -1
@@ -71,7 +71,11 @@ func run(ipfsPath, watchPath string) error {
71 // TODO handle case: repo doesn't exist or isn't initialized
72 return err
73 }
74 - node, err := core.NewIPFSNode(context.Background(), core.Online(r))
74 +
75 + node, err := core.NewNode(context.Background(), &core.BuildCfg{
76 + Online: true,
77 + Repo: r,
78 + })
79 if err != nil {
80 return err
81 }
core/builder.go
+98 -53
@@ -7,31 +7,60 @@ import (
7
8 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
9 dsync "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
10 + goprocessctx "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
11 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
12 + bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
13 key "github.com/ipfs/go-ipfs/blocks/key"
14 + bserv "github.com/ipfs/go-ipfs/blockservice"
15 + offline "github.com/ipfs/go-ipfs/exchange/offline"
16 + dag "github.com/ipfs/go-ipfs/merkledag"
17 ci "github.com/ipfs/go-ipfs/p2p/crypto"
18 + peer "github.com/ipfs/go-ipfs/p2p/peer"
19 + path "github.com/ipfs/go-ipfs/path"
20 + pin "github.com/ipfs/go-ipfs/pin"
21 repo "github.com/ipfs/go-ipfs/repo"
22 cfg "github.com/ipfs/go-ipfs/repo/config"
23 )
24
17 -var ErrAlreadyBuilt = errors.New("this builder has already been used")
25 +type BuildCfg struct {
26 + // If online is set, the node will have networking enabled
27 + Online bool
28
19 -// NodeBuilder is an object used to generate an IpfsNode
20 -type NodeBuilder struct {
21 - online bool
22 - routing RoutingOption
23 - peerhost HostOption
24 - repo repo.Repo
25 - built bool
26 - nilrepo bool
29 + // If NilRepo is set, a repo backed by a nil datastore will be constructed
30 + NilRepo bool
31 +
32 + Routing RoutingOption
33 + Host HostOption
34 + Repo repo.Repo
35 }
36
29 -func NewNodeBuilder() *NodeBuilder {
30 - return &NodeBuilder{
31 - online: false,
32 - routing: DHTOption,
33 - peerhost: DefaultHostOption,
37 +func (cfg *BuildCfg) fillDefaults() error {
38 + if cfg.Repo != nil && cfg.NilRepo {
39 + return errors.New("cannot set a repo and specify nilrepo at the same time")
40 }
41 +
42 + if cfg.Repo == nil {
43 + var d ds.Datastore
44 + d = ds.NewMapDatastore()
45 + if cfg.NilRepo {
46 + d = ds.NewNullDatastore()
47 + }
48 + r, err := defaultRepo(dsync.MutexWrap(d))
49 + if err != nil {
50 + return err
51 + }
52 + cfg.Repo = r
53 + }
54 +
55 + if cfg.Routing == nil {
56 + cfg.Routing = DHTOption
57 + }
58 +
59 + if cfg.Host == nil {
60 + cfg.Host = DefaultHostOption
61 + }
62 +
63 + return nil
64 }
65
66 func defaultRepo(dstore ds.ThreadSafeDatastore) (repo.Repo, error) {
@@ -62,53 +91,69 @@ func defaultRepo(dstore ds.ThreadSafeDatastore) (repo.Repo, error) {
91 }, nil
92 }
93
65 -func (nb *NodeBuilder) Online() *NodeBuilder {
66 - nb.online = true
67 - return nb
68 -}
94 +func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
95 + if cfg == nil {
96 + cfg = new(BuildCfg)
97 + }
98
70 -func (nb *NodeBuilder) Offline() *NodeBuilder {
71 - nb.online = false
72 - return nb
73 -}
99 + err := cfg.fillDefaults()
100 + if err != nil {
101 + return nil, err
102 + }
103
75 -func (nb *NodeBuilder) SetRouting(ro RoutingOption) *NodeBuilder {
76 - nb.routing = ro
77 - return nb
78 -}
104 + n := &IpfsNode{
105 + mode: offlineMode,
106 + Repo: cfg.Repo,
107 + ctx: ctx,
108 + Peerstore: peer.NewPeerstore(),
109 + }
110 + if cfg.Online {
111 + n.mode = onlineMode
112 + }
113
80 -func (nb *NodeBuilder) SetHost(ho HostOption) *NodeBuilder {
81 - nb.peerhost = ho
82 - return nb
83 -}
114 + // TODO: this is a weird circular-ish dependency, rework it
115 + n.proc = goprocessctx.WithContextAndTeardown(ctx, n.teardown)
116
85 -func (nb *NodeBuilder) SetRepo(r repo.Repo) *NodeBuilder {
86 - nb.repo = r
87 - return nb
88 -}
117 + if err := setupNode(ctx, n, cfg); err != nil {
118 + n.Close()
119 + return nil, err
120 + }
121
90 -func (nb *NodeBuilder) NilRepo() *NodeBuilder {
91 - nb.nilrepo = true
92 - return nb
122 + return n, nil
123 }
124
95 -func (nb *NodeBuilder) Build(ctx context.Context) (*IpfsNode, error) {
96 - if nb.built {
97 - return nil, ErrAlreadyBuilt
125 +func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
126 + // setup local peer ID (private key is loaded in online setup)
127 + if err := n.loadID(); err != nil {
128 + return err
129 }
99 - nb.built = true
100 - if nb.repo == nil {
101 - var d ds.Datastore
102 - d = ds.NewMapDatastore()
103 - if nb.nilrepo {
104 - d = ds.NewNullDatastore()
105 - }
106 - r, err := defaultRepo(dsync.MutexWrap(d))
107 - if err != nil {
108 - return nil, err
130 +
131 + var err error
132 + n.Blockstore, err = bstore.WriteCached(bstore.NewBlockstore(n.Repo.Datastore()), kSizeBlockstoreWriteCache)
133 + if err != nil {
134 + return err
135 + }
136 +
137 + if cfg.Online {
138 + do := setupDiscoveryOption(n.Repo.Config().Discovery)
139 + if err := n.startOnlineServices(ctx, cfg.Routing, cfg.Host, do); err != nil {
140 + return err
141 }
110 - nb.repo = r
142 + } else {
143 + n.Exchange = offline.Exchange(n.Blockstore)
144 }
112 - conf := standardWithRouting(nb.repo, nb.online, nb.routing, nb.peerhost)
113 - return NewIPFSNode(ctx, conf)
145 +
146 + n.Blocks = bserv.New(n.Blockstore, n.Exchange)
147 + n.DAG = dag.NewDAGService(n.Blocks)
148 + n.Pinning, err = pin.LoadPinner(n.Repo.Datastore(), n.DAG)
149 + if err != nil {
150 + // TODO: we should move towards only running 'NewPinner' explicity on
151 + // node init instead of implicitly here as a result of the pinner keys
152 + // not being found in the datastore.
153 + // this is kinda sketchy and could cause data loss
154 + n.Pinning = pin.NewPinner(n.Repo.Datastore(), n.DAG)
155 + }
156 + n.Resolver = &path.Resolver{DAG: n.DAG}
157 +
158 + return nil
159 }
core/commands/add.go
+5 -1
@@ -104,7 +104,11 @@ remains to be implemented.
104 chunker, _, _ := req.Option(chunkerOptionName).String()
105
106 if hash {
107 - nilnode, err := core.NewNodeBuilder().Build(n.Context())
107 + nilnode, err := core.NewNode(n.Context(), &core.BuildCfg{
108 + //TODO: need this to be true or all files
109 + // hashed will be stored in memory!
110 + NilRepo: false,
111 + })
112 if err != nil {
113 res.SetError(err, cmds.ErrNormal)
114 return
core/core.go
+4 -121
@@ -20,7 +20,6 @@ import (
20 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
21 ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
22 goprocess "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
23 - goprocessctx "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
23 mamask "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/whyrusleeping/multiaddr-filter"
24 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
25 diag "github.com/ipfs/go-ipfs/diagnostics"
@@ -46,7 +45,6 @@ import (
45 exchange "github.com/ipfs/go-ipfs/exchange"
46 bitswap "github.com/ipfs/go-ipfs/exchange/bitswap"
47 bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
49 - offline "github.com/ipfs/go-ipfs/exchange/offline"
48 rp "github.com/ipfs/go-ipfs/exchange/reprovide"
49
50 mount "github.com/ipfs/go-ipfs/fuse/mount"
@@ -123,124 +121,6 @@ type Mounts struct {
121 Ipns mount.Mount
122 }
123
126 -type ConfigOption func(ctx context.Context) (*IpfsNode, error)
127 -
128 -func NewIPFSNode(ctx context.Context, option ConfigOption) (*IpfsNode, error) {
129 - node, err := option(ctx)
130 - if err != nil {
131 - return nil, err
132 - }
133 -
134 - if node.ctx == nil {
135 - node.ctx = ctx
136 - }
137 - if node.proc == nil {
138 - node.proc = goprocessctx.WithContextAndTeardown(node.ctx, node.teardown)
139 - }
140 -
141 - success := false // flip to true after all sub-system inits succeed
142 - defer func() {
143 - if !success {
144 - node.proc.Close()
145 - }
146 - }()
147 -
148 - // Need to make sure it's perfectly clear 1) which variables are expected
149 - // to be initialized at this point, and 2) which variables will be
150 - // initialized after this point.
151 -
152 - node.Blocks = bserv.New(node.Blockstore, node.Exchange)
153 -
154 - if node.Peerstore == nil {
155 - node.Peerstore = peer.NewPeerstore()
156 - }
157 - node.DAG = merkledag.NewDAGService(node.Blocks)
158 - node.Pinning, err = pin.LoadPinner(node.Repo.Datastore(), node.DAG)
159 - if err != nil {
160 - node.Pinning = pin.NewPinner(node.Repo.Datastore(), node.DAG)
161 - }
162 - node.Resolver = &path.Resolver{DAG: node.DAG}
163 -
164 - success = true
165 - return node, nil
166 -}
167 -
168 -func Offline(r repo.Repo) ConfigOption {
169 - return Standard(r, false)
170 -}
171 -
172 -func OnlineWithOptions(r repo.Repo, router RoutingOption, ho HostOption) ConfigOption {
173 - return standardWithRouting(r, true, router, ho)
174 -}
175 -
176 -func Online(r repo.Repo) ConfigOption {
177 - return Standard(r, true)
178 -}
179 -
180 -// DEPRECATED: use Online, Offline functions
181 -func Standard(r repo.Repo, online bool) ConfigOption {
182 - return standardWithRouting(r, online, DHTOption, DefaultHostOption)
183 -}
184 -
185 -// TODO refactor so maybeRouter isn't special-cased in this way
186 -func standardWithRouting(r repo.Repo, online bool, routingOption RoutingOption, hostOption HostOption) ConfigOption {
187 - return func(ctx context.Context) (n *IpfsNode, err error) {
188 - // FIXME perform node construction in the main constructor so it isn't
189 - // necessary to perform this teardown in this scope.
190 - success := false
191 - defer func() {
192 - if !success && n != nil {
193 - n.teardown()
194 - }
195 - }()
196 -
197 - // TODO move as much of node initialization as possible into
198 - // NewIPFSNode. The larger these config options are, the harder it is
199 - // to test all node construction code paths.
200 -
201 - if r == nil {
202 - return nil, fmt.Errorf("repo required")
203 - }
204 - n = &IpfsNode{
205 - mode: func() mode {
206 - if online {
207 - return onlineMode
208 - }
209 - return offlineMode
210 - }(),
211 - Repo: r,
212 - }
213 -
214 - n.ctx = ctx
215 - n.proc = goprocessctx.WithContextAndTeardown(ctx, n.teardown)
216 -
217 - // setup Peerstore
218 - n.Peerstore = peer.NewPeerstore()
219 -
220 - // setup local peer ID (private key is loaded in online setup)
221 - if err := n.loadID(); err != nil {
222 - return nil, err
223 - }
224 -
225 - n.Blockstore, err = bstore.WriteCached(bstore.NewBlockstore(n.Repo.Datastore()), kSizeBlockstoreWriteCache)
226 - if err != nil {
227 - return nil, err
228 - }
229 -
230 - if online {
231 - do := setupDiscoveryOption(n.Repo.Config().Discovery)
232 - if err := n.startOnlineServices(ctx, routingOption, hostOption, do); err != nil {
233 - return nil, err
234 - }
235 - } else {
236 - n.Exchange = offline.Exchange(n.Blockstore)
237 - }
238 -
239 - success = true
240 - return n, nil
241 - }
242 -}
243 -
124 func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption RoutingOption, hostOption HostOption, do DiscoveryOption) error {
125
126 if n.PeerHost != nil { // already online.
@@ -371,10 +251,13 @@ func (n *IpfsNode) teardown() error {
251 // owned objects are closed in this teardown to ensure that they're closed
252 // regardless of which constructor was used to add them to the node.
253 closers := []io.Closer{
374 - n.Exchange,
254 n.Repo,
255 }
256
257 + if n.Exchange != nil {
258 + closers = append(closers, n.Exchange)
259 + }
260 +
261 if n.Mounts.Ipfs != nil {
262 closers = append(closers, mount.Closer(n.Mounts.Ipfs))
263 }
core/core_test.go
+2 -2
@@ -48,7 +48,7 @@ func TestInitialization(t *testing.T) {
48 C: *c,
49 D: testutil.ThreadSafeCloserMapDatastore(),
50 }
51 - n, err := NewIPFSNode(ctx, Standard(r, false))
51 + n, err := NewNode(ctx, &BuildCfg{Repo: r})
52 if n == nil || err != nil {
53 t.Error("Should have constructed.", i, err)
54 }
@@ -59,7 +59,7 @@ func TestInitialization(t *testing.T) {
59 C: *c,
60 D: testutil.ThreadSafeCloserMapDatastore(),
61 }
62 - n, err := NewIPFSNode(ctx, Standard(r, false))
62 + n, err := NewNode(ctx, &BuildCfg{Repo: r})
63 if n != nil || err == nil {
64 t.Error("Should have failed to construct.", i)
65 }
core/corehttp/gateway_test.go
+1 -1
@@ -47,7 +47,7 @@ func newNodeWithMockNamesys(ns mockNamesys) (*core.IpfsNode, error) {
47 C: c,
48 D: testutil.ThreadSafeCloserMapDatastore(),
49 }
50 - n, err := core.NewIPFSNode(context.Background(), core.Offline(r))
50 + n, err := core.NewNode(context.Background(), &core.BuildCfg{Repo: r})
51 if err != nil {
52 return nil, err
53 }
core/coreunix/add_test.go
+1 -1
@@ -25,7 +25,7 @@ func TestAddRecursive(t *testing.T) {
25 },
26 D: testutil.ThreadSafeCloserMapDatastore(),
27 }
28 - node, err := core.NewIPFSNode(context.Background(), core.Offline(r))
28 + node, err := core.NewNode(context.Background(), &core.BuildCfg{Repo: r})
29 if err != nil {
30 t.Fatal(err)
31 }
core/mock/mock.go
+19 -62
@@ -1,86 +1,39 @@
1 package coremock
2
3 import (
4 + "net"
5 +
6 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
7 syncds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
8 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
9
8 - "github.com/ipfs/go-ipfs/blocks/blockstore"
9 - blockservice "github.com/ipfs/go-ipfs/blockservice"
10 commands "github.com/ipfs/go-ipfs/commands"
11 core "github.com/ipfs/go-ipfs/core"
12 - "github.com/ipfs/go-ipfs/exchange/offline"
13 - mdag "github.com/ipfs/go-ipfs/merkledag"
14 - nsys "github.com/ipfs/go-ipfs/namesys"
12 + metrics "github.com/ipfs/go-ipfs/metrics"
13 + host "github.com/ipfs/go-ipfs/p2p/host"
14 mocknet "github.com/ipfs/go-ipfs/p2p/net/mock"
15 peer "github.com/ipfs/go-ipfs/p2p/peer"
17 - path "github.com/ipfs/go-ipfs/path"
18 - pin "github.com/ipfs/go-ipfs/pin"
16 "github.com/ipfs/go-ipfs/repo"
17 config "github.com/ipfs/go-ipfs/repo/config"
21 - offrt "github.com/ipfs/go-ipfs/routing/offline"
18 ds2 "github.com/ipfs/go-ipfs/util/datastore2"
19 testutil "github.com/ipfs/go-ipfs/util/testutil"
20 )
21
26 -// TODO this is super sketch. Deprecate and initialize one that shares code
27 -// with the actual core constructor. Lots of fields aren't initialized.
28 -// "This is as good as broken." --- is it?
29 -
22 // NewMockNode constructs an IpfsNode for use in tests.
23 func NewMockNode() (*core.IpfsNode, error) {
24 ctx := context.Background()
25
34 - // Generate Identity
35 - ident, err := testutil.RandIdentity()
36 - if err != nil {
37 - return nil, err
38 - }
39 - p := ident.ID()
40 -
41 - c := config.Config{
42 - Identity: config.Identity{
43 - PeerID: p.String(),
44 - },
45 - }
46 -
47 - nd, err := core.Offline(&repo.Mock{
48 - C: c,
49 - D: ds2.CloserWrap(syncds.MutexWrap(datastore.NewMapDatastore())),
50 - })(ctx)
51 - if err != nil {
52 - return nil, err
53 - }
54 -
55 - nd.PrivateKey = ident.PrivateKey()
56 - nd.Peerstore = peer.NewPeerstore()
57 - nd.Peerstore.AddPrivKey(p, ident.PrivateKey())
58 - nd.Peerstore.AddPubKey(p, ident.PublicKey())
59 - nd.Identity = p
26 + // effectively offline, only peer in its network
27 + return core.NewNode(ctx, &core.BuildCfg{
28 + Online: true,
29 + Host: MockHostOption(mocknet.New(ctx)),
30 + })
31 +}
32
61 - nd.PeerHost, err = mocknet.New(nd.Context()).AddPeer(ident.PrivateKey(), ident.Address()) // effectively offline
62 - if err != nil {
63 - return nil, err
33 +func MockHostOption(mn mocknet.Mocknet) core.HostOption {
34 + return func(ctx context.Context, id peer.ID, ps peer.Peerstore, bwr metrics.Reporter, fs []*net.IPNet) (host.Host, error) {
35 + return mn.AddPeerWithPeerstore(id, ps)
36 }
65 -
66 - // Routing
67 - nd.Routing = offrt.NewOfflineRouter(nd.Repo.Datastore(), nd.PrivateKey)
68 -
69 - // Bitswap
70 - bstore := blockstore.NewBlockstore(nd.Repo.Datastore())
71 - bserv := blockservice.New(bstore, offline.Exchange(bstore))
72 -
73 - nd.DAG = mdag.NewDAGService(bserv)
74 -
75 - nd.Pinning = pin.NewPinner(nd.Repo.Datastore(), nd.DAG)
76 -
77 - // Namespace resolver
78 - nd.Namesys = nsys.NewNameSystem(nd.Routing)
79 -
80 - // Path resolver
81 - nd.Resolver = &path.Resolver{DAG: nd.DAG}
82 -
83 - return nd, nil
37 }
38
39 func MockCmdsCtx() (commands.Context, error) {
@@ -97,10 +50,14 @@ func MockCmdsCtx() (commands.Context, error) {
50 },
51 }
52
100 - node, err := core.NewIPFSNode(context.Background(), core.Offline(&repo.Mock{
53 + r := &repo.Mock{
54 D: ds2.CloserWrap(syncds.MutexWrap(datastore.NewMapDatastore())),
55 C: conf,
103 - }))
56 + }
57 +
58 + node, err := core.NewNode(context.Background(), &core.BuildCfg{
59 + Repo: r,
60 + })
61
62 return commands.Context{
63 Online: true,
p2p/net/mock/interface.go
+1
@@ -25,6 +25,7 @@ type Mocknet interface {
25 // AddPeer adds an existing peer. we need both a privkey and addr.
26 // ID is derived from PrivKey
27 AddPeer(ic.PrivKey, ma.Multiaddr) (host.Host, error)
28 + AddPeerWithPeerstore(peer.ID, peer.Peerstore) (host.Host, error)
29
30 // retrieve things (with randomized iteration order)
31 Peers() []peer.ID
p2p/net/mock/mock_net.go
+15 -2
@@ -64,13 +64,26 @@ func (mn *mocknet) GenPeer() (host.Host, error) {
64 }
65
66 func (mn *mocknet) AddPeer(k ic.PrivKey, a ma.Multiaddr) (host.Host, error) {
67 - n, err := newPeernet(mn.ctx, mn, k, a)
67 + p, err := peer.IDFromPublicKey(k.GetPublic())
68 + if err != nil {
69 + return nil, err
70 + }
71 +
72 + ps := peer.NewPeerstore()
73 + ps.AddAddr(p, a, peer.PermanentAddrTTL)
74 + ps.AddPrivKey(p, k)
75 + ps.AddPubKey(p, k.GetPublic())
76 +
77 + return mn.AddPeerWithPeerstore(p, ps)
78 +}
79 +
80 +func (mn *mocknet) AddPeerWithPeerstore(p peer.ID, ps peer.Peerstore) (host.Host, error) {
81 + n, err := newPeernet(mn.ctx, mn, p, ps)
82 if err != nil {
83 return nil, err
84 }
85
86 h := bhost.New(n)
73 - log.Debugf("mocknet added listen addr for peer: %s -- %s", n.LocalPeer(), a)
87
88 mn.proc.AddChild(n.proc)
89
p2p/net/mock/mock_peernet.go
+1 -14
@@ -5,7 +5,6 @@ import (
5 "math/rand"
6 "sync"
7
8 - ic "github.com/ipfs/go-ipfs/p2p/crypto"
8 inet "github.com/ipfs/go-ipfs/p2p/net"
9 peer "github.com/ipfs/go-ipfs/p2p/peer"
10
@@ -40,19 +39,7 @@ type peernet struct {
39 }
40
41 // newPeernet constructs a new peernet
43 -func newPeernet(ctx context.Context, m *mocknet, k ic.PrivKey,
44 - a ma.Multiaddr) (*peernet, error) {
45 -
46 - p, err := peer.IDFromPublicKey(k.GetPublic())
47 - if err != nil {
48 - return nil, err
49 - }
50 -
51 - // create our own entirely, so that peers knowledge doesn't get shared
52 - ps := peer.NewPeerstore()
53 - ps.AddAddr(p, a, peer.PermanentAddrTTL)
54 - ps.AddPrivKey(p, k)
55 - ps.AddPubKey(p, k.GetPublic())
42 +func newPeernet(ctx context.Context, m *mocknet, p peer.ID, ps peer.Peerstore) (*peernet, error) {
43
44 n := &peernet{
45 mocknet: m,
test/integration/addcat_test.go
+19 -12
@@ -15,12 +15,16 @@ import (
15
16 "github.com/ipfs/go-ipfs/core"
17 coreunix "github.com/ipfs/go-ipfs/core/coreunix"
18 + mock "github.com/ipfs/go-ipfs/core/mock"
19 mocknet "github.com/ipfs/go-ipfs/p2p/net/mock"
20 "github.com/ipfs/go-ipfs/p2p/peer"
21 + eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
22 "github.com/ipfs/go-ipfs/thirdparty/unit"
23 testutil "github.com/ipfs/go-ipfs/util/testutil"
24 )
25
26 +var log = eventlog.Logger("epictest")
27 +
28 const kSeed = 1
29
30 func Test1KBInstantaneous(t *testing.T) {
@@ -87,35 +91,38 @@ func RandomBytes(n int64) []byte {
91 func DirectAddCat(data []byte, conf testutil.LatencyConfig) error {
92 ctx, cancel := context.WithCancel(context.Background())
93 defer cancel()
90 - const numPeers = 2
94
95 // create network
93 - mn, err := mocknet.FullMeshLinked(ctx, numPeers)
94 - if err != nil {
95 - return err
96 - }
96 + mn := mocknet.New(ctx)
97 mn.SetLinkDefaults(mocknet.LinkOptions{
98 Latency: conf.NetworkLatency,
99 // TODO add to conf. This is tricky because we want 0 values to be functional.
100 Bandwidth: math.MaxInt32,
101 })
102
103 - peers := mn.Peers()
104 - if len(peers) < numPeers {
105 - return errors.New("test initialization error")
106 - }
107 -
108 - adder, err := core.NewIPFSNode(ctx, core.ConfigOption(MocknetTestRepo(peers[0], mn.Host(peers[0]), conf, core.DHTOption)))
103 + adder, err := core.NewNode(ctx, &core.BuildCfg{
104 + Online: true,
105 + Host: mock.MockHostOption(mn),
106 + })
107 if err != nil {
108 return err
109 }
110 defer adder.Close()
113 - catter, err := core.NewIPFSNode(ctx, core.ConfigOption(MocknetTestRepo(peers[1], mn.Host(peers[1]), conf, core.DHTOption)))
111 +
112 + catter, err := core.NewNode(ctx, &core.BuildCfg{
113 + Online: true,
114 + Host: mock.MockHostOption(mn),
115 + })
116 if err != nil {
117 return err
118 }
119 defer catter.Close()
120
121 + err = mn.LinkAll()
122 + if err != nil {
123 + return err
124 + }
125 +
126 bs1 := []peer.PeerInfo{adder.Peerstore.PeerInfo(adder.Identity)}
127 bs2 := []peer.PeerInfo{catter.Peerstore.PeerInfo(catter.Identity)}
128
test/integration/bench_cat_test.go
+16 -12
@@ -10,6 +10,7 @@ import (
10 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
11 "github.com/ipfs/go-ipfs/core"
12 coreunix "github.com/ipfs/go-ipfs/core/coreunix"
13 + mock "github.com/ipfs/go-ipfs/core/mock"
14 mocknet "github.com/ipfs/go-ipfs/p2p/net/mock"
15 "github.com/ipfs/go-ipfs/p2p/peer"
16 "github.com/ipfs/go-ipfs/thirdparty/unit"
@@ -35,35 +36,38 @@ func benchCat(b *testing.B, data []byte, conf testutil.LatencyConfig) error {
36 b.StopTimer()
37 ctx, cancel := context.WithCancel(context.Background())
38 defer cancel()
38 - const numPeers = 2
39
40 // create network
41 - mn, err := mocknet.FullMeshLinked(ctx, numPeers)
42 - if err != nil {
43 - return err
44 - }
41 + mn := mocknet.New(ctx)
42 mn.SetLinkDefaults(mocknet.LinkOptions{
43 Latency: conf.NetworkLatency,
44 // TODO add to conf. This is tricky because we want 0 values to be functional.
45 Bandwidth: math.MaxInt32,
46 })
47
51 - peers := mn.Peers()
52 - if len(peers) < numPeers {
53 - return errors.New("test initialization error")
54 - }
55 -
56 - adder, err := core.NewIPFSNode(ctx, core.ConfigOption(MocknetTestRepo(peers[0], mn.Host(peers[0]), conf, core.DHTOption)))
48 + adder, err := core.NewNode(ctx, &core.BuildCfg{
49 + Online: true,
50 + Host: mock.MockHostOption(mn),
51 + })
52 if err != nil {
53 return err
54 }
55 defer adder.Close()
61 - catter, err := core.NewIPFSNode(ctx, core.ConfigOption(MocknetTestRepo(peers[1], mn.Host(peers[1]), conf, core.DHTOption)))
56 +
57 + catter, err := core.NewNode(ctx, &core.BuildCfg{
58 + Online: true,
59 + Host: mock.MockHostOption(mn),
60 + })
61 if err != nil {
62 return err
63 }
64 defer catter.Close()
65
66 + err = mn.LinkAll()
67 + if err != nil {
68 + return err
69 + }
70 +
71 bs1 := []peer.PeerInfo{adder.Peerstore.PeerInfo(adder.Identity)}
72 bs2 := []peer.PeerInfo{catter.Peerstore.PeerInfo(catter.Identity)}
73
test/integration/bitswap_wo_routing_test.go
+10 -17
@@ -2,15 +2,13 @@ package integrationtest
2
3 import (
4 "bytes"
5 - "errors"
5 "testing"
7 - "time"
6
7 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
8 "github.com/ipfs/go-ipfs/blocks"
9 "github.com/ipfs/go-ipfs/core"
10 + "github.com/ipfs/go-ipfs/core/mock"
11 mocknet "github.com/ipfs/go-ipfs/p2p/net/mock"
13 - testutil "github.com/ipfs/go-ipfs/util/testutil"
12 )
13
14 func TestBitswapWithoutRouting(t *testing.T) {
@@ -19,22 +17,15 @@ func TestBitswapWithoutRouting(t *testing.T) {
17 const numPeers = 4
18
19 // create network
22 - mn, err := mocknet.FullMeshLinked(ctx, numPeers)
23 - if err != nil {
24 - t.Fatal(err)
25 - }
26 -
27 - peers := mn.Peers()
28 - if len(peers) < numPeers {
29 - t.Fatal(errors.New("test initialization error"))
30 - }
31 -
32 - // set the routing latency to infinity.
33 - conf := testutil.LatencyConfig{RoutingLatency: (525600 * time.Minute)}
20 + mn := mocknet.New(ctx)
21
22 var nodes []*core.IpfsNode
36 - for _, p := range peers {
37 - n, err := core.NewIPFSNode(ctx, core.ConfigOption(MocknetTestRepo(p, mn.Host(p), conf, core.NilRouterOption)))
23 + for i := 0; i < numPeers; i++ {
24 + n, err := core.NewNode(ctx, &core.BuildCfg{
25 + Online: true,
26 + Host: coremock.MockHostOption(mn),
27 + Routing: core.NilRouterOption, // no routing
28 + })
29 if err != nil {
30 t.Fatal(err)
31 }
@@ -42,6 +33,8 @@ func TestBitswapWithoutRouting(t *testing.T) {
33 nodes = append(nodes, n)
34 }
35
36 + mn.LinkAll()
37 +
38 // connect them
39 for _, n1 := range nodes {
40 for _, n2 := range nodes {
test/integration/core.go deleted
-54
@@ -1,54 +0,0 @@
1 -package integrationtest
2 -
3 -import (
4 - "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
5 - syncds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
6 - context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
7 - blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
8 - core "github.com/ipfs/go-ipfs/core"
9 - bitswap "github.com/ipfs/go-ipfs/exchange/bitswap"
10 - bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
11 - host "github.com/ipfs/go-ipfs/p2p/host"
12 - peer "github.com/ipfs/go-ipfs/p2p/peer"
13 - "github.com/ipfs/go-ipfs/repo"
14 - delay "github.com/ipfs/go-ipfs/thirdparty/delay"
15 - eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
16 - ds2 "github.com/ipfs/go-ipfs/util/datastore2"
17 - testutil "github.com/ipfs/go-ipfs/util/testutil"
18 -)
19 -
20 -var log = eventlog.Logger("epictest")
21 -
22 -func MocknetTestRepo(p peer.ID, h host.Host, conf testutil.LatencyConfig, routing core.RoutingOption) core.ConfigOption {
23 - return func(ctx context.Context) (*core.IpfsNode, error) {
24 - const kWriteCacheElems = 100
25 - const alwaysSendToPeer = true
26 - dsDelay := delay.Fixed(conf.BlockstoreLatency)
27 - r := &repo.Mock{
28 - D: ds2.CloserWrap(syncds.MutexWrap(ds2.WithDelay(datastore.NewMapDatastore(), dsDelay))),
29 - }
30 - ds := r.Datastore()
31 -
32 - n := &core.IpfsNode{
33 - Peerstore: h.Peerstore(),
34 - Repo: r,
35 - PeerHost: h,
36 - Identity: p,
37 - }
38 - dhtt, err := routing(ctx, n.PeerHost, n.Repo.Datastore())
39 - if err != nil {
40 - return nil, err
41 - }
42 -
43 - bsn := bsnet.NewFromIpfsHost(h, dhtt)
44 - bstore, err := blockstore.WriteCached(blockstore.NewBlockstore(ds), kWriteCacheElems)
45 - if err != nil {
46 - return nil, err
47 - }
48 - exch := bitswap.New(ctx, p, bsn, bstore, alwaysSendToPeer)
49 - n.Blockstore = bstore
50 - n.Exchange = exch
51 - n.Routing = dhtt
52 - return n, nil
53 - }
54 -}
test/integration/grandcentral_test.go
+15 -19
@@ -16,9 +16,9 @@ import (
16 core "github.com/ipfs/go-ipfs/core"
17 "github.com/ipfs/go-ipfs/core/corerouting"
18 "github.com/ipfs/go-ipfs/core/coreunix"
19 + mock "github.com/ipfs/go-ipfs/core/mock"
20 mocknet "github.com/ipfs/go-ipfs/p2p/net/mock"
21 "github.com/ipfs/go-ipfs/p2p/peer"
21 - "github.com/ipfs/go-ipfs/thirdparty/iter"
22 "github.com/ipfs/go-ipfs/thirdparty/unit"
23 ds2 "github.com/ipfs/go-ipfs/util/datastore2"
24 testutil "github.com/ipfs/go-ipfs/util/testutil"
@@ -82,28 +82,21 @@ func InitializeSupernodeNetwork(
82 conf testutil.LatencyConfig) ([]*core.IpfsNode, []*core.IpfsNode, error) {
83
84 // create network
85 - mn, err := mocknet.FullMeshLinked(ctx, numServers+numClients)
86 - if err != nil {
87 - return nil, nil, err
88 - }
85 + mn := mocknet.New(ctx)
86
87 mn.SetLinkDefaults(mocknet.LinkOptions{
88 Latency: conf.NetworkLatency,
89 Bandwidth: math.MaxInt32,
90 })
91
95 - peers := mn.Peers()
96 - if len(peers) < numServers+numClients {
97 - return nil, nil, errors.New("test initialization error")
98 - }
99 - clientPeers, serverPeers := peers[0:numClients], peers[numClients:]
100 -
92 routingDatastore := ds2.CloserWrap(syncds.MutexWrap(datastore.NewMapDatastore()))
93 var servers []*core.IpfsNode
103 - for i := range iter.N(numServers) {
104 - p := serverPeers[i]
105 - bootstrap, err := core.NewIPFSNode(ctx, MocknetTestRepo(p, mn.Host(p), conf,
106 - corerouting.SupernodeServer(routingDatastore)))
94 + for i := 0; i < numServers; i++ {
95 + bootstrap, err := core.NewNode(ctx, &core.BuildCfg{
96 + Online: true,
97 + Host: mock.MockHostOption(mn),
98 + Routing: corerouting.SupernodeServer(routingDatastore),
99 + })
100 if err != nil {
101 return nil, nil, err
102 }
@@ -117,15 +110,18 @@ func InitializeSupernodeNetwork(
110 }
111
112 var clients []*core.IpfsNode
120 - for i := range iter.N(numClients) {
121 - p := clientPeers[i]
122 - n, err := core.NewIPFSNode(ctx, MocknetTestRepo(p, mn.Host(p), conf,
123 - corerouting.SupernodeClient(bootstrapInfos...)))
113 + for i := 0; i < numClients; i++ {
114 + n, err := core.NewNode(ctx, &core.BuildCfg{
115 + Online: true,
116 + Host: mock.MockHostOption(mn),
117 + Routing: corerouting.SupernodeClient(bootstrapInfos...),
118 + })
119 if err != nil {
120 return nil, nil, err
121 }
122 clients = append(clients, n)
123 }
124 + mn.LinkAll()
125
126 bcfg := core.BootstrapConfigWithPeers(bootstrapInfos)
127 for _, n := range clients {
test/integration/three_legged_cat_test.go
+17 -11
@@ -12,6 +12,7 @@ import (
12
13 core "github.com/ipfs/go-ipfs/core"
14 coreunix "github.com/ipfs/go-ipfs/core/coreunix"
15 + mock "github.com/ipfs/go-ipfs/core/mock"
16 mocknet "github.com/ipfs/go-ipfs/p2p/net/mock"
17 "github.com/ipfs/go-ipfs/p2p/peer"
18 "github.com/ipfs/go-ipfs/thirdparty/unit"
@@ -67,35 +68,40 @@ func RunThreeLeggedCat(data []byte, conf testutil.LatencyConfig) error {
68 const numPeers = 3
69
70 // create network
70 - mn, err := mocknet.FullMeshLinked(ctx, numPeers)
71 - if err != nil {
72 - return err
73 - }
71 + mn := mocknet.New(ctx)
72 mn.SetLinkDefaults(mocknet.LinkOptions{
73 Latency: conf.NetworkLatency,
74 // TODO add to conf. This is tricky because we want 0 values to be functional.
75 Bandwidth: math.MaxInt32,
76 })
77
80 - peers := mn.Peers()
81 - if len(peers) < numPeers {
82 - return errors.New("test initialization error")
83 - }
84 - bootstrap, err := core.NewIPFSNode(ctx, MocknetTestRepo(peers[2], mn.Host(peers[2]), conf, core.DHTOption))
78 + bootstrap, err := core.NewNode(ctx, &core.BuildCfg{
79 + Online: true,
80 + Host: mock.MockHostOption(mn),
81 + })
82 if err != nil {
83 return err
84 }
85 defer bootstrap.Close()
89 - adder, err := core.NewIPFSNode(ctx, MocknetTestRepo(peers[0], mn.Host(peers[0]), conf, core.DHTOption))
86 +
87 + adder, err := core.NewNode(ctx, &core.BuildCfg{
88 + Online: true,
89 + Host: mock.MockHostOption(mn),
90 + })
91 if err != nil {
92 return err
93 }
94 defer adder.Close()
94 - catter, err := core.NewIPFSNode(ctx, MocknetTestRepo(peers[1], mn.Host(peers[1]), conf, core.DHTOption))
95 +
96 + catter, err := core.NewNode(ctx, &core.BuildCfg{
97 + Online: true,
98 + Host: mock.MockHostOption(mn),
99 + })
100 if err != nil {
101 return err
102 }
103 defer catter.Close()
104 + mn.LinkAll()
105
106 bis := bootstrap.Peerstore.PeerInfo(bootstrap.PeerHost.ID())
107 bcfg := core.BootstrapConfigWithPeers([]peer.PeerInfo{bis})
test/supernode_client/main.go
+10 -10
@@ -93,14 +93,11 @@ func run() error {
93 })
94 }
95
96 - node, err := core.NewIPFSNode(
97 - ctx,
98 - core.OnlineWithOptions(
99 - repo,
100 - corerouting.SupernodeClient(infos...),
101 - core.DefaultHostOption,
102 - ),
103 - )
96 + node, err := core.NewNode(ctx, &core.BuildCfg{
97 + Online: true,
98 + Repo: repo,
99 + Routing: corerouting.SupernodeClient(infos...),
100 + })
101 if err != nil {
102 return err
103 }
@@ -168,10 +165,13 @@ func runFileCattingWorker(ctx context.Context, n *core.IpfsNode) error {
165 return err
166 }
167
171 - dummy, err := core.NewIPFSNode(ctx, core.Offline(&repo.Mock{
168 + r := &repo.Mock{
169 D: ds2.CloserWrap(syncds.MutexWrap(datastore.NewMapDatastore())),
170 C: *conf,
174 - }))
171 + }
172 + dummy, err := core.NewNode(ctx, &core.BuildCfg{
173 + Repo: r,
174 + })
175 if err != nil {
176 return err
177 }