func (sp *FileStorageProvider) CreateStream(uuid []byte, collection string, tags map[string]string, annotation []byte) bte.BTE {
	panic("yo not supported bro")
}
CreateStream makes a stream with the given uuid, collection and tags. Returns
an error if the uuid already exists.
package fileprovider
import (
	"fmt"
	"io"
	"os"
	"sync"
	"github.com/BTrDB/btrdb-server/bte"
	"github.com/BTrDB/btrdb-server/internal/bprovider"
	"github.com/BTrDB/btrdb-server/internal/configprovider"
	"github.com/op/go-logging"
)
var log *logging.Logger
func init() {
	log = logging.MustGetLogger("log")
}
const NUMFILES = 256
type writeparams struct {
	Address uint64
	Data    []byte
}
type FileProviderSegment struct {
	sp    *FileStorageProvider
	fidx  int
	f     *os.File
	base  int64
	ptr   int64
	wchan chan writeparams
	wg    sync.WaitGroup
}
type FileStorageProvider struct {
	fidx     chan int
	retfidx  chan int
	dbf      []*os.File
	dbrf     []*os.File
	dbrf_mtx []sync.Mutex
	favail   []bool
}
func (seg *FileProviderSegment) writer() {
	for args := range seg.wchan {
		off := int64(args.Address & ((1 << 50) - 1))
		lenarr := make([]byte, 2)
		lenarr[0] = byte(len(args.Data))
		lenarr[1] = byte(len(args.Data) >> 8)
		_, err := seg.f.WriteAt(lenarr, off)
		if err != nil {
			log.Panic("File writing error %v", err)
		}
		_, err = seg.f.WriteAt(args.Data, off+2)
		if err != nil {
			log.Panic("File writing error %v", err)
		}
	}
	seg.wg.Done()
}
func (seg *FileProviderSegment) init() {
	seg.wchan = make(chan writeparams, 16)
	seg.wg.Add(1)
	go seg.writer()
}
func (seg *FileProviderSegment) BaseAddress() uint64 {
	return (uint64(seg.fidx) << 50) + uint64(seg.base)
}
func (seg *FileProviderSegment) Unlock() {
	seg.Flush()
	seg.sp.retfidx <- seg.fidx
}
func (seg *FileProviderSegment) Write(uuid []byte, address uint64, data []byte) (uint64, error) {
	if seg.ptr != int64(address&((1<<50)-1)) {
		log.Panic("Pointer does not match address %x vs %x", seg.ptr, int64(address&((1<<50)-1)))
	}
	wp := writeparams{Address: address, Data: data}
	seg.wchan <- wp
	seg.ptr = int64(address&((1<<50)-1)) + int64(len(data)) + 2
	return uint64(seg.ptr) + (uint64(seg.fidx) << 50), nil
}
func (seg *FileProviderSegment) Flush() {
	close(seg.wchan)
	seg.wg.Wait()
}
func (sp *FileStorageProvider) provideFiles() {
	for {
	ldretfi:
		for {
			select {
			case fi := <-sp.retfidx:
				sp.favail[fi] = true
			default:
				break ldretfi
			}
		}
		minidx := -1
		var minv int64 = 0
		for i := 0; i < NUMFILES; i++ {
			if !sp.favail[i] {
				continue
			}
			off, err := sp.dbf[i].Seek(0, os.SEEK_CUR)
			if err != nil {
				log.Panic(err)
			}
			if minidx == -1 || off < minv {
				minidx = i
				minv = off
			}
		}
		if minidx != -1 {
			sp.favail[minidx] = false
			sp.fidx <- minidx
		} else {
			fi := <-sp.retfidx
			sp.favail[fi] = true
		}
	}
}
func (sp *FileStorageProvider) Initialize(cfg configprovider.Configuration) {
	sp.fidx = make(chan int)
	sp.retfidx = make(chan int, NUMFILES+1)
	sp.dbf = make([]*os.File, NUMFILES)
	sp.dbrf = make([]*os.File, NUMFILES)
	sp.dbrf_mtx = make([]sync.Mutex, NUMFILES)
	sp.favail = make([]bool, NUMFILES)
	for i := 0; i < NUMFILES; i++ {
		dbpath := cfg.StorageFilepath()
		fname := fmt.Sprintf("%s/blockstore.%02x.db", dbpath, i)
		{
			f, err := os.OpenFile(fname, os.O_RDWR, 0666)
			if err != nil && os.IsNotExist(err) {
				log.Critical("Aborting: seems database does not exist. Have you run `btrdbd -makedb`?")
				os.Exit(1)
			}
			if err != nil {
				log.Panicf("Problem with blockstore DB: ", err)
			}
			sp.dbf[i] = f
		}
		{
			f, err := os.OpenFile(fname, os.O_RDONLY, 0666)
			if err != nil {
				log.Panicf("Problem with blockstore DB: ", err)
			}
			sp.dbrf[i] = f
		}
		sp.favail[i] = true
	}
	go sp.provideFiles()
}
func (sp *FileStorageProvider) LockSegment(uuid []byte) bprovider.Segment {
	fidx := <-sp.fidx
	f := sp.dbf[fidx]
	l, err := f.Seek(0, os.SEEK_END)
	if err != nil {
		log.Panicf("Error on lock segment: %v", err)
	}
	seg := &FileProviderSegment{sp: sp, fidx: fidx, f: sp.dbf[fidx], base: l, ptr: l}
	seg.init()
	return seg
}
const FIRSTREAD = 3459
func (sp *FileStorageProvider) Read(uuid []byte, address uint64, buffer []byte) []byte {
	fidx := address >> 50
	off := int64(address & ((1 << 50) - 1))
	if fidx > NUMFILES {
		log.Panic("Encoded file idx too large")
	}
	sp.dbrf_mtx[fidx].Lock()
	nread, err := sp.dbrf[fidx].ReadAt(buffer[:FIRSTREAD], off)
	if err != nil && err != io.EOF {
		log.Panic("Non EOF read error: %v", err)
	}
	if nread < 2 {
		log.Panic("Unexpected (very) short read")
	}
	bsize := int(buffer[0]) + (int(buffer[1]) << 8)
	if bsize > nread-2 {
		_, err := sp.dbrf[fidx].ReadAt(buffer[nread:bsize+2], off+int64(nread))
		if err != nil {
			log.Panic("Read error: %v", err)
		}
	}
	sp.dbrf_mtx[fidx].Unlock()
	return buffer[2 : bsize+2]
}
func (sp *FileStorageProvider) CreateDatabase(cfg configprovider.Configuration) error {
	for i := 0; i < NUMFILES; i++ {
		dbpath := cfg.StorageFilepath()
		fname := fmt.Sprintf("%s/blockstore.%02x.db", dbpath, i)
		{
			f, err := os.OpenFile(fname, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0666)
			if err != nil && !os.IsExist(err) {
				log.Panicf("Problem with blockstore DB: ", err)
			} else if os.IsExist(err) {
				return bprovider.ErrExists
			}
			_, err = f.Write([]byte("QUASARDB"))
			if err != nil {
				log.Panicf("Could not write to blockstore:", err)
			}
			err = f.Close()
			if err != nil {
				log.Panicf("Error on close %v", err)
			}
		}
	}
	return nil
}
func (sp *FileStorageProvider) ReadSuperBlock(uuid []byte, version uint64, buffer []byte) []byte {
	panic("yo not supported bro")
}
func (sp *FileStorageProvider) WriteSuperBlock(uuid []byte, version uint64, buffer []byte) {
	panic("yo not supported bro")
}
func (sp *FileStorageProvider) SetStreamVersion(uuid []byte, version uint64) {
	panic("yo not supported bro")
}
func (sp *FileStorageProvider) GetStreamInfo(uuid []byte) (bprovider.Stream, uint64) {
	panic("yo not supported bro")
}
func (sp *FileStorageProvider) GetStreamVersion(uuid []byte) uint64 {
	panic("yo not supported bro")
}