139 lines
3.8 KiB
Go
139 lines
3.8 KiB
Go
package blobstor
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobovnicza"
|
|
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/fstree"
|
|
"github.com/nspcc-dev/neofs-sdk-go/object"
|
|
addressSDK "github.com/nspcc-dev/neofs-sdk-go/object/address"
|
|
)
|
|
|
|
// IterationElement represents a unit of elements through which Iterate operation passes.
|
|
type IterationElement struct {
|
|
data []byte
|
|
|
|
blzID *blobovnicza.ID
|
|
}
|
|
|
|
// ObjectData returns the stored object in a binary representation.
|
|
func (x IterationElement) ObjectData() []byte {
|
|
return x.data
|
|
}
|
|
|
|
// BlobovniczaID returns the identifier of Blobovnicza in which object is stored.
|
|
// Returns nil if the object isn't in Blobovnicza.
|
|
func (x IterationElement) BlobovniczaID() *blobovnicza.ID {
|
|
return x.blzID
|
|
}
|
|
|
|
// IterationHandler is a generic processor of IterationElement.
|
|
type IterationHandler func(IterationElement) error
|
|
|
|
// IteratePrm groups the parameters of Iterate operation.
|
|
type IteratePrm struct {
|
|
handler IterationHandler
|
|
ignoreErrors bool
|
|
}
|
|
|
|
// IterateRes groups the resulting values of Iterate operation.
|
|
type IterateRes struct{}
|
|
|
|
// SetIterationHandler sets the action to be performed on each iteration.
|
|
func (i *IteratePrm) SetIterationHandler(h IterationHandler) {
|
|
i.handler = h
|
|
}
|
|
|
|
// IgnoreErrors sets the flag signifying whether errors should be ignored.
|
|
func (i *IteratePrm) IgnoreErrors() {
|
|
i.ignoreErrors = true
|
|
}
|
|
|
|
// Iterate traverses the storage over the stored objects and calls the handler
|
|
// on each element.
|
|
//
|
|
// Returns any error encountered that
|
|
// did not allow to completely iterate over the storage.
|
|
//
|
|
// If handler returns an error, method wraps and returns it immediately.
|
|
func (b *BlobStor) Iterate(prm IteratePrm) (*IterateRes, error) {
|
|
var elem IterationElement
|
|
|
|
err := b.blobovniczas.iterateBlobovniczas(prm.ignoreErrors, func(p string, blz *blobovnicza.Blobovnicza) error {
|
|
err := blobovnicza.IterateObjects(blz, func(data []byte) error {
|
|
var err error
|
|
|
|
// decompress the data
|
|
elem.data, err = b.decompressor(data)
|
|
if err != nil {
|
|
if prm.ignoreErrors {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("could not decompress object data: %w", err)
|
|
}
|
|
|
|
elem.blzID = blobovnicza.NewIDFromBytes([]byte(p))
|
|
|
|
return prm.handler(elem)
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("blobovnicza iterator failure %s: %w", p, err)
|
|
}
|
|
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("blobovniczas iterator failure: %w", err)
|
|
}
|
|
|
|
elem.blzID = nil
|
|
|
|
err = b.fsTree.Iterate(new(fstree.IterationPrm).WithHandler(func(_ *addressSDK.Address, data []byte) error {
|
|
// decompress the data
|
|
elem.data, err = b.decompressor(data)
|
|
if err != nil {
|
|
if prm.ignoreErrors {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("could not decompress object data: %w", err)
|
|
}
|
|
|
|
return prm.handler(elem)
|
|
}).WithIgnoreErrors(prm.ignoreErrors))
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("fs tree iterator failure: %w", err)
|
|
}
|
|
|
|
return new(IterateRes), nil
|
|
}
|
|
|
|
// IterateBinaryObjects is a helper function which iterates over BlobStor and passes binary objects to f.
|
|
func IterateBinaryObjects(blz *BlobStor, f func(data []byte, blzID *blobovnicza.ID) error) error {
|
|
var prm IteratePrm
|
|
|
|
prm.SetIterationHandler(func(elem IterationElement) error {
|
|
return f(elem.ObjectData(), elem.BlobovniczaID())
|
|
})
|
|
|
|
_, err := blz.Iterate(prm)
|
|
|
|
return err
|
|
}
|
|
|
|
// IterateObjects is a helper function which iterates over BlobStor and passes decoded objects to f.
|
|
func IterateObjects(blz *BlobStor, f func(obj *object.Object, blzID *blobovnicza.ID) error) error {
|
|
var obj *object.Object
|
|
|
|
return IterateBinaryObjects(blz, func(data []byte, blzID *blobovnicza.ID) error {
|
|
if obj == nil {
|
|
obj = object.New()
|
|
}
|
|
|
|
if err := obj.Unmarshal(data); err != nil {
|
|
return fmt.Errorf("could not unmarshal the object: %w", err)
|
|
}
|
|
|
|
return f(obj, blzID)
|
|
})
|
|
}
|