Blob

blob.go

package storage

import (
	"encoding/json"
	"io"
	"os"
	"sync"
	"sync/atomic"

	"github.com/boltdb/bolt"
)

type Object struct {
	Offset int64
	Length int64
}

type Blob struct {
	m      sync.Mutex
	header *bolt.DB
	data   *os.File
	offset int64
}

func NewBlob(name string) (*Blob, error) {
	// create header db
	header, err := bolt.Open(name+".index", 0600, nil)
	if err != nil {
		return nil, err
	}
	err = header.Update(func(tx *bolt.Tx) error {
		_, err := tx.CreateBucketIfNotExists([]byte("index"))
		return err
	})
	if err != nil {
		return nil, err
	}
	// create data file
	data, err := os.Create(name)
	if err != nil {
		return nil, err
	}
	dataInfo, _ := data.Stat()
	return &Blob{data: data, header: header, offset: dataInfo.Size()}, nil
}

func (b *Blob) Close() error {
	err := b.data.Sync()
	if err != nil {
		return err
	}
	err = b.data.Close()
	if err != nil {
		return err
	}
	err = b.header.Sync()
	if err != nil {
		return err
	}
	return b.header.Close()
}

func (b *Blob) Append(key string, size int64, data io.Reader) error {
	// allocate space in blob
	offset := atomic.LoadInt64(&b.offset)
	atomic.AddInt64(&b.offset, size)
	index := Object{offset, size}
	// write data to blob
	buf := make([]byte, 2048)
	for {
		n, err := data.Read(buf)
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		_, err = b.data.WriteAt(buf[:n], offset)
		if err != nil {
			return err
		}
		offset += int64(n)
	}
	b.data.Sync()
	// add to index
	bIndex, err := json.Marshal(&index)
	if err != nil {
		return err
	}
	err = b.header.Update(func(tx *bolt.Tx) error {
		bt := tx.Bucket([]byte("index"))
		return bt.Put([]byte(key), bIndex)
	})
	return err
}