@cryptotaxi247 / kubo / commits / 9ab59e44a

implement in memory model for ipns filesystem, to be used as backing for ipns fuse interface

Jeromy committed Mar 8, 2015 at 17:59 UTC 9ab59e44ad25d241ca734553844b47db8866b5b2
6 files changed +888 -3
core/core.go
+21 -1
@@ -37,6 +37,7 @@ import (
37 rp "github.com/jbenet/go-ipfs/exchange/reprovide"
38
39 mount "github.com/jbenet/go-ipfs/fuse/mount"
40 + ipnsfs "github.com/jbenet/go-ipfs/ipnsfs"
41 merkledag "github.com/jbenet/go-ipfs/merkledag"
42 namesys "github.com/jbenet/go-ipfs/namesys"
43 path "github.com/jbenet/go-ipfs/path"
@@ -89,6 +90,8 @@ type IpfsNode struct {
90 Diagnostics *diag.Diagnostics // the diagnostics service
91 Reprovider *rp.Reprovider // the value reprovider system
92
93 + IpnsFs *ipnsfs.Filesystem
94 +
95 ctxgroup.ContextGroup
96
97 mode mode
@@ -138,6 +141,16 @@ func NewIPFSNode(parent context.Context, option ConfigOption) (*IpfsNode, error)
141 node.Pinning = pin.NewPinner(node.Repo.Datastore(), node.DAG)
142 }
143 node.Resolver = &path.Resolver{DAG: node.DAG}
144 +
145 + // Setup the mutable ipns filesystem structure
146 + if node.OnlineMode() {
147 + fs, err := ipnsfs.NewFilesystem(ctx, node.DAG, node.Namesys, node.Pinning, node.PrivateKey)
148 + if err != nil {
149 + return nil, debugerror.Wrap(err)
150 + }
151 + node.IpnsFs = fs
152 + }
153 +
154 success = true
155 return node, nil
156 }
@@ -268,6 +281,7 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost
281
282 // setup name system
283 n.Namesys = namesys.NewNameSystem(n.Routing)
284 +
285 return nil
286 }
287
@@ -278,7 +292,6 @@ func (n *IpfsNode) teardown() error {
292 // owned objects are closed in this teardown to ensure that they're closed
293 // regardless of which constructor was used to add them to the node.
294 closers := []io.Closer{
281 - n.Blocks,
295 n.Exchange,
296 n.Repo,
297 }
@@ -288,6 +301,13 @@ func (n *IpfsNode) teardown() error {
301 }
302 }
303
304 + if n.Blocks != nil {
305 + addCloser(n.Blocks)
306 + }
307 + if n.IpnsFs != nil {
308 + addCloser(n.IpnsFs)
309 + }
310 +
311 addCloser(n.Bootstrapper)
312 if dht, ok := n.Routing.(*dht.IpfsDHT); ok {
313 addCloser(dht)
core/mock.go
+5 -2
@@ -13,8 +13,9 @@ import (
13 mocknet "github.com/jbenet/go-ipfs/p2p/net/mock"
14 peer "github.com/jbenet/go-ipfs/p2p/peer"
15 path "github.com/jbenet/go-ipfs/path"
16 + pin "github.com/jbenet/go-ipfs/pin"
17 "github.com/jbenet/go-ipfs/repo"
17 - mockrouting "github.com/jbenet/go-ipfs/routing/mock"
18 + offrt "github.com/jbenet/go-ipfs/routing/offline"
19 ds2 "github.com/jbenet/go-ipfs/util/datastore2"
20 testutil "github.com/jbenet/go-ipfs/util/testutil"
21 )
@@ -54,7 +55,7 @@ func NewMockNode() (*IpfsNode, error) {
55 }
56
57 // Routing
57 - nd.Routing = mockrouting.NewServer().Client(ident)
58 + nd.Routing = offrt.NewOfflineRouter(nd.Repo.Datastore(), nd.PrivateKey)
59
60 // Bitswap
61 bstore := blockstore.NewBlockstore(nd.Repo.Datastore())
@@ -65,6 +66,8 @@ func NewMockNode() (*IpfsNode, error) {
66
67 nd.DAG = mdag.NewDAGService(bserv)
68
69 + nd.Pinning = pin.NewPinner(nd.Repo.Datastore(), nd.DAG)
70 +
71 // Namespace resolver
72 nd.Namesys = nsys.NewNameSystem(nd.Routing)
73
ipnsfs/dir.go new
+340
@@ -0,0 +1,340 @@
1 +package ipnsfs
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 + "os"
7 + "sync"
8 +
9 + dag "github.com/jbenet/go-ipfs/merkledag"
10 + ft "github.com/jbenet/go-ipfs/unixfs"
11 + ufspb "github.com/jbenet/go-ipfs/unixfs/pb"
12 +)
13 +
14 +type Directory struct {
15 + fs *Filesystem
16 + parent childCloser
17 + childDirs map[string]*Directory
18 + files map[string]*file
19 +
20 + node *dag.Node
21 + name string
22 + lock sync.Mutex
23 +
24 + ref int
25 + refLock sync.Mutex
26 +}
27 +
28 +func NewDirectory(name string, node *dag.Node, parent childCloser, fs *Filesystem) *Directory {
29 + return &Directory{
30 + fs: fs,
31 + name: name,
32 + node: node,
33 + parent: parent,
34 + childDirs: make(map[string]*Directory),
35 + files: make(map[string]*file),
36 + }
37 +}
38 +
39 +func (d *Directory) Open(tpath []string, mode int) (File, error) {
40 + if len(tpath) == 0 {
41 + return nil, ErrIsDirectory
42 + }
43 + if len(tpath) == 1 {
44 + fi, err := d.childFile(tpath[0])
45 + if err == nil {
46 + return fi.withMode(mode), nil
47 + }
48 +
49 + if mode|os.O_CREATE != 0 {
50 + fnode := new(dag.Node)
51 + fnode.Data = ft.FilePBData(nil, 0)
52 + nfi, err := NewFile(tpath[0], fnode, d, d.fs)
53 + if err != nil {
54 + return nil, err
55 + }
56 + d.files[tpath[0]] = nfi
57 + return nfi.withMode(mode), nil
58 + }
59 +
60 + return nil, ErrNoSuch
61 + }
62 +
63 + dir, err := d.childDir(tpath[0])
64 + if err != nil {
65 + return nil, err
66 + }
67 + return dir.Open(tpath[1:], mode)
68 +}
69 +
70 +// consider combining into a single method...
71 +type childCloser interface {
72 + closeChild(string, *dag.Node) error
73 +}
74 +
75 +func (d *Directory) closeChild(name string, nd *dag.Node) error {
76 + _, err := d.fs.dserv.Add(nd)
77 + if err != nil {
78 + return err
79 + }
80 +
81 + d.lock.Lock()
82 + err = d.node.RemoveNodeLink(name)
83 + if err != nil && err != dag.ErrNotFound {
84 + d.lock.Unlock()
85 + return err
86 + }
87 +
88 + err = d.node.AddNodeLinkClean(name, nd)
89 + if err != nil {
90 + d.lock.Unlock()
91 + return err
92 + }
93 + d.lock.Unlock()
94 +
95 + return d.parent.closeChild(d.name, d.node)
96 +}
97 +
98 +func (d *Directory) Type() NodeType {
99 + return TDir
100 +}
101 +
102 +func (d *Directory) childFile(name string) (*file, error) {
103 + fi, ok := d.files[name]
104 + if ok {
105 + return fi, nil
106 + }
107 +
108 + // search dag
109 + for _, lnk := range d.node.Links {
110 + if lnk.Name == name {
111 + nd, err := lnk.GetNode(d.fs.dserv)
112 + if err != nil {
113 + return nil, err
114 + }
115 + i, err := ft.FromBytes(nd.Data)
116 + if err != nil {
117 + return nil, err
118 + }
119 +
120 + switch i.GetType() {
121 + case ufspb.Data_Directory:
122 + return nil, ErrIsDirectory
123 + case ufspb.Data_File:
124 + nfi, err := NewFile(name, nd, d, d.fs)
125 + if err != nil {
126 + return nil, err
127 + }
128 + d.files[name] = nfi
129 + return nfi, nil
130 + case ufspb.Data_Metadata:
131 + panic("NOT YET IMPLEMENTED")
132 + default:
133 + panic("NO!")
134 + }
135 + }
136 + }
137 + return nil, ErrNoSuch
138 +}
139 +
140 +func (d *Directory) childDir(name string) (*Directory, error) {
141 + dir, ok := d.childDirs[name]
142 + if ok {
143 + return dir, nil
144 + }
145 +
146 + for _, lnk := range d.node.Links {
147 + if lnk.Name == name {
148 + nd, err := lnk.GetNode(d.fs.dserv)
149 + if err != nil {
150 + return nil, err
151 + }
152 + i, err := ft.FromBytes(nd.Data)
153 + if err != nil {
154 + return nil, err
155 + }
156 +
157 + switch i.GetType() {
158 + case ufspb.Data_Directory:
159 + ndir := NewDirectory(name, nd, d, d.fs)
160 + d.childDirs[name] = ndir
161 + return ndir, nil
162 + case ufspb.Data_File:
163 + return nil, fmt.Errorf("%s is not a directory", name)
164 + case ufspb.Data_Metadata:
165 + panic("NOT YET IMPLEMENTED")
166 + default:
167 + panic("NO!")
168 + }
169 + }
170 +
171 + }
172 +
173 + return nil, ErrNoSuch
174 +}
175 +
176 +func (d *Directory) Child(name string) (FSNode, error) {
177 + d.lock.Lock()
178 + defer d.lock.Unlock()
179 + dir, err := d.childDir(name)
180 + if err == nil {
181 + return dir, nil
182 + }
183 + fi, err := d.childFile(name)
184 + if err == nil {
185 + return fi, nil
186 + }
187 +
188 + return nil, ErrNoSuch
189 +}
190 +
191 +func (d *Directory) List() []string {
192 + d.lock.Lock()
193 + defer d.lock.Unlock()
194 +
195 + var out []string
196 + for _, lnk := range d.node.Links {
197 + out = append(out, lnk.Name)
198 + }
199 + return out
200 +}
201 +
202 +func (d *Directory) Mkdir(name string) (*Directory, error) {
203 + d.lock.Lock()
204 +
205 + _, err := d.childDir(name)
206 + if err == nil {
207 + d.lock.Unlock()
208 + return nil, errors.New("directory by that name already exists")
209 + }
210 + _, err = d.childFile(name)
211 + if err == nil {
212 + d.lock.Unlock()
213 + return nil, errors.New("file by that name already exists")
214 + }
215 +
216 + ndir := &dag.Node{Data: ft.FolderPBData()}
217 + err = d.node.AddNodeLinkClean(name, ndir)
218 + if err != nil {
219 + d.lock.Unlock()
220 + return nil, err
221 + }
222 + d.lock.Unlock()
223 +
224 + err = d.parent.closeChild(d.name, d.node)
225 + if err != nil {
226 + return nil, err
227 + }
228 +
229 + d.lock.Lock()
230 + defer d.lock.Unlock()
231 +
232 + return d.childDir(name)
233 +}
234 +
235 +func (d *Directory) Unlink(name string) error {
236 + d.lock.Lock()
237 + delete(d.childDirs, name)
238 + delete(d.files, name)
239 +
240 + err := d.node.RemoveNodeLink(name)
241 + if err != nil {
242 + d.lock.Unlock()
243 + return err
244 + }
245 + d.lock.Unlock()
246 +
247 + return d.parent.closeChild(d.name, d.node)
248 +}
249 +
250 +func (d *Directory) RenameEntry(oldname, newname string) error {
251 + dir, err := d.childDir(oldname)
252 + if err == nil {
253 + dir.name = newname
254 +
255 + err := d.node.RemoveNodeLink(oldname)
256 + if err != nil {
257 + return err
258 + }
259 + err = d.node.AddNodeLinkClean(newname, dir.node)
260 + if err != nil {
261 + return err
262 + }
263 +
264 + delete(d.childDirs, oldname)
265 + d.childDirs[newname] = dir
266 + return d.parent.closeChild(d.name, d.node)
267 + }
268 +
269 + fi, err := d.childFile(oldname)
270 + if err == nil {
271 + fi.name = newname
272 +
273 + err := d.node.RemoveNodeLink(oldname)
274 + if err != nil {
275 + return err
276 + }
277 +
278 + nd, err := fi.GetNode()
279 + if err != nil {
280 + return err
281 + }
282 +
283 + err = d.node.AddNodeLinkClean(newname, nd)
284 + if err != nil {
285 + return err
286 + }
287 +
288 + delete(d.childDirs, oldname)
289 + d.files[newname] = fi
290 + return d.parent.closeChild(d.name, d.node)
291 + }
292 + return ErrNoSuch
293 +}
294 +
295 +func (d *Directory) AddChild(name string, nd *dag.Node) error {
296 + pbn, err := ft.FromBytes(nd.Data)
297 + if err != nil {
298 + return err
299 + }
300 +
301 + _, err = d.Child(name)
302 + if err == nil {
303 + return errors.New("directory already has entry by that name")
304 + }
305 +
306 + err = d.node.AddNodeLinkClean(name, nd)
307 + if err != nil {
308 + return err
309 + }
310 +
311 + switch pbn.GetType() {
312 + case ft.TDirectory:
313 + d.childDirs[name] = NewDirectory(name, nd, d, d.fs)
314 + case ft.TFile, ft.TMetadata, ft.TRaw:
315 + nfi, err := NewFile(name, nd, d, d.fs)
316 + if err != nil {
317 + return err
318 + }
319 + d.files[name] = nfi
320 + default:
321 + panic("invalid unixfs node")
322 + }
323 + return d.parent.closeChild(d.name, d.node)
324 +}
325 +
326 +func (d *Directory) GetNode() (*dag.Node, error) {
327 + return d.node, nil
328 +}
329 +
330 +func (d *Directory) Upref() {
331 + d.refLock.Lock()
332 + d.ref++
333 + d.refLock.Unlock()
334 +}
335 +
336 +func (d *Directory) Deref() {
337 + d.refLock.Lock()
338 + d.ref--
339 + d.refLock.Unlock()
340 +}
ipnsfs/file.go new
+145
@@ -0,0 +1,145 @@
1 +package ipnsfs
2 +
3 +import (
4 + "errors"
5 + "io"
6 + "os"
7 + "sync"
8 +
9 + chunk "github.com/jbenet/go-ipfs/importer/chunk"
10 + dag "github.com/jbenet/go-ipfs/merkledag"
11 + mod "github.com/jbenet/go-ipfs/unixfs/mod"
12 +
13 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
14 +)
15 +
16 +type File interface {
17 + io.ReadWriteCloser
18 + io.WriterAt
19 + Seek(int64, int) (int64, error)
20 + Size() (int64, error)
21 + Flush() error
22 + Truncate(int64) error
23 + FSNode
24 +}
25 +
26 +type file struct {
27 + parent childCloser
28 + fs *Filesystem
29 +
30 + name string
31 + hasChanges bool
32 +
33 + // TODO: determine whether or not locking here is actually required...
34 + lk sync.Mutex
35 + mod *mod.DagModifier
36 +}
37 +
38 +func NewFile(name string, node *dag.Node, parent childCloser, fs *Filesystem) (*file, error) {
39 + dmod, err := mod.NewDagModifier(context.TODO(), node, fs.dserv, fs.pins.GetManual(), chunk.DefaultSplitter)
40 + if err != nil {
41 + return nil, err
42 + }
43 +
44 + return &file{
45 + fs: fs,
46 + parent: parent,
47 + name: name,
48 + mod: dmod,
49 + }, nil
50 +}
51 +
52 +func (fi *file) Write(b []byte) (int, error) {
53 + fi.lk.Lock()
54 + defer fi.lk.Unlock()
55 + fi.hasChanges = true
56 + return fi.mod.Write(b)
57 +}
58 +
59 +func (fi *file) Read(b []byte) (int, error) {
60 + fi.lk.Lock()
61 + defer fi.lk.Unlock()
62 + return fi.mod.Read(b)
63 +}
64 +
65 +func (fi *file) Close() error {
66 + fi.lk.Lock()
67 + defer fi.lk.Unlock()
68 + if fi.hasChanges {
69 + err := fi.mod.Flush()
70 + if err != nil {
71 + return err
72 + }
73 +
74 + nd, err := fi.mod.GetNode()
75 + if err != nil {
76 + return err
77 + }
78 +
79 + err = fi.parent.closeChild(fi.name, nd)
80 + if err != nil {
81 + return err
82 + }
83 +
84 + fi.hasChanges = false
85 + }
86 +
87 + return nil
88 +}
89 +
90 +func (fi *file) Flush() error {
91 + fi.lk.Lock()
92 + defer fi.lk.Unlock()
93 + return fi.mod.Flush()
94 +}
95 +
96 +func (fi *file) withMode(mode int) File {
97 + if mode == os.O_RDONLY {
98 + return &readOnlyFile{fi}
99 + }
100 + return fi
101 +}
102 +
103 +func (fi *file) Seek(offset int64, whence int) (int64, error) {
104 + fi.lk.Lock()
105 + defer fi.lk.Unlock()
106 + return fi.mod.Seek(offset, whence)
107 +}
108 +
109 +func (fi *file) WriteAt(b []byte, at int64) (int, error) {
110 + fi.lk.Lock()
111 + defer fi.lk.Unlock()
112 + fi.hasChanges = true
113 + return fi.mod.WriteAt(b, at)
114 +}
115 +
116 +func (fi *file) Size() (int64, error) {
117 + fi.lk.Lock()
118 + defer fi.lk.Unlock()
119 + return fi.mod.Size()
120 +}
121 +
122 +func (fi *file) GetNode() (*dag.Node, error) {
123 + fi.lk.Lock()
124 + defer fi.lk.Unlock()
125 + return fi.mod.GetNode()
126 +}
127 +
128 +func (fi *file) Truncate(size int64) error {
129 + fi.lk.Lock()
130 + defer fi.lk.Unlock()
131 + fi.hasChanges = true
132 + return fi.mod.Truncate(size)
133 +}
134 +
135 +func (fi *file) Type() NodeType {
136 + return TFile
137 +}
138 +
139 +type readOnlyFile struct {
140 + *file
141 +}
142 +
143 +func (ro *readOnlyFile) Write([]byte) (int, error) {
144 + return 0, errors.New("permission denied: file readonly")
145 +}
ipnsfs/system.go new
+285
@@ -0,0 +1,285 @@
1 +package ipnsfs
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 + "strings"
7 + "time"
8 +
9 + dag "github.com/jbenet/go-ipfs/merkledag"
10 + namesys "github.com/jbenet/go-ipfs/namesys"
11 + ci "github.com/jbenet/go-ipfs/p2p/crypto"
12 + pin "github.com/jbenet/go-ipfs/pin"
13 + ft "github.com/jbenet/go-ipfs/unixfs"
14 + u "github.com/jbenet/go-ipfs/util"
15 +
16 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
17 + eventlog "github.com/jbenet/go-ipfs/thirdparty/eventlog"
18 +)
19 +
20 +var log = eventlog.Logger("ipnsfs")
21 +
22 +var ErrIsDirectory = errors.New("error: is a directory")
23 +
24 +var ErrNoSuch = errors.New("no such file or directory")
25 +
26 +// Filesystem is the writeable fuse filesystem structure
27 +type Filesystem struct {
28 + dserv dag.DAGService
29 +
30 + nsys namesys.NameSystem
31 +
32 + pins pin.Pinner
33 +
34 + roots map[string]*KeyRoot
35 +}
36 +
37 +func NewFilesystem(ctx context.Context, ds dag.DAGService, nsys namesys.NameSystem, pins pin.Pinner, keys ...ci.PrivKey) (*Filesystem, error) {
38 + roots := make(map[string]*KeyRoot)
39 + fs := &Filesystem{
40 + roots: roots,
41 + nsys: nsys,
42 + dserv: ds,
43 + pins: pins,
44 + }
45 + for _, k := range keys {
46 + pkh, err := k.GetPublic().Hash()
47 + if err != nil {
48 + return nil, err
49 + }
50 +
51 + root, err := fs.NewKeyRoot(ctx, k)
52 + if err != nil {
53 + return nil, err
54 + }
55 + roots[u.Key(pkh).Pretty()] = root
56 + }
57 +
58 + return fs, nil
59 +}
60 +
61 +func (fs *Filesystem) Open(tpath string, mode int) (File, error) {
62 + pathelem := strings.Split(tpath, "/")
63 + r, ok := fs.roots[pathelem[0]]
64 + if !ok {
65 + return nil, ErrNoSuch
66 + }
67 +
68 + return r.Open(pathelem[1:], mode)
69 +}
70 +
71 +func (fs *Filesystem) Close() error {
72 + for _, r := range fs.roots {
73 + err := r.Publish(context.TODO())
74 + if err != nil {
75 + return err
76 + }
77 + }
78 + return nil
79 +}
80 +
81 +func (fs *Filesystem) GetRoot(name string) (*KeyRoot, error) {
82 + r, ok := fs.roots[name]
83 + if ok {
84 + return r, nil
85 + }
86 + return nil, ErrNoSuch
87 +}
88 +
89 +type NodeType int
90 +
91 +const (
92 + TFile NodeType = iota
93 + TDir
94 +)
95 +
96 +type FSNode interface {
97 + GetNode() (*dag.Node, error)
98 + Type() NodeType
99 +}
100 +
101 +// KeyRoot represents the root of a filesystem tree pointed to by a given keypair
102 +type KeyRoot struct {
103 + key ci.PrivKey
104 +
105 + // node is the merkledag node pointed to by this keypair
106 + node *dag.Node
107 +
108 + // A pointer to the filesystem to access components
109 + fs *Filesystem
110 +
111 + // val represents the node pointed to by this key. It can either be a File or a Directory
112 + val FSNode
113 +
114 + repub *Republisher
115 +}
116 +
117 +func (fs *Filesystem) NewKeyRoot(parent context.Context, k ci.PrivKey) (*KeyRoot, error) {
118 + hash, err := k.GetPublic().Hash()
119 + if err != nil {
120 + return nil, err
121 + }
122 +
123 + name := u.Key(hash).Pretty()
124 +
125 + root := new(KeyRoot)
126 + root.key = k
127 + root.fs = fs
128 +
129 + ctx, cancel := context.WithCancel(parent)
130 + defer cancel()
131 +
132 + pointsTo, err := fs.nsys.Resolve(ctx, name)
133 + if err != nil {
134 + err = namesys.InitializeKeyspace(ctx, fs.dserv, fs.nsys, fs.pins, k)
135 + if err != nil {
136 + return nil, err
137 + }
138 +
139 + pointsTo, err = fs.nsys.Resolve(ctx, name)
140 + if err != nil {
141 + return nil, err
142 + }
143 + }
144 +
145 + mnode, err := fs.dserv.Get(pointsTo)
146 + if err != nil {
147 + return nil, err
148 + }
149 +
150 + root.node = mnode
151 +
152 + root.repub = NewRepublisher(root, time.Millisecond*300, time.Second*3)
153 + go root.repub.Run(parent)
154 +
155 + pbn, err := ft.FromBytes(mnode.Data)
156 + if err != nil {
157 + log.Error("IPNS pointer was not unixfs node")
158 + return nil, err
159 + }
160 +
161 + switch pbn.GetType() {
162 + case ft.TDirectory:
163 + root.val = NewDirectory(pointsTo.B58String(), mnode, root, fs)
164 + case ft.TFile, ft.TMetadata, ft.TRaw:
165 + fi, err := NewFile(pointsTo.B58String(), mnode, root, fs)
166 + if err != nil {
167 + return nil, err
168 + }
169 + root.val = fi
170 + default:
171 + panic("unrecognized! (NYI)")
172 + }
173 + return root, nil
174 +}
175 +
176 +func (kr *KeyRoot) GetValue() FSNode {
177 + return kr.val
178 +}
179 +
180 +func (kr *KeyRoot) Open(tpath []string, mode int) (File, error) {
181 + if kr.val == nil {
182 + // No entry here... what should we do?
183 + panic("nyi")
184 + }
185 + if len(tpath) > 0 {
186 + // Make sure our root is a directory
187 + dir, ok := kr.val.(*Directory)
188 + if !ok {
189 + return nil, fmt.Errorf("no such file or directory: %s", tpath[0])
190 + }
191 +
192 + return dir.Open(tpath, mode)
193 + }
194 +
195 + switch t := kr.val.(type) {
196 + case *Directory:
197 + return nil, ErrIsDirectory
198 + case File:
199 + return t, nil
200 + default:
201 + panic("unrecognized type, should not happen")
202 + }
203 +}
204 +
205 +// closeChild implements the childCloser interface, and signals to the publisher that
206 +// there are changes ready to be published
207 +func (kr *KeyRoot) closeChild(name string, nd *dag.Node) error {
208 + kr.repub.Touch()
209 + return nil
210 +}
211 +
212 +// Publish publishes the ipns entry associated with this key
213 +func (kr *KeyRoot) Publish(ctx context.Context) error {
214 + child, ok := kr.val.(FSNode)
215 + if !ok {
216 + return errors.New("child of key root not valid type")
217 + }
218 +
219 + nd, err := child.GetNode()
220 + if err != nil {
221 + return err
222 + }
223 +
224 + k, err := kr.fs.dserv.Add(nd)
225 + if err != nil {
226 + return err
227 + }
228 +
229 + fmt.Println("Publishing!")
230 + return kr.fs.nsys.Publish(ctx, kr.key, k)
231 +}
232 +
233 +// Republisher manages when to publish the ipns entry associated with a given key
234 +type Republisher struct {
235 + TimeoutLong time.Duration
236 + TimeoutShort time.Duration
237 + Publish chan struct{}
238 + root *KeyRoot
239 +}
240 +
241 +func NewRepublisher(root *KeyRoot, tshort, tlong time.Duration) *Republisher {
242 + return &Republisher{
243 + TimeoutShort: tshort,
244 + TimeoutLong: tlong,
245 + Publish: make(chan struct{}, 1),
246 + root: root,
247 + }
248 +}
249 +
250 +func (np *Republisher) Touch() {
251 + select {
252 + case np.Publish <- struct{}{}:
253 + default:
254 + }
255 +}
256 +
257 +func (np *Republisher) Run(ctx context.Context) {
258 + for {
259 + select {
260 + case <-np.Publish:
261 + quick := time.After(np.TimeoutShort)
262 + longer := time.After(np.TimeoutLong)
263 +
264 + wait:
265 + select {
266 + case <-quick:
267 + case <-longer:
268 + case <-ctx.Done():
269 + return
270 + case <-np.Publish:
271 + quick = time.After(np.TimeoutShort)
272 + goto wait
273 + }
274 +
275 + log.Info("Publishing Changes!")
276 + err := np.root.Publish(ctx)
277 + if err != nil {
278 + log.Critical("republishRoot error: %s", err)
279 + }
280 +
281 + case <-ctx.Done():
282 + return
283 + }
284 + }
285 +}
ipnsfs/system_test.go new
+92
@@ -0,0 +1,92 @@
1 +package ipnsfs_test
2 +
3 +import (
4 + "bytes"
5 + "io/ioutil"
6 + "os"
7 + "path"
8 + "testing"
9 +
10 + core "github.com/jbenet/go-ipfs/core"
11 + . "github.com/jbenet/go-ipfs/ipnsfs"
12 + u "github.com/jbenet/go-ipfs/util"
13 +)
14 +
15 +func testFS(t *testing.T, nd *core.IpfsNode) *Filesystem {
16 + fs, err := NewFilesystem(nd.Context(), nd.DAG, nd.Namesys, nd.Pinning, nd.PrivateKey)
17 + if err != nil {
18 + t.Fatal(err)
19 + }
20 +
21 + return fs
22 +}
23 +
24 +func TestBasic(t *testing.T) {
25 + mock, err := core.NewMockNode()
26 + if err != nil {
27 + t.Fatal(err)
28 + }
29 +
30 + fs := testFS(t, mock)
31 +
32 + k := u.Key(mock.Identity)
33 + p := path.Join(k.B58String(), "file")
34 + fi, err := fs.Open(p, os.O_CREATE)
35 + if err != nil {
36 + t.Fatal(err)
37 + }
38 +
39 + data := []byte("Hello World")
40 + n, err := fi.Write(data)
41 + if err != nil {
42 + t.Fatal(err)
43 + }
44 +
45 + if n != len(data) {
46 + t.Fatal("wrote incorrect amount")
47 + }
48 +
49 + err = fi.Close()
50 + if err != nil {
51 + t.Fatal(err)
52 + }
53 +
54 + nfi, err := fs.Open(p, os.O_RDONLY)
55 + if err != nil {
56 + t.Fatal(err)
57 + }
58 +
59 + out, err := ioutil.ReadAll(nfi)
60 + if err != nil {
61 + t.Fatal(err)
62 + }
63 +
64 + err = nfi.Close()
65 + if err != nil {
66 + t.Fatal(err)
67 + }
68 +
69 + if !bytes.Equal(out, data) {
70 + t.Fatal("Write failed.")
71 + }
72 +
73 + err = fs.Close()
74 + if err != nil {
75 + t.Fatal(err)
76 + }
77 +
78 + // Open the filesystem again, and try to read our file
79 + nfs := testFS(t, mock)
80 +
81 + fi, err = nfs.Open(p, os.O_RDONLY)
82 + nb, err := ioutil.ReadAll(fi)
83 + if err != nil {
84 + t.Fatal(err)
85 + }
86 +
87 + t.Log(nb)
88 +
89 + if !bytes.Equal(nb, data) {
90 + t.Fatal("data not the same after closing down fs")
91 + }
92 +}