[WIP] Partial Transition from Writeable/Readable to Codecs (#51)

* Sample Signatures for put_enc and get_dec
* Implement put_enc and get_dec
* Implement ChainCodec in grin_chain
* Truncate src only on complete Blocks
* Truncate src only on complete Tip + Check Len
* Move BlockHeader Encoding to BlockHeaderCodec
* Define put_enc for store::Batch
* Replace BlockCodec and BlockHeaderCodec with generic BlockCodec<T>
* Implement Default for BlockCodec Manually
* Replace get_ser/put_ser with get_enc/get_dec for chain::ChainKVStore
* Remove Writeable/Readable for chain::Tip
* Add Tokio-io and Bytes to grin_p2p
* Additional Setup for Message enum + Msg{Encode,Decode} traits
* base msg ping pong encoding and test
* fill out msg-codec tests
* Implement Hand Encoding/Decoding
* msg-encode shake
* msg-encode getpeeraddr
* codec peer-addrs message, SockAddr struct wierdness
* header message codec
* msg encoding finished prelim
* Implement PeerCodec Encoding/Decoding
* Set PeerStore to use PeerCodec for Encoding/Decoding
* Add a DecIterator
* Prune PeerStore
* Replace Decoding and Encoding in handle_payload
* Prune Writeable/Readable methods in store::Store
* Remove Incomplete Frame Testing ( Not Nessesary right now )
* separate block and tx codec tests
* Refactor {Tx,Block}Codec Tests

chain/src/codec.rs

@@ -0,0 +1,184 @@
+// Copyright 2016 The Grin Developers
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Implementation of the chain block encoding and decoding.
+
+use std::io;
+
+use tokio_io::*;
+use bytes::{BytesMut, BigEndian, BufMut, Buf, IntoBuf};
+use num_bigint::BigUint;
+
+use types::Tip;
+use core::core::hash::Hash;
+use core::core::target::Difficulty;
+
+// Convenience Macro for Option Handling in Decoding
+macro_rules! try_opt_dec {
+	($e: expr) => (match $e {
+		Some(val) => val,
+		None => return Ok(None),
+	});
+}
+
+/// Codec for Decoding and Encoding a `Tip`
+#[derive(Debug, Clone, Default)]
+pub struct ChainCodec;
+
+impl codec::Encoder for ChainCodec {
+	type Item = Tip;
+	type Error = io::Error;
+	fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
+		// Put Height
+		dst.reserve(8);
+		dst.put_u64::<BigEndian>(item.height);
+
+		// Put Last Block Hash
+		item.last_block_h.chain_encode(dst)?;
+
+		// Put Previous Block Hash
+		item.prev_block_h.chain_encode(dst)?;
+
+		// Put Difficulty
+		item.total_difficulty.chain_encode(dst)?;
+
+		Ok(())
+	}
+}
+
+impl codec::Decoder for ChainCodec {
+	type Item = Tip;
+	type Error = io::Error;
+	fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+
+		// Create Temporary Buffer
+		let ref mut temp = src.clone();
+
+		// Get Height
+		if temp.len() < 8 {
+			return Ok(None);
+		}
+		let mut buf = temp.split_to(8).into_buf();
+		let height = buf.get_u64::<BigEndian>();
+
+		// Get Last Block Hash
+		let last_block_h = try_opt_dec!(Hash::chain_decode(temp)?);
+
+		// Get Previous Block Hash
+		let prev_block_h = try_opt_dec!(Hash::chain_decode(temp)?);
+
+		// Get Difficulty
+		let total_difficulty = try_opt_dec!(Difficulty::chain_decode(temp)?);
+
+		// If succesfull truncate src by bytes read from temp;
+		let diff = src.len() - temp.len();
+		src.split_to(diff);
+
+		Ok(Some(Tip {
+			height: height,
+			last_block_h: last_block_h,
+			prev_block_h: prev_block_h,
+			total_difficulty: total_difficulty,
+		}))
+	}
+}
+
+/// Internal Convenience Trait
+trait ChainEncode: Sized {
+	fn chain_encode(&self, dst: &mut BytesMut) -> Result<(), io::Error>;
+}
+
+/// Internal Convenience Trait
+trait ChainDecode: Sized {
+	fn chain_decode(src: &mut BytesMut) -> Result<Option<Self>, io::Error>;
+}
+
+impl ChainEncode for Difficulty {
+	fn chain_encode(&self, dst: &mut BytesMut) -> Result<(), io::Error> {
+		let data = self.clone().into_biguint().to_bytes_be();
+		dst.reserve(1 + data.len());
+		dst.put_u8(data.len() as u8);
+		dst.put_slice(&data);
+		Ok(())
+	}
+}
+
+impl ChainDecode for Difficulty {
+	fn chain_decode(src: &mut BytesMut) -> Result<Option<Self>, io::Error> {
+		if src.len() < 1 {
+			return Ok(None);
+		}
+		let mut buf = src.split_to(1).into_buf();
+		let dlen = buf.get_u8() as usize;
+
+		if src.len() < dlen {
+			return Ok(None);
+		}
+
+		let buf = src.split_to(dlen).into_buf();
+		let data = Buf::bytes(&buf);
+
+		Ok(Some(Difficulty::from_biguint(BigUint::from_bytes_be(data))))
+	}
+}
+
+impl ChainEncode for Hash {
+	fn chain_encode(&self, dst: &mut BytesMut) -> Result<(), io::Error> {
+		dst.reserve(32);
+		dst.put_slice(self.as_ref());
+		Ok(())
+	}
+}
+
+impl ChainDecode for Hash {
+	fn chain_decode(src: &mut BytesMut) -> Result<Option<Self>, io::Error> {
+		if src.len() < 32 {
+			return Ok(None);
+		}
+
+		let mut buf = src.split_to(32).into_buf();
+		let mut hash_data = [0; 32];
+		buf.copy_to_slice(&mut hash_data);
+
+		Ok(Some(Hash(hash_data)))
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn should_have_chain_codec_roundtrip() {
+		use tokio_io::codec::{Encoder, Decoder};
+
+		let sample_gdb = Hash([1u8; 32]);
+		let tip = Tip::new(sample_gdb);
+
+		let mut buf = BytesMut::with_capacity(0);
+		let mut codec = ChainCodec {};
+		codec.encode(tip.clone(), &mut buf).expect("Error During Tip Encoding");
+
+		let d_tip =
+			codec.decode(&mut buf).expect("Error During Tip Decoding").expect("Unfinished Tip");
+
+		// Check if all bytes are read
+		assert_eq!(buf.len(), 0);
+
+		assert_eq!(tip.height, d_tip.height);
+		assert_eq!(tip.last_block_h, d_tip.last_block_h);
+		assert_eq!(tip.prev_block_h, d_tip.prev_block_h);
+		assert_eq!(tip.total_difficulty, d_tip.total_difficulty);
+	}
+}

chain/src/lib.rs

@@ -29,6 +29,9 @@ extern crate serde;
 #[macro_use]
 extern crate serde_derive;
 extern crate time;
+extern crate tokio_io;
+extern crate bytes;
+extern crate num_bigint;
 
 extern crate grin_core as core;
 extern crate grin_store;
@@ -37,6 +40,7 @@ extern crate secp256k1zkp as secp;
 pub mod pipe;
 pub mod store;
 pub mod types;
+pub mod codec;
 
 // Re-export the base interface
 

chain/src/store.rs

@@ -17,9 +17,11 @@
 use secp::pedersen::Commitment;
 
 use types::*;
+use codec::ChainCodec;
 use core::core::hash::{Hash, Hashed};
 use core::core::{Block, BlockHeader, Output};
 use grin_store::{self, Error, to_key, u64_to_key, option_to_not_found};
+use grin_store::codec::BlockCodec;
 
 const STORE_SUBPATH: &'static str = "chain";
 
@@ -40,46 +42,49 @@ pub struct ChainKVStore {
 impl ChainKVStore {
 	pub fn new(root_path: String) -> Result<ChainKVStore, Error> {
 		let db = grin_store::Store::open(format!("{}/{}", root_path, STORE_SUBPATH).as_str())?;
+		let codec = ChainCodec::default();
 		Ok(ChainKVStore { db: db })
 	}
 }
 
 impl ChainStore for ChainKVStore {
 	fn head(&self) -> Result<Tip, Error> {
-		option_to_not_found(self.db.get_ser(&vec![HEAD_PREFIX]))
+		option_to_not_found(self.db.get_dec(&mut ChainCodec, &[HEAD_PREFIX]))
 	}
 
 	fn head_header(&self) -> Result<BlockHeader, Error> {
-		let head: Tip = try!(option_to_not_found(self.db.get_ser(&vec![HEAD_PREFIX])));
+		let head: Tip = option_to_not_found(self.db.get_dec(&mut ChainCodec, &[HEAD_PREFIX]))?;
 		self.get_block_header(&head.last_block_h)
 	}
 
 	fn save_head(&self, t: &Tip) -> Result<(), Error> {
 		self.db
 			.batch()
-			.put_ser(&vec![HEAD_PREFIX], t)?
-			.put_ser(&vec![HEADER_HEAD_PREFIX], t)?
+			.put_enc(&mut ChainCodec, &[HEAD_PREFIX], t.clone())?
+			.put_enc(&mut ChainCodec, &[HEADER_HEAD_PREFIX], t.clone())?
 			.write()
 	}
 
 	fn save_body_head(&self, t: &Tip) -> Result<(), Error> {
-		self.db.put_ser(&vec![HEAD_PREFIX], t)
+		self.db.put_enc(&mut ChainCodec, &[HEAD_PREFIX], t.clone())
 	}
 
 	fn get_header_head(&self) -> Result<Tip, Error> {
-		option_to_not_found(self.db.get_ser(&vec![HEADER_HEAD_PREFIX]))
+		option_to_not_found(self.db.get_dec(&mut ChainCodec, &[HEADER_HEAD_PREFIX]))
 	}
 
 	fn save_header_head(&self, t: &Tip) -> Result<(), Error> {
-		self.db.put_ser(&vec![HEADER_HEAD_PREFIX], t)
+		self.db.put_enc(&mut ChainCodec, &[HEADER_HEAD_PREFIX], t.clone())
 	}
 
 	fn get_block(&self, h: &Hash) -> Result<Block, Error> {
-		option_to_not_found(self.db.get_ser(&to_key(BLOCK_PREFIX, &mut h.to_vec())))
+		option_to_not_found(self.db.get_dec(&mut BlockCodec::default(),
+		                                    &to_key(BLOCK_PREFIX, &mut h.to_vec())))
 	}
 
 	fn get_block_header(&self, h: &Hash) -> Result<BlockHeader, Error> {
-		option_to_not_found(self.db.get_ser(&to_key(BLOCK_HEADER_PREFIX, &mut h.to_vec())))
+		option_to_not_found(self.db.get_dec(&mut BlockCodec::default(),
+		                                    &to_key(BLOCK_HEADER_PREFIX, &mut h.to_vec())))
 	}
 
 	fn check_block_exists(&self, h: &Hash) -> Result<bool, Error> {
@@ -90,39 +95,52 @@ impl ChainStore for ChainKVStore {
 		// saving the block and its header
 		let mut batch = self.db
 			.batch()
-			.put_ser(&to_key(BLOCK_PREFIX, &mut b.hash().to_vec())[..], b)?
-			.put_ser(&to_key(BLOCK_HEADER_PREFIX, &mut b.hash().to_vec())[..],
-			         &b.header)?;
+			.put_enc(&mut BlockCodec::default(),
+			         &to_key(BLOCK_PREFIX, &mut b.hash().to_vec())[..],
+			         b.clone())?
+			.put_enc(&mut BlockCodec::default(),
+			         &to_key(BLOCK_HEADER_PREFIX, &mut b.hash().to_vec())[..],
+			         b.header.clone())?;
 
 		// saving the full output under its hash, as well as a commitment to hash index
 		for out in &b.outputs {
-			batch = batch.put_ser(&to_key(OUTPUT_PREFIX, &mut out.hash().to_vec())[..], out)?
-				.put_ser(&to_key(OUTPUT_COMMIT_PREFIX, &mut out.commit.as_ref().to_vec())[..],
-				         &out.hash())?;
+			batch = batch.put_enc(&mut BlockCodec::default(),
+				         &to_key(OUTPUT_PREFIX, &mut out.hash().to_vec())[..],
+				         out.clone())?
+				.put_enc(&mut BlockCodec::default(),
+				         &to_key(OUTPUT_COMMIT_PREFIX, &mut out.commit.as_ref().to_vec())[..],
+				         out.hash().clone())?;
 		}
 		batch.write()
 	}
 
 	fn save_block_header(&self, bh: &BlockHeader) -> Result<(), Error> {
-		self.db.put_ser(&to_key(BLOCK_HEADER_PREFIX, &mut bh.hash().to_vec())[..],
-		                bh)
+		self.db.put_enc(&mut BlockCodec::default(),
+		                &to_key(BLOCK_HEADER_PREFIX, &mut bh.hash().to_vec())[..],
+		                bh.clone())
 	}
 
 	fn get_header_by_height(&self, height: u64) -> Result<BlockHeader, Error> {
-		option_to_not_found(self.db.get_ser(&u64_to_key(HEADER_HEIGHT_PREFIX, height)))
+		option_to_not_found(self.db.get_dec(&mut BlockCodec::default(),
+		                                    &u64_to_key(HEADER_HEIGHT_PREFIX, height)))
 	}
 
 	fn get_output(&self, h: &Hash) -> Result<Output, Error> {
-		option_to_not_found(self.db.get_ser(&to_key(OUTPUT_PREFIX, &mut h.to_vec())))
+		option_to_not_found(self.db.get_dec(&mut BlockCodec::default(),
+		                                    &to_key(OUTPUT_PREFIX, &mut h.to_vec())))
 	}
 
 	fn has_output_commit(&self, commit: &Commitment) -> Result<Hash, Error> {
 		option_to_not_found(self.db
-			.get_ser(&to_key(OUTPUT_COMMIT_PREFIX, &mut commit.as_ref().to_vec())))
+			.get_dec(&mut BlockCodec::default(),
+			         &to_key(OUTPUT_COMMIT_PREFIX, &mut commit.as_ref().to_vec())))
 	}
 
 	fn setup_height(&self, bh: &BlockHeader) -> Result<(), Error> {
-		self.db.put_ser(&u64_to_key(HEADER_HEIGHT_PREFIX, bh.height), bh)?;
+		self.db
+			.put_enc(&mut BlockCodec::default(),
+			         &u64_to_key(HEADER_HEIGHT_PREFIX, bh.height),
+			         bh.clone())?;
 
 		let mut prev_h = bh.previous;
 		let mut prev_height = bh.height - 1;
@@ -131,8 +149,9 @@ impl ChainStore for ChainKVStore {
 			if prev.hash() != prev_h {
 				let real_prev = self.get_block_header(&prev_h)?;
 				self.db
-					.put_ser(&u64_to_key(HEADER_HEIGHT_PREFIX, real_prev.height),
-					         &real_prev);
+					.put_enc(&mut BlockCodec::default(),
+					         &u64_to_key(HEADER_HEIGHT_PREFIX, real_prev.height),
+					         real_prev.clone())?;
 				prev_h = real_prev.previous;
 				prev_height = real_prev.height - 1;
 			} else {

chain/src/types.rs

@@ -20,7 +20,6 @@ use grin_store::Error;
 use core::core::{Block, BlockHeader, Output};
 use core::core::hash::{Hash, Hashed};
 use core::core::target::Difficulty;
-use core::ser;
 
 /// The tip of a fork. A handle to the fork ancestry from its leaf in the
 /// blockchain tree. References the max height and the latest and previous
@@ -60,31 +59,6 @@ impl Tip {
 	}
 }
 
-/// Serialization of a tip, required to save to datastore.
-impl ser::Writeable for Tip {
-	fn write<W: ser::Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
-		try!(writer.write_u64(self.height));
-		try!(writer.write_fixed_bytes(&self.last_block_h));
-		try!(writer.write_fixed_bytes(&self.prev_block_h));
-		self.total_difficulty.write(writer)
-	}
-}
-
-impl ser::Readable for Tip {
-	fn read(reader: &mut ser::Reader) -> Result<Tip, ser::Error> {
-		let height = try!(reader.read_u64());
-		let last = try!(Hash::read(reader));
-		let prev = try!(Hash::read(reader));
-		let diff = try!(Difficulty::read(reader));
-		Ok(Tip {
-			height: height,
-			last_block_h: last,
-			prev_block_h: prev,
-			total_difficulty: diff,
-		})
-	}
-}
-
 /// Trait the chain pipeline requires an implementor for in order to process
 /// blocks.
 pub trait ChainStore: Send + Sync {