b7a4dbb07f
And use the new CidBuilder from the spec actors. This patch does not switch over to inline CIDs by default, but paves the way.
136 lines
3.3 KiB
Go
136 lines
3.3 KiB
Go
package ipfsbstore
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"io/ioutil"
|
|
|
|
"golang.org/x/xerrors"
|
|
|
|
"github.com/multiformats/go-multiaddr"
|
|
"github.com/multiformats/go-multihash"
|
|
|
|
blocks "github.com/ipfs/go-block-format"
|
|
"github.com/ipfs/go-cid"
|
|
httpapi "github.com/ipfs/go-ipfs-http-client"
|
|
iface "github.com/ipfs/interface-go-ipfs-core"
|
|
"github.com/ipfs/interface-go-ipfs-core/options"
|
|
"github.com/ipfs/interface-go-ipfs-core/path"
|
|
|
|
"github.com/filecoin-project/lotus/lib/blockstore"
|
|
)
|
|
|
|
type IpfsBstore struct {
|
|
ctx context.Context
|
|
api iface.CoreAPI
|
|
}
|
|
|
|
func NewIpfsBstore(ctx context.Context) (*IpfsBstore, error) {
|
|
localApi, err := httpapi.NewLocalApi()
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("getting local ipfs api: %w", err)
|
|
}
|
|
api, err := localApi.WithOptions(options.Api.Offline(true))
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("setting offline mode: %s", err)
|
|
}
|
|
|
|
return &IpfsBstore{
|
|
ctx: ctx,
|
|
api: api,
|
|
}, nil
|
|
}
|
|
|
|
func NewRemoteIpfsBstore(ctx context.Context, maddr multiaddr.Multiaddr) (*IpfsBstore, error) {
|
|
httpApi, err := httpapi.NewApi(maddr)
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("setting remote ipfs api: %w", err)
|
|
}
|
|
api, err := httpApi.WithOptions(options.Api.Offline(true))
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("applying offline mode: %s", err)
|
|
}
|
|
|
|
return &IpfsBstore{
|
|
ctx: ctx,
|
|
api: api,
|
|
}, nil
|
|
}
|
|
|
|
func (i *IpfsBstore) DeleteBlock(cid cid.Cid) error {
|
|
return xerrors.Errorf("not supported")
|
|
}
|
|
|
|
func (i *IpfsBstore) Has(cid cid.Cid) (bool, error) {
|
|
_, err := i.api.Block().Stat(i.ctx, path.IpldPath(cid))
|
|
if err != nil {
|
|
// The underlying client is running in Offline mode.
|
|
// Stat() will fail with an err if the block isn't in the
|
|
// blockstore. If that's the case, return false without
|
|
// an error since that's the original intention of this method.
|
|
if err.Error() == "blockservice: key not found" {
|
|
return false, nil
|
|
}
|
|
return false, xerrors.Errorf("getting ipfs block: %w", err)
|
|
}
|
|
|
|
return true, nil
|
|
}
|
|
|
|
func (i *IpfsBstore) Get(cid cid.Cid) (blocks.Block, error) {
|
|
rd, err := i.api.Block().Get(i.ctx, path.IpldPath(cid))
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("getting ipfs block: %w", err)
|
|
}
|
|
|
|
data, err := ioutil.ReadAll(rd)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return blocks.NewBlockWithCid(data, cid)
|
|
}
|
|
|
|
func (i *IpfsBstore) GetSize(cid cid.Cid) (int, error) {
|
|
st, err := i.api.Block().Stat(i.ctx, path.IpldPath(cid))
|
|
if err != nil {
|
|
return 0, xerrors.Errorf("getting ipfs block: %w", err)
|
|
}
|
|
|
|
return st.Size(), nil
|
|
}
|
|
|
|
func (i *IpfsBstore) Put(block blocks.Block) error {
|
|
mhd, err := multihash.Decode(block.Cid().Hash())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
_, err = i.api.Block().Put(i.ctx, bytes.NewReader(block.RawData()),
|
|
options.Block.Hash(mhd.Code, mhd.Length),
|
|
options.Block.Format(cid.CodecToStr[block.Cid().Type()]))
|
|
return err
|
|
}
|
|
|
|
func (i *IpfsBstore) PutMany(blocks []blocks.Block) error {
|
|
// TODO: could be done in parallel
|
|
|
|
for _, block := range blocks {
|
|
if err := i.Put(block); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (i *IpfsBstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
|
return nil, xerrors.Errorf("not supported")
|
|
}
|
|
|
|
func (i *IpfsBstore) HashOnRead(enabled bool) {
|
|
return // TODO: We could technically support this, but..
|
|
}
|
|
|
|
var _ blockstore.Blockstore = &IpfsBstore{}
|