From 02ada3809061705df13a448f18b2a9d565b103c6 Mon Sep 17 00:00:00 2001
From: "Sascha L. Teichmann" <sascha.teichmann@intevation.de>
Date: Wed, 20 Aug 2014 15:26:31 +0200
Subject: [PATCH] Added block consumer for LevelDB in interleaver.

---
 interleaver/interfaces.go |  2 +-
 interleaver/leveldb.go    | 54 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 55 insertions(+), 1 deletion(-)
 create mode 100644 interleaver/leveldb.go

diff --git a/interleaver/interfaces.go b/interleaver/interfaces.go
index d70845d..0535e38 100644
--- a/interleaver/interfaces.go
+++ b/interleaver/interfaces.go
@@ -26,7 +26,7 @@ type (
 	}
 
 	BlockConsumer interface {
-		Consume(Block)
+		Consume(Block) error
 		Close() error
 	}
 )
diff --git a/interleaver/leveldb.go b/interleaver/leveldb.go
new file mode 100644
index 0000000..fae0ac0
--- /dev/null
+++ b/interleaver/leveldb.go
@@ -0,0 +1,54 @@
+// 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
+}