-
Notifications
You must be signed in to change notification settings - Fork 337
/
traversal.go
96 lines (85 loc) · 2.97 KB
/
traversal.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
// Copyright 2020 The Swarm Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Package traversal provides abstraction and implementation
// needed to traverse all chunks below a given root hash.
// It tries to parse all manifests and collections in its
// attempt to log all chunk addresses on the way.
package traversal
import (
"context"
"errors"
"fmt"
"github.com/ethersphere/bee/pkg/file/joiner"
"github.com/ethersphere/bee/pkg/file/loadsave"
"github.com/ethersphere/bee/pkg/manifest"
"github.com/ethersphere/bee/pkg/manifest/mantaray"
"github.com/ethersphere/bee/pkg/soc"
"github.com/ethersphere/bee/pkg/storage"
"github.com/ethersphere/bee/pkg/swarm"
)
// Traverser represents service which traverse through address dependent chunks.
type Traverser interface {
// Traverse iterates through each address related to the supplied one, if possible.
Traverse(context.Context, swarm.Address, swarm.AddressIterFunc) error
}
type PutGetter interface {
storage.Putter
storage.Getter
}
// New constructs for a new Traverser.
func New(store PutGetter) Traverser {
return &service{store: store}
}
// service is implementation of Traverser using storage.Storer as its storage.
type service struct {
store PutGetter
}
// Traverse implements Traverser.Traverse method.
func (s *service) Traverse(ctx context.Context, addr swarm.Address, iterFn swarm.AddressIterFunc) error {
processBytes := func(ref swarm.Address) error {
j, _, err := joiner.New(ctx, s.store, ref)
if err != nil {
return fmt.Errorf("traversal: joiner error on %q: %w", ref, err)
}
err = j.IterateChunkAddresses(iterFn)
if err != nil {
return fmt.Errorf("traversal: iterate chunk address error for %q: %w", ref, err)
}
return nil
}
// skip SOC check for encrypted references
if addr.IsValidLength() {
ch, err := s.store.Get(ctx, storage.ModeGetRequest, addr)
if err != nil {
return fmt.Errorf("traversal: failed to get root chunk %s: %w", addr.String(), err)
}
if soc.Valid(ch) {
// if this is a SOC, the traversal will be just be the single chunk
return iterFn(addr)
}
}
ls := loadsave.NewReadonly(s.store)
switch mf, err := manifest.NewDefaultManifestReference(addr, ls); {
case errors.Is(err, manifest.ErrInvalidManifestType):
break
case err != nil:
return fmt.Errorf("traversal: unable to create manifest reference for %q: %w", addr, err)
default:
err := mf.IterateAddresses(ctx, processBytes)
if errors.Is(err, mantaray.ErrTooShort) || errors.Is(err, mantaray.ErrInvalidVersionHash) {
// Based on the returned errors we conclude that it might
// not be a manifest, so we try non-manifest processing.
break
}
if err != nil {
return fmt.Errorf("traversal: unable to process bytes for %q: %w", addr, err)
}
return nil
}
// Non-manifest processing.
if err := processBytes(addr); err != nil {
return fmt.Errorf("traversal: unable to process bytes for %q: %w", addr, err)
}
return nil
}