refactor(core) Close in teardown
This declarative style is simpler to compose than the imperative wiring up of objects. + pass context to StartOnlineServices as parameter. one by one, trying to remove dependencies on node state so these initialization steps can be broken down.
Brian Tiger Chow committed
Jan 15, 2015 at 16:06 UTC
1edc5c0a65c12d91982cff8da516a2f75f5939ed
1 file changed
+40
-23
core/core.go
+40
-23
@@ -2,6 +2,7 @@ package core
2
3
import (
4
"fmt"
5
+ "io"
6
"time"
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
@@ -100,11 +101,22 @@ type Mounts struct {
101
102
type ConfigOption func(ctx context.Context) (*IpfsNode, error)
103
103
-func NewIPFSNode(ctx context.Context, option ConfigOption) (*IpfsNode, error) {
104
+func NewIPFSNode(parent context.Context, option ConfigOption) (*IpfsNode, error) {
105
+ ctxg := ctxgroup.WithContext(parent)
106
+ ctx := ctxg.Context()
107
+ success := false // flip to true after all sub-system inits succeed
108
+ defer func() {
109
+ if !success {
110
+ ctxg.Close()
111
+ }
112
+ }()
113
+
114
node, err := option(ctx)
115
if err != nil {
116
return nil, err
117
}
118
+ node.ContextGroup = ctxg
119
+ ctxg.SetTeardown(node.teardown)
120
121
// Need to make sure it's perfectly clear 1) which variables are expected
122
// to be initialized at this point, and 2) which variables will be
@@ -123,6 +135,7 @@ func NewIPFSNode(ctx context.Context, option ConfigOption) (*IpfsNode, error) {
135
node.Pinning = pin.NewPinner(node.Repo.Datastore(), node.DAG)
136
}
137
node.Resolver = &path.Resolver{DAG: node.DAG}
138
+ success = true
139
return node, nil
140
}
141
@@ -138,13 +151,6 @@ func Online(r repo.Repo) ConfigOption {
151
func Standard(r repo.Repo, online bool) ConfigOption {
152
return func(ctx context.Context) (n *IpfsNode, err error) {
153
141
- success := false // flip to true after all sub-system inits succeed
142
- defer func() {
143
- if !success && n != nil {
144
- n.Close()
145
- }
146
- }()
147
-
154
if r == nil {
155
return nil, debugerror.Errorf("repo required")
156
}
@@ -158,9 +164,6 @@ func Standard(r repo.Repo, online bool) ConfigOption {
164
Repo: r,
165
}
166
161
- n.ContextGroup = ctxgroup.WithContextAndTeardown(ctx, n.teardown)
162
- ctx = n.ContextGroup.Context()
163
-
167
// setup Peerstore
168
n.Peerstore = peer.NewPeerstore()
169
@@ -176,20 +179,18 @@ func Standard(r repo.Repo, online bool) ConfigOption {
179
180
// setup online services
181
if online {
179
- if err := n.StartOnlineServices(); err != nil {
182
+ if err := n.StartOnlineServices(ctx); err != nil {
183
return nil, err // debugerror.Wraps.
184
}
185
} else {
186
n.Exchange = offline.Exchange(n.Blockstore)
187
}
188
186
- success = true
189
return n, nil
190
}
191
}
192
191
-func (n *IpfsNode) StartOnlineServices() error {
192
- ctx := n.Context()
193
+func (n *IpfsNode) StartOnlineServices(ctx context.Context) error {
194
195
if n.PeerHost != nil { // already online.
196
return debugerror.New("node already online")
@@ -200,7 +201,7 @@ func (n *IpfsNode) StartOnlineServices() error {
201
return err
202
}
203
203
- peerhost, err := constructPeerHost(ctx, n.ContextGroup, n.Repo.Config(), n.Identity, n.Peerstore)
204
+ peerhost, err := constructPeerHost(ctx, n.Repo.Config(), n.Identity, n.Peerstore)
205
if err != nil {
206
return debugerror.Wrap(err)
207
}
@@ -210,7 +211,7 @@ func (n *IpfsNode) StartOnlineServices() error {
211
n.Diagnostics = diag.NewDiagnostics(n.Identity, n.PeerHost)
212
213
// setup routing service
213
- dhtRouting, err := constructDHTRouting(ctx, n.ContextGroup, n.PeerHost, n.Repo.Datastore())
214
+ dhtRouting, err := constructDHTRouting(ctx, n.PeerHost, n.Repo.Datastore())
215
if err != nil {
216
return debugerror.Wrap(err)
217
}
@@ -250,9 +251,27 @@ func (n *IpfsNode) StartOnlineServices() error {
251
return nil
252
}
253
254
+// teardown closes children
255
func (n *IpfsNode) teardown() error {
254
- if err := n.Repo.Close(); err != nil {
255
- return err
256
+ var errs []error
257
+ closers := []io.Closer{
258
+ n.Repo,
259
+ }
260
+ if n.DHT != nil {
261
+ closers = append(closers, n.DHT)
262
+ }
263
+ if n.PeerHost != nil {
264
+ closers = append(closers, n.PeerHost)
265
+ }
266
+ for _, closer := range closers {
267
+ if closer != nil {
268
+ if err := closer.Close(); err != nil {
269
+ errs = append(errs, err)
270
+ }
271
+ }
272
+ }
273
+ if len(errs) > 0 {
274
+ return errs[0]
275
}
276
return nil
277
}
@@ -351,7 +370,7 @@ func listenAddresses(cfg *config.Config) ([]ma.Multiaddr, error) {
370
}
371
372
// isolates the complex initialization steps
354
-func constructPeerHost(ctx context.Context, ctxg ctxgroup.ContextGroup, cfg *config.Config, id peer.ID, ps peer.Peerstore) (p2phost.Host, error) {
373
+func constructPeerHost(ctx context.Context, cfg *config.Config, id peer.ID, ps peer.Peerstore) (p2phost.Host, error) {
374
listenAddrs, err := listenAddresses(cfg)
375
if err != nil {
376
return nil, debugerror.Wrap(err)
@@ -369,7 +388,6 @@ func constructPeerHost(ctx context.Context, ctxg ctxgroup.ContextGroup, cfg *con
388
if err != nil {
389
return nil, debugerror.Wrap(err)
390
}
372
- ctxg.AddChildGroup(network.CtxGroup())
391
392
peerhost := p2pbhost.New(network)
393
// explicitly set these as our listen addrs.
@@ -384,9 +402,8 @@ func constructPeerHost(ctx context.Context, ctxg ctxgroup.ContextGroup, cfg *con
402
return peerhost, nil
403
}
404
387
-func constructDHTRouting(ctx context.Context, ctxg ctxgroup.ContextGroup, host p2phost.Host, ds datastore.ThreadSafeDatastore) (*dht.IpfsDHT, error) {
405
+func constructDHTRouting(ctx context.Context, host p2phost.Host, ds datastore.ThreadSafeDatastore) (*dht.IpfsDHT, error) {
406
dhtRouting := dht.NewDHT(ctx, host, ds)
407
dhtRouting.Validators[IpnsValidatorTag] = namesys.ValidateIpnsRecord
390
- ctxg.AddChildGroup(dhtRouting)
408
return dhtRouting, nil
409
}