// Copyright 2014 by Sascha L. Teichmann
// Use of this source code is governed by the MIT license
// that can be found in the LICENSE file.

package main

import (
	"bitbucket.org/s_l_teichmann/mtredisalize/common"

	leveldb "github.com/jmhodges/levigo"
)

type LevelDBBlockConsumer struct {
	db      *leveldb.DB
	wo      *leveldb.WriteOptions
	joiner  common.KeyJoiner
	encoder common.KeyEncoder
}

func NewLevelDBBlockConsumer(
	path string,
	joiner common.KeyJoiner,
	encoder common.KeyEncoder) (ldbc *LevelDBBlockConsumer, err error) {

	opts := leveldb.NewOptions()
	opts.SetCreateIfMissing(true)

	var db *leveldb.DB
	if db, err = leveldb.Open(path, opts); err != nil {
		return
	}

	ldbc = &LevelDBBlockConsumer{
		db:      db,
		wo:      leveldb.NewWriteOptions(),
		joiner:  joiner,
		encoder: encoder}
	return
}

func (ldbc *LevelDBBlockConsumer) Close() error {
	ldbc.wo.Close()
	ldbc.db.Close()
	return nil
}

func (ldbc *LevelDBBlockConsumer) Consume(block Block) (err error) {
	var encodedKey []byte
	if encodedKey, err = ldbc.encoder(ldbc.joiner(block.Coord)); err != nil {
		return
	}
	err = ldbc.db.Put(ldbc.wo, encodedKey, block.Data)
	return
}