| // Copyright 2015 The Vanadium Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style |
| // license that can be found in the LICENSE file. |
| |
| // This file implements the BlobMetadata table of a blobmap. |
| |
| package blobmap |
| |
| import "v.io/v23/context" |
| import "v.io/v23/services/syncbase/nosql" |
| import "v.io/v23/vom" |
| import "v.io/x/ref/services/syncbase/localblobstore" |
| import "v.io/x/ref/services/syncbase/store" |
| |
| // metadataKey() returns the key used in the store to reference the metadata |
| // for blobId. |
| func metadataKey(blobId nosql.BlobRef) []byte { |
| return []byte(metadataPrefix + string(blobId)) |
| } |
| |
| // SetBlobMetadata() sets the BlobMetadata associated with a blob to *bmd. |
| func (bm *BlobMap) SetBlobMetadata(ctx *context.T, blobId nosql.BlobRef, bmd *localblobstore.BlobMetadata) (err error) { |
| var val []byte |
| |
| val, err = vom.Encode(bmd) |
| if err == nil { |
| err = bm.st.Put(metadataKey(blobId), val) |
| } |
| return err |
| } |
| |
| // GetBlobMetadata() yields in *bmd the BlobMetadata associated with a blob. |
| func (bm *BlobMap) GetBlobMetadata(ctx *context.T, blobId nosql.BlobRef, bmd *localblobstore.BlobMetadata) (err error) { |
| var valBuf [maxBlobMetadataLen]byte |
| var val []byte |
| |
| val, err = bm.st.Get(metadataKey(blobId), valBuf[:]) |
| if err == nil { |
| err = vom.Decode(val, bmd) |
| } |
| return err |
| } |
| |
| // DeleteBlobMetadata() deletes the BlobMetadata for the specified blob. |
| func (bm *BlobMap) DeleteBlobMetadata(ctx *context.T, blobId nosql.BlobRef) error { |
| return bm.st.Delete(metadataKey(blobId)) |
| } |
| |
| // A BlobMetadataStream allows the client to iterate over the BlobMetadatas associated with blob IDs. |
| // bms := bm.NewBlobMetadataStream(ctx, blob) |
| // for bms.Advance() { |
| // blob := bms.BlobId() |
| // metadata := bms.BlobMetadata() |
| // ...process blob, metadata... |
| // } |
| // if bms.Err() != nil { |
| // ...there was an error... |
| // } |
| type BlobMetadataStream struct { |
| bm *BlobMap |
| stream store.Stream |
| |
| keyBuf [maxKeyLen]byte // buffer for keys |
| valBuf [maxBlobMetadataLen]byte // buffer for values |
| blobId nosql.BlobRef // blobId key for current element |
| metadata localblobstore.BlobMetadata // BlobMetadata of current element |
| err error // error encountered. |
| more bool // whether stream may be consulted again |
| } |
| |
| // NewBlobMetadataStream() returns, as a localblobstore.BlobMetadataStream |
| // interface, a pointer to a new blobmap.BlobMetadataStream that allows the |
| // client to enumerate the BlobMetadatas associated with blob IDs, in order. |
| func (bm *BlobMap) NewBlobMetadataStream(ctx *context.T) localblobstore.BlobMetadataStream { |
| // ctx is currently unused. |
| bms := new(BlobMetadataStream) |
| bms.bm = bm |
| bms.stream = bm.st.Scan([]byte(metadataPrefix), metadataStreamKeyLimit) |
| bms.more = true |
| return bms |
| } |
| |
| // Advance() stages an element so the client can retrieve the blob ID hash with |
| // BlobId(), or its BlobMetadata with BlobMetadata(). Advance() returns true iff there |
| // is an element to retrieve. The client must call Advance() before calling |
| // BlobId() or BlobMetadata() The client must call Cancel if it does not iterate |
| // through all elements (i.e. until Advance() returns false). Advance() may |
| // block if an element is not immediately available. |
| func (bms *BlobMetadataStream) Advance() (ok bool) { |
| if bms.more && bms.err == nil { |
| if !bms.stream.Advance() { |
| bms.err = bms.stream.Err() |
| bms.more = false // no more stream, even if no error |
| } else { |
| bms.blobId = nosql.BlobRef(bms.stream.Key(bms.keyBuf[:])[1:]) |
| var value []byte = bms.stream.Value(bms.valBuf[:]) |
| bms.err = vom.Decode(value, &bms.metadata) |
| ok = (bms.err == nil) |
| if !ok { |
| bms.stream.Cancel() |
| } |
| } |
| } |
| return ok |
| } |
| |
| // BlobId() returns the blob ID of the blob Id and BlobMetadata staged by |
| // Advance(). BlobId() may panic if Advance() returned false or was not called |
| // at all. BlobId() does not block. |
| func (bms *BlobMetadataStream) BlobId() (result nosql.BlobRef) { |
| return bms.blobId |
| } |
| |
| // BlobMetadata() returns the BlobMetadata associated with the blob ID staged by |
| // Advance(). BlobMetadata() may panic if Advance() returned false or was not |
| // called at all. BlobMetadata() does not block. |
| func (bms *BlobMetadataStream) BlobMetadata() localblobstore.BlobMetadata { |
| return bms.metadata |
| } |
| |
| // Err() returns a non-nil error iff the stream encountered any errors. Err() |
| // does not block. |
| func (bms *BlobMetadataStream) Err() error { |
| return bms.err |
| } |
| |
| // Cancel() notifies the stream provider that it can stop producing elements. |
| // The client must call Cancel() if it does not iterate through all elements |
| // (i.e. until Advance() returns false). Cancel() is idempotent and can be |
| // called concurrently with a goroutine that is iterating via Advance() and |
| // BlobId(). Cancel() causes Advance() to subsequently return false. |
| // Cancel() does not block. |
| func (bms *BlobMetadataStream) Cancel() { |
| bms.stream.Cancel() |
| } |