public static final SimpleTreeIdProvider TREE_ID_PROVIDER = new SimpleTreeIdProvider(); public class HTReentrantLockProvider implements LockProvider {

	private final ConcurrentHashMap<Long, ReentrantLock> internalLocks = new ConcurrentHashMap<>();

	private ReentrantLock getLock(long treeId) {
		if (!internalLocks.contains(treeId)) {
			internalLocks.putIfAbsent(treeId, new ReentrantLock());
		}
		return internalLocks.get(treeId);
	}

	@Override
	public boolean acquireLock(long treeId) {
		ReentrantLock lock = getLock(treeId);
		return lock.tryLock();
	}

	@Override
	public void releaseLock(long treeId) {
		ReentrantLock lock = getLock(treeId);
		lock.unlock();
	}

} public interface HashTrees {

	/**
	 * Adds the (key,value) pair to the store. Intended to be used while synch
	 * operation.
	 * 
	 * @param keyValuePairs
	 */
	void sPut(List<KeyValue> keyValuePairs) throws IOException;

	/**
	 * Deletes the keys from the store. Intended to be used while synch
	 * operation.
	 * 
	 * @param keys
	 * 
	 */
	void sRemove(List<ByteBuffer> keys) throws IOException;

	/**
	 * Hash tree internal nodes store the hash of their children nodes. Given a
	 * set of internal node ids, this returns the hashes that are stored on the
	 * internal node.
	 * 
	 * @param treeId
	 * @param nodeIds
	 *            , internal tree node ids.
	 * @return
	 * 
	 */
	List<SegmentHash> getSegmentHashes(long treeId, List<Integer> nodeIds)
			throws IOException;

	/**
	 * Returns the segment hash that is stored on the tree.
	 * 
	 * @param treeId
	 *            , hash tree id.
	 * @param nodeId
	 *            , node id
	 * @return
	 * 
	 */
	SegmentHash getSegmentHash(long treeId, int nodeId) throws IOException;

	/**
	 * Hash tree data is stored on the leaf blocks. Given a segment id this
	 * method is supposed to return (key,hash) pairs.
	 * 
	 * @param treeId
	 * @param segId
	 *            , id of the segment block.
	 * @return
	 * 
	 */
	List<SegmentData> getSegment(long treeId, int segId) throws IOException;

	/**
	 * Returns the (key,digest) for the given key in the given segment.
	 * 
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 */
	SegmentData getSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException;

	/**
	 * Deletes the given tree node.
	 * 
	 * @param treeId
	 * @param nodeId
	 * @throws IOException
	 */
	void deleteTreeNode(long treeId, int nodeId) throws IOException;

	/**
	 * Adds the key, and digest of value to the segment block in HashTree.
	 * 
	 * @param key
	 * @param value
	 * @throws IOException
	 */
	void hPut(ByteBuffer key, ByteBuffer value) throws IOException;

	/**
	 * Deletes the key from the hash tree.
	 * 
	 * @param key
	 * @throws IOException
	 */
	void hRemove(ByteBuffer key) throws IOException;

	/**
	 * Updates the other HTree based on the differences with local objects. This
	 * function should be running on primary to synch with other replicas, and
	 * not the other way.
	 * 
	 * @param remoteTree
	 * @param syncType
	 * @return, gives stats about total number of differences between local node
	 *          and remote node.
	 */
	SyncDiffResult synch(long treeId, HashTrees remoteTree, SyncType syncType)
			throws IOException;

	/**
	 * Same as {@link #synch(long, HashTrees, SyncType)} except that, syncType
	 * is passed as SyncType.UPDATE.
	 * 
	 * @param treeId
	 * @param remoteTree
	 * @return
	 */
	SyncDiffResult synch(long treeId, HashTrees remoteTree) throws IOException;

	/**
	 * Updates segment hashes based on the dirty entries.
	 * 
	 * @param treeId
	 * @param fullRebuildPeriod
	 *            , indicates a time interval in milliseconds, and if a tree is
	 *            not fully rebuilt for more than this value, a full rebuild
	 *            will be triggered. Otherwise only dirty segments will be
	 *            updated. If there is no information about when was the last
	 *            time the tree was fully rebuilt, then a full rebuild will be
	 *            triggered. A negative value indicates disabling full rebuild.
	 * @return, returns how many dirty segments are marked before the rebuild.
	 */
	int rebuildHashTree(long treeId, long fullRebuildPeriod) throws IOException;

	/**
	 * Similar to {@link #rebuildHashTree(long, long)}, except that allows to
	 * specify whether a full rebuild should be triggered or not.
	 * 
	 * @param treeId
	 * @param fullRebuild
	 * @return, returns how many dirty segments are marked before the rebuild.
	 * @throws IOException
	 */
	int rebuildHashTree(long treeId, boolean fullRebuild) throws IOException;

	void addObserver(HashTreesObserver observer) throws IOException;

	void removeObserver(HashTreesObserver observer) throws IOException;
} @ThreadSafe
public class HashTreesImpl implements HashTrees, Service {

	/**
	 * Specifies how much element can be queued when hash tree is backed by a
	 * non blocking queue. When the queue is full, the new puts or removes are
	 * rejected.
	 */
	public final static int DEFAULT_NB_QUE_SIZE = Integer.MAX_VALUE;
	/**
	 * Specifies at most how many segments can be used by a single HashTree.
	 */
	public final static int MAX_NO_OF_SEGMENTS = 1 << 30;

	private final static Logger LOGGER = LoggerFactory
			.getLogger(HashTreesImpl.class.getName());
	private final static char COMMA_DELIMETER = ',';
	private final static char NEW_LINE_DELIMETER = '\n';
	private final static int ROOT_NODE = 0;
	private final static int BINARY_TREE = 2;

	private final int noOfChildren;
	private final int internalNodesCount;
	private final int segmentsCount;
	private final int height;

	private final Store store;
	private final HashTreesStore htStore;
	private final HashTreesIdProvider treeIdProvider;
	private final SegmentIdProvider segIdProvider;

	private final boolean enabledNonBlockingCalls;
	private final int nonBlockingQueueSize;
	private final Object nonBlockingCallsLock = new Object();
	@LockedBy("nonBlockingCallsLock")
	private volatile NonBlockingHTDataUpdater bgDataUpdater;
	private final HashTreesObserverNotifier notifier = new HashTreesObserverNotifier();
	private final LockProvider lockProvider;

	public HashTreesImpl(int noOfSegments, boolean enabledNonBlockingCalls,
			int nonBlockingQueueSize, final HashTreesIdProvider treeIdProvider,
			final SegmentIdProvider segIdProvider,
			final HashTreesStore htStore, final Store store,
			final LockProvider lockProvider) {
		this.noOfChildren = BINARY_TREE;
		this.segmentsCount = getValidSegmentsCount(noOfSegments);
		this.enabledNonBlockingCalls = enabledNonBlockingCalls;
		this.nonBlockingQueueSize = nonBlockingQueueSize;
		this.height = height(this.segmentsCount, noOfChildren);
		this.internalNodesCount = getNoOfNodes((height - 1), noOfChildren);
		this.treeIdProvider = treeIdProvider;
		this.segIdProvider = segIdProvider;
		this.htStore = htStore;
		this.store = store;
		this.lockProvider = lockProvider;
	}

	@Override
	public void hPut(final ByteBuffer key, final ByteBuffer value)
			throws IOException {
		hPutInternal(HTOperation.PUT, key, value);
	}

	private void hPutInternal(HTOperation operation, final ByteBuffer key,
			final ByteBuffer value) throws IOException {
		if (enabledNonBlockingCalls) {
			List<ByteBuffer> input = new ArrayList<ByteBuffer>();
			input.add(key);
			input.add(value);
			bgDataUpdater.enque(Pair.create(operation, input));
		} else
			hPutInternal(key, value);
	}

	private void hPutInternal(final ByteBuffer key, final ByteBuffer value)
			throws IOException {
		notifier.preHPut(key, value);
		long treeId = treeIdProvider.getTreeId(key.array());
		int segId = segIdProvider.getSegmentId(key.array());
		ByteBuffer digest = ByteBuffer.wrap(sha1(value.array()));
		htStore.setDirtySegment(treeId, segId);
		htStore.putSegmentData(treeId, segId, key, digest);
		notifier.postHPut(key, value);
	}

	@Override
	public void hRemove(final ByteBuffer key) throws IOException {
		hRemoveInternal(HTOperation.REMOVE, key);
	}

	private void hRemoveInternal(HTOperation operation, final ByteBuffer key)
			throws IOException {
		if (enabledNonBlockingCalls) {
			List<ByteBuffer> input = new ArrayList<ByteBuffer>();
			input.add(key);
			bgDataUpdater.enque(Pair.create(operation, input));
		} else {
			hRemoveInternal(key);
		}
	}

	private void hRemoveInternal(final ByteBuffer key) throws IOException {
		notifier.preHRemove(key);
		long treeId = treeIdProvider.getTreeId(key.array());
		int segId = segIdProvider.getSegmentId(key.array());
		htStore.setDirtySegment(treeId, segId);
		htStore.deleteSegmentData(treeId, segId, key);
		notifier.postHRemove(key);
	}

	@Override
	public SyncDiffResult synch(long treeId, final HashTrees remoteTree)
			throws IOException {
		return synch(treeId, remoteTree, SyncType.UPDATE);
	}

	@Override
	public SyncDiffResult synch(long treeId, final HashTrees remoteTree,
			SyncType syncType) throws IOException {
		if (lockProvider.acquireLock(treeId)) {
			try {
				boolean doUpdate = (syncType == SyncType.UPDATE) ? true : false;

				PeekingIterator<SegmentHash> localItr = null, remoteItr = null;
				SegmentHash local, remote;

				List<Integer> pQueue = new ArrayList<Integer>();
				pQueue.add(ROOT_NODE);

				int totKeyDifferences = 0, totExtrinsicSegments = 0;

				while (!pQueue.isEmpty()) {

					localItr = Iterators.peekingIterator(getSegmentHashes(
							treeId, pQueue).iterator());
					remoteItr = Iterators.peekingIterator(remoteTree
							.getSegmentHashes(treeId, pQueue).iterator());
					pQueue = new ArrayList<Integer>();

					while (localItr.hasNext() || remoteItr.hasNext()) {
						local = localItr.hasNext() ? localItr.peek() : null;
						remote = remoteItr.hasNext() ? remoteItr.peek() : null;

						int compareRes = compareSegNodeIds(local, remote);

						if (compareRes == 0) {
							if (!Arrays.equals(local.getHash(),
									remote.getHash())) {
								if (isLeafNode(local.getNodeId())) {
									totKeyDifferences += syncSegment(treeId,
											getSegmentIdFromLeafId(local
													.getNodeId()), remoteTree,
											doUpdate);
								} else
									pQueue.addAll(getImmediateChildren(
											local.getNodeId(), noOfChildren));

							}
							localItr.next();
							remoteItr.next();
						} else if (compareRes < 0) {
							totKeyDifferences += updateRemoteTreeWithMissingSegment(
									treeId, local.getNodeId(), remoteTree,
									doUpdate);
							localItr.next();
						} else {
							if (doUpdate)
								remoteTree.deleteTreeNode(treeId,
										remote.getNodeId());
							remoteItr.next();
							totExtrinsicSegments += 1;
						}
					}
				}
				return new SyncDiffResult(totKeyDifferences,
						totExtrinsicSegments);
			} finally {
				lockProvider.releaseLock(treeId);
			}
		}
		return new SyncDiffResult(0, 0);
	}

	private int syncSegment(long treeId, int segId, HashTrees remoteTree,
			boolean doUpdate) throws IOException {
		PeekingIterator<SegmentData> localDataItr = Iterators
				.peekingIterator(getSegment(treeId, segId).iterator());
		PeekingIterator<SegmentData> remoteDataItr = Iterators
				.peekingIterator(remoteTree.getSegment(treeId, segId)
						.iterator());

		List<KeyValue> kvsForAddition = new ArrayList<KeyValue>();
		List<ByteBuffer> keysForRemoval = new ArrayList<ByteBuffer>();

		SegmentData local, remote;
		while (localDataItr.hasNext() || remoteDataItr.hasNext()) {
			local = localDataItr.hasNext() ? localDataItr.peek() : null;
			remote = remoteDataItr.hasNext() ? remoteDataItr.peek() : null;

			int compRes = compareSegmentKeys(local, remote);
			if (compRes == 0) {
				if (!Arrays.equals(local.getDigest(), remote.getDigest())) {
					ByteBuffer key = ByteBuffer.wrap(local.getKey());
					byte[] value = store.get(local.getKey());
					if (value != null)
						kvsForAddition.add(new KeyValue(key, ByteBuffer
								.wrap(value)));
				}
				localDataItr.next();
				remoteDataItr.next();
			} else if (compRes < 0) {
				ByteBuffer key = ByteBuffer.wrap(local.getKey());
				byte[] value = store.get(local.getKey());
				if (value != null)
					kvsForAddition
							.add(new KeyValue(key, ByteBuffer.wrap(value)));
				localDataItr.next();
			} else {
				keysForRemoval.add(ByteBuffer.wrap(remote.getKey()));
				remoteDataItr.next();
			}
		}

		if (doUpdate) {
			if (kvsForAddition.size() > 0)
				remoteTree.sPut(kvsForAddition);
			if (keysForRemoval.size() > 0)
				remoteTree.sRemove(keysForRemoval);
		}

		return kvsForAddition.size() + keysForRemoval.size();
	}

	private int updateRemoteTreeWithMissingSegment(long treeId,
			int rootMissingNodeId, HashTrees remoteTree, boolean doUpdate)
			throws IOException {
		final List<KeyValue> keyValuePairs = new ArrayList<>();
		int maxSizeToTransfer = 5000;
		int totKeyUpdates = 0;
		int leftMostNodeId = getSegmentIdFromLeafId(getLeftMostChildNode(
				rootMissingNodeId, noOfChildren, height));
		int rightMostNodeId = getSegmentIdFromLeafId(getRightMostChildNode(
				rootMissingNodeId, noOfChildren, height));
		Iterator<SegmentData> sdItr = htStore.getSegmentDataIterator(treeId,
				leftMostNodeId, rightMostNodeId);
		while (sdItr.hasNext()) {
			SegmentData sd = sdItr.next();
			byte[] value = store.get(sd.getKey());
			if (value != null)
				keyValuePairs.add(new KeyValue(ByteBuffer.wrap(sd.getKey()),
						ByteBuffer.wrap(value)));
			if (keyValuePairs.size() > maxSizeToTransfer) {
				if (doUpdate)
					remoteTree.sPut(keyValuePairs);
				totKeyUpdates += keyValuePairs.size();
				keyValuePairs.clear();
			}
		}
		if (keyValuePairs.size() > 0) {
			if (doUpdate)
				remoteTree.sPut(keyValuePairs);
			totKeyUpdates += keyValuePairs.size();
			keyValuePairs.clear();
		}
		return totKeyUpdates;
	}

	@Override
	public SegmentHash getSegmentHash(long treeId, int nodeId)
			throws IOException {
		return htStore.getSegmentHash(treeId, nodeId);
	}

	@Override
	public List<SegmentHash> getSegmentHashes(long treeId,
			final List<Integer> nodeIds) throws IOException {
		return htStore.getSegmentHashes(treeId, nodeIds);
	}

	@Override
	public SegmentData getSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException {
		return htStore.getSegmentData(treeId, segId, key);
	}

	@Override
	public List<SegmentData> getSegment(long treeId, int segId)
			throws IOException {
		return htStore.getSegment(treeId, segId);
	}

	@Override
	public int rebuildHashTree(long treeId, long fullRebuildPeriod)
			throws IOException {
		long lastFullRebuiltTime = htStore.getCompleteRebuiltTimestamp(treeId);
		boolean fullRebuild = (lastFullRebuiltTime == 0) ? true
				: ((fullRebuildPeriod < 0) ? false
						: (System.currentTimeMillis() - lastFullRebuiltTime) > fullRebuildPeriod);
		return rebuildHashTree(treeId, fullRebuild);
	}

	@Override
	public int rebuildHashTree(long treeId, boolean fullRebuild)
			throws IOException {
		List<Integer> dirtySegments = null;
		if (lockProvider.acquireLock(treeId)) {
			try {
				notifier.preRebuild(treeId, fullRebuild);
				long buildBeginTS = System.currentTimeMillis();
				if (fullRebuild)
					rebuildCompleteTree(treeId);
				dirtySegments = htStore.getDirtySegments(treeId);
				htStore.markSegments(treeId, dirtySegments);
				List<Integer> dirtyNodes = rebuildLeaves(treeId, dirtySegments);
				rebuildInternalNodes(treeId, dirtyNodes);
				htStore.unmarkSegments(treeId, dirtySegments);
				if (fullRebuild)
					htStore.setCompleteRebuiltTimestamp(treeId, buildBeginTS);
				notifier.postRebuild(treeId, fullRebuild);
				return dirtySegments.size();
			} catch (HashTreesCustomRuntimeException e) {
				if (dirtySegments != null)
					htStore.markSegments(treeId, dirtySegments);
			} finally {
				lockProvider.releaseLock(treeId);
			}
		}
		return 0;
	}

	/**
	 * This reads all the entries from the {@link Store}, and updates
	 * {@link HashTreesStore} with those (key,value) pairs. Also reads all the
	 * existing entries from {@link HashTreesStore} and if they don't exist on
	 * the {@link Store}, removes from {@link HashTreesStore}.
	 * 
	 * @param treeId
	 * @throws IOException
	 */
	private void rebuildCompleteTree(long treeId) throws IOException {
		Iterator<Map.Entry<byte[], byte[]>> itr = store.iterator(treeId);
		while (itr.hasNext()) {
			Map.Entry<byte[], byte[]> pair = itr.next();
			hPutInternal(HTOperation.PUT_IF_ABSENT,
					ByteBuffer.wrap(pair.getKey()),
					ByteBuffer.wrap(pair.getValue()));
		}
		Iterator<SegmentData> segDataItr = htStore
				.getSegmentDataIterator(treeId);
		while (segDataItr.hasNext()) {
			SegmentData sd = segDataItr.next();
			if (!store.contains(sd.getKey())) {
				hRemoveInternal(HTOperation.REMOVE_IF_ABSENT, sd.key);
			}
		}
	}

	/**
	 * Rebuilds the dirty segments, and updates the segment hashes of the
	 * leaves.
	 * 
	 * @param treeId
	 * @param dirtySegments
	 * @return corresponding nodeIds of the segments.
	 * @throws IOException
	 */
	private List<Integer> rebuildLeaves(long treeId,
			final List<Integer> dirtySegments) throws IOException {
		List<Integer> nodeIds = new ArrayList<>();
		for (int dirtySegId : dirtySegments) {
			if (htStore.clearDirtySegment(treeId, dirtySegId)) {
				ByteBuffer digest = digestSegmentData(treeId, dirtySegId);
				int nodeId = getLeafIdFromSegmentId(dirtySegId);
				htStore.putSegmentHash(treeId, nodeId, digest);
				nodeIds.add(nodeId);
			}
		}
		return nodeIds;
	}

	private ByteBuffer digestSegmentData(long treeId, int segId)
			throws IOException {
		List<SegmentData> dirtySegmentData = htStore.getSegment(treeId, segId);
		List<String> hexStrings = new ArrayList<String>();
		for (SegmentData sd : dirtySegmentData)
			hexStrings.add(getHexString(sd.key, sd.digest));
		return digestHexStrings(hexStrings);
	}

	/**
	 * Updates the segment hashes iteratively for each level on the tree.
	 * 
	 * @param dirtyNodeIds
	 * @throws IOException
	 */
	private void rebuildInternalNodes(long treeId,
			final List<Integer> dirtyNodeIds) throws IOException {
		Set<Integer> parentNodeIds = new TreeSet<Integer>();
		Set<Integer> nodeIds = new TreeSet<Integer>(dirtyNodeIds);
		while (!nodeIds.isEmpty()) {
			for (int nodeId : nodeIds)
				parentNodeIds.add(getParent(nodeId, noOfChildren));
			rebuildParentNodes(treeId, parentNodeIds);
			nodeIds.clear();
			nodeIds.addAll(parentNodeIds);
			parentNodeIds.clear();
			if (nodeIds.contains(ROOT_NODE))
				break;
		}
	}

	/**
	 * For each parent id, gets all the child hashes, and updates the parent
	 * hash.
	 * 
	 * @param parentIds
	 * @throws IOException
	 */
	private void rebuildParentNodes(long treeId, final Set<Integer> parentIds)
			throws IOException {
		List<Integer> children;
		List<ByteBuffer> segHashes = new ArrayList<>(noOfChildren);
		ByteBuffer segHashBB;
		SegmentHash segHash;

		for (int parentId : parentIds) {
			children = getImmediateChildren(parentId, noOfChildren);
			for (int child : children) {
				segHash = htStore.getSegmentHash(treeId, child);
				segHashBB = (segHash == null) ? null : segHash.hash.duplicate();
				if (segHashBB != null)
					segHashes.add(segHashBB);
			}
			ByteBuffer digest = digestByteBuffers(segHashes);
			htStore.putSegmentHash(treeId, parentId, digest);
			segHashes.clear();
		}
	}

	@Override
	public void sPut(final List<KeyValue> keyValuePairs) throws IOException {
		notifier.preSPut(keyValuePairs);
		for (KeyValue keyValuePair : keyValuePairs)
			store.put(keyValuePair.getKey(), keyValuePair.getValue());
		notifier.postSPut(keyValuePairs);
	}

	@Override
	public void sRemove(final List<ByteBuffer> keys) throws IOException {
		notifier.preSRemove(keys);
		for (ByteBuffer key : keys)
			store.delete(key.array());
		notifier.postSRemove(keys);
	}

	@Override
	public void deleteTreeNode(long treeId, int nodeId) throws IOException {
		int leftMostNodeId = getSegmentIdFromLeafId(getLeftMostChildNode(
				nodeId, noOfChildren, height));
		int rightMostNodeId = getSegmentIdFromLeafId(getRightMostChildNode(
				nodeId, noOfChildren, height));
		Iterator<SegmentData> sdItr = htStore.getSegmentDataIterator(treeId,
				leftMostNodeId, rightMostNodeId);
		while (sdItr.hasNext()) {
			SegmentData sd = sdItr.next();
			store.delete(sd.getKey());
		}
	}

	// Should be always called before using this instance.
	@Override
	public void start() {
		enableNonBlockingOperations();
	}

	private void enableNonBlockingOperations() {
		synchronized (nonBlockingCallsLock) {
			if (enabledNonBlockingCalls) {
				if (bgDataUpdater == null)
					bgDataUpdater = new NonBlockingHTDataUpdater(this,
							nonBlockingQueueSize);
				new Thread(bgDataUpdater).start();
				LOGGER.info("Non blocking calls are enabled.");
			}
		}
	}

	private void disableNonblockingOperations() {
		synchronized (nonBlockingCallsLock) {
			if (enabledNonBlockingCalls && bgDataUpdater != null) {
				CountDownLatch countDownLatch = new CountDownLatch(1);
				bgDataUpdater.stopAsync(countDownLatch);
				try {
					countDownLatch.await();
				} catch (InterruptedException e) {
					LOGGER.warn(
							"Exception occurred while waiting data updater to stop",
							e);
				}
				bgDataUpdater = null;
				LOGGER.info("Non blocking calls are disabled.");
			}
		}
	}

	@Override
	public void stop() {
		disableNonblockingOperations();
	}

	@Override
	public void addObserver(HashTreesObserver observer) {
		notifier.addObserver(observer);
	}

	@Override
	public void removeObserver(HashTreesObserver observer) {
		notifier.removeObserver(observer);
	}

	private static int compareSegNodeIds(SegmentHash left, SegmentHash right) {
		if (left == null && right == null)
			return 0;
		if (left == null)
			return 1;
		if (right == null)
			return -1;
		return left.getNodeId() - right.getNodeId();
	}

	private static int compareSegmentKeys(SegmentData left, SegmentData right) {
		if (left == null && right == null)
			return 0;
		if (left == null)
			return 1;
		if (right == null)
			return -1;
		return ByteUtils.compareTo(left.getKey(), right.getKey());
	}

	private static int getValidSegmentsCount(int noOfSegments) {
		return ((noOfSegments > MAX_NO_OF_SEGMENTS) || (noOfSegments < 0)) ? MAX_NO_OF_SEGMENTS
				: roundUpToPowerOf2(noOfSegments);
	}

	/**
	 * Concatenates the given ByteBuffer values by first converting them to the
	 * equivalent hex strings, and then concatenates by adding the comma
	 * delimiter.
	 * 
	 * @param values
	 * @return
	 */
	public static String getHexString(ByteBuffer... values) {
		StringBuffer sb = new StringBuffer();
		for (int i = 0; i < values.length - 1; i++)
			sb.append(Hex.encodeHexString(values[i].array()) + COMMA_DELIMETER);
		if (values.length > 0)
			sb.append(Hex.encodeHexString(values[values.length - 1].array()));
		return sb.toString();
	}

	public static ByteBuffer digestByteBuffers(List<ByteBuffer> bbList) {
		List<String> hexStrings = new ArrayList<String>();
		for (ByteBuffer bb : bbList)
			hexStrings.add(Hex.encodeHexString(bb.array()));
		return digestHexStrings(hexStrings);
	}

	public static ByteBuffer digestHexStrings(List<String> hexStrings) {
		StringBuilder sb = new StringBuilder();
		for (String hexString : hexStrings)
			sb.append(hexString + NEW_LINE_DELIMETER);
		return ByteBuffer.wrap(sha1(sb.toString().getBytes()));
	}

	/**
	 * Segment block id starts with 0. Each leaf node corresponds to a segment
	 * block. This function does the mapping from segment block id to leaf node
	 * id.
	 * 
	 * @param segId
	 * @return
	 */
	private int getLeafIdFromSegmentId(int segId) {
		return internalNodesCount + segId;
	}

	private int getSegmentIdFromLeafId(int leafNodeId) {
		return leafNodeId - internalNodesCount;
	}

	private boolean isLeafNode(int nodeId) {
		return nodeId >= internalNodesCount;
	}

	/**
	 * Used to tag type of operation when the input is fed into the non blocking
	 * version of {@link HashTreesImpl} hPut and hRemove methods.
	 * 
	 */
	private static enum HTOperation {
		PUT, REMOVE, PUT_IF_ABSENT, REMOVE_IF_ABSENT
	}

	/**
	 * A task to enable non blocking calls on all
	 * {@link HashTreesImpl#hPut(ByteArray, ByteArray)} and
	 * {@link HashTreesImpl#hRemove(ByteArray)} operation.
	 * 
	 */
	@ThreadSafe
	private static class NonBlockingHTDataUpdater extends
			NonBlockingQueuingTask<Pair<HTOperation, List<ByteBuffer>>> {

		private static final Pair<HTOperation, List<ByteBuffer>> STOP_MARKER = new Pair<HTOperation, List<ByteBuffer>>(
				HTOperation.PUT, null);
		private final ConcurrentSkipListSet<ByteBuffer> keysOnQueue = new ConcurrentSkipListSet<>();
		private final HashTreesImpl hTree;

		public NonBlockingHTDataUpdater(final HashTreesImpl hTree,
				int maxElementsToQue) {
			super(STOP_MARKER, maxElementsToQue);
			this.hTree = hTree;
		}

		@Override
		public void enque(Pair<HTOperation, List<ByteBuffer>> pair) {
			if (pair != STOP_MARKER) {
				ByteBuffer key = pair.getSecond().get(0);
				boolean isAbsent = keysOnQueue.add(key);
				switch (pair.getFirst()) {
				case PUT:
				case REMOVE:
					super.enque(pair);
					break;
				case PUT_IF_ABSENT:
				case REMOVE_IF_ABSENT:
					if (isAbsent)
						super.enque(pair);
					break;
				}
			}
		}

		@Override
		public void handleElement(Pair<HTOperation, List<ByteBuffer>> pair) {
			ByteBuffer key = pair.getSecond().get(0);
			try {
				switch (pair.getFirst()) {
				case PUT:
				case PUT_IF_ABSENT:
					hTree.hPutInternal(key, pair.getSecond().get(1));
					break;
				case REMOVE:
				case REMOVE_IF_ABSENT:
					hTree.hRemoveInternal(key);
					break;
				}
			} catch (IOException e) {
				throw new RuntimeException(e);
			} finally {
				keysOnQueue.remove(key);
			}
		}

	}

	@NotThreadSafe
	public static class Builder {

		private final Store store;
		private final HashTreesStore htStore;
		private final HashTreesIdProvider treeIdProvider;

		private SegmentIdProvider segIdProvider;
		private int noOfSegments = 1 << 17,
				nonBlockingQueueSize = DEFAULT_NB_QUE_SIZE;
		private boolean enabledNonBlockingCalls = true;
		private LockProvider lockProvider;

		public Builder(Store store, HashTreesIdProvider treeIdProvider,
				HashTreesStore htStore) {
			this.store = store;
			this.htStore = htStore;
			this.treeIdProvider = treeIdProvider;
		}

		/**
		 * Creates a in memory based hashtrees storage, and stores the digests
		 * over there. Look at {@link HashTreesMemStore} for more information.
		 * 
		 * @param store
		 * @param treeIdProvider
		 */
		public Builder(Store store, HashTreesIdProvider treeIdProvider) {
			this(store, treeIdProvider, new HashTreesMemStore());
		}

		/**
		 * Creates a LevelDB based store for storing hash tree contents. LevelDB
		 * files are stored in htStoreDirName. Look at
		 * {@link HashTreesPersistentStore}.
		 * 
		 * @param store
		 * @param treeIdProvider
		 * @param htStoreDirName
		 * @throws IOException
		 */
		public Builder(Store store, HashTreesIdProvider treeIdProvider,
				String htStoreDirName) throws IOException {
			this(store, treeIdProvider, new HashTreesPersistentStore(
					htStoreDirName));
		}

		/**
		 * By default {@link ModuloSegIdProvider} is used.
		 * 
		 * @param segIdProvider
		 * @return
		 */
		public Builder setSegmentIdProvider(SegmentIdProvider segIdProvider) {
			this.segIdProvider = segIdProvider;
			return this;
		}

		/**
		 * Depends upon data size, this should be set. Data size and
		 * noOfSegments should be directly proportional. With higher dataSize,
		 * and lesser noOfSegments means each segment will get more amount of
		 * data. When a segment is marked as dirty, the rebuild process has to
		 * read huge data unnecessarily.
		 * 
		 * Default value is 1 << 17. (More than 1 million).
		 * 
		 * @param noOfSegments
		 *            , value should be a power of 2, otherwise it will be
		 *            converted to the equivalent.
		 * @return
		 */
		public Builder setNoOfSegments(int noOfSegments) {
			this.noOfSegments = getValidSegmentsCount(noOfSegments);
			return this;
		}

		/**
		 * Enable/Disable non blocking calls. By default non blocking calls are
		 * enabled.
		 * 
		 * @param enabledNonBlockingCalls
		 *            , true enables non blocking calls, false disables non
		 *            blocking calls.
		 * @return
		 */
		public Builder setEnabledNonBlockingCalls(
				boolean enabledNonBlockingCalls) {
			this.enabledNonBlockingCalls = enabledNonBlockingCalls;
			return this;
		}

		/**
		 * Sets queue size about how many elements can be queued in memory.
		 * Default value is {@link Integer#MAX_VALUE}.
		 * 
		 * @param nonBlockingQueueSize
		 * @return
		 */
		public Builder setNonBlockingQueueSize(int nonBlockingQueueSize) {
			assert (nonBlockingQueueSize > 0);
			this.nonBlockingQueueSize = nonBlockingQueueSize;
			return this;
		}

		/**
		 * Sets a lock provider to use. By default this uses
		 * {@link HTReentrantLockProvider}.
		 * 
		 * @param lockProvider
		 * @return
		 */
		public Builder setLockProvider(LockProvider lockProvider) {
			assert (lockProvider != null);
			this.lockProvider = lockProvider;
			return this;
		}

		public HashTreesImpl build() {
			if (segIdProvider == null)
				segIdProvider = new ModuloSegIdProvider(noOfSegments);
			if (lockProvider == null)
				lockProvider = new HTReentrantLockProvider();
			return new HashTreesImpl(noOfSegments, enabledNonBlockingCalls,
					nonBlockingQueueSize, treeIdProvider, segIdProvider,
					htStore, store, lockProvider);
		}
	}
} public class ModuloSegIdProvider implements SegmentIdProvider {

	private final int noOfBuckets;

	public ModuloSegIdProvider(int noOfBuckets) {
		this.noOfBuckets = noOfBuckets;
	}

	@Override
	public int getSegmentId(byte[] key) {
		int hcode = ByteBuffer.wrap(key).hashCode();
		return hcode & (noOfBuckets - 1);
	}

} public class SyncDiffResult {
	public final int totKeyDifferences, totExtrinsicSegments;

	public SyncDiffResult(int totKeyDifferences, int totExtrinsicSegments) {
		this.totKeyDifferences = totKeyDifferences;
		this.totExtrinsicSegments = totExtrinsicSegments;
	}

	public boolean isAnyUpdatesMade() {
		return (totKeyDifferences > 0) || (totExtrinsicSegments > 0);
	}

	@Override
	public String toString() {
		return MoreObjects.toStringHelper(this)
				.add("totKeyDifferences", totKeyDifferences)
				.add("totExtrinsicSegments", totExtrinsicSegments).toString();
	}
} public enum SyncType {
	UPDATE // Updates the remote hash tree on finding differences
	, FIND_DIFF_ONLY // Only finds the differences between remote hash tree and
						// local tree.
} public interface HashTreesStore extends Service {

	/**
	 * A segment data is the value inside a segment block.
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 * @param digest
	 */
	void putSegmentData(long treeId, int segId, ByteBuffer key,
			ByteBuffer digest) throws IOException;

	/**
	 * Returns the SegmentData for the given key if available, otherwise returns
	 * null.
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 * @return
	 */
	SegmentData getSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException;

	/**
	 * Deletes the given segement data from the block.
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 */
	void deleteSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException;

	/**
	 * Returns an iterator to read all the segment data of the given tree id.
	 * 
	 * Note: Iterator implementations should throw
	 * {@link HashTreesCustomRuntimeException} so that failure cases can be
	 * handled properly by {@link HashTrees}
	 * 
	 * @param treeId
	 * @return
	 */
	Iterator<SegmentData> getSegmentDataIterator(long treeId)
			throws IOException;

	/**
	 * Returns an iterator to read all the segment data of the given tree id,
	 * and starting from a given segId and to segId.
	 * 
	 * Note: Iterator implementations should throw
	 * {@link HashTreesCustomRuntimeException} so that failure cases can be
	 * handled properly by {@link HashTrees}
	 * 
	 * @param treeId
	 * @param fromSegId
	 *            , inclusive
	 * @param toSegId
	 *            , inclusive
	 * @return
	 * @throws IOException
	 */
	Iterator<SegmentData> getSegmentDataIterator(long treeId, int fromSegId,
			int toSegId) throws IOException;

	/**
	 * Given a segment id, returns the list of all segment data in the
	 * individual segment block.
	 * 
	 * @param treeId
	 * @param segId
	 * @return
	 */
	List<SegmentData> getSegment(long treeId, int segId) throws IOException;

	/**
	 * Segment hash is the hash of all data inside a segment block. A segment
	 * hash is stored on a tree node.
	 * 
	 * @param treeId
	 * @param nodeId
	 *            , identifier of the node in the hash tree.
	 * @param digest
	 */
	void putSegmentHash(long treeId, int nodeId, ByteBuffer digest)
			throws IOException;

	/**
	 * 
	 * @param treeId
	 * @param nodeId
	 * @return
	 */
	SegmentHash getSegmentHash(long treeId, int nodeId) throws IOException;

	/**
	 * Returns the data inside the nodes of the hash tree. If the node id is not
	 * present in the hash tree, the entry will be missing in the result.
	 * 
	 * @param treeId
	 * @param nodeIds
	 *            , internal tree node ids.
	 * @return
	 */
	List<SegmentHash> getSegmentHashes(long treeId, Collection<Integer> nodeIds)
			throws IOException;

	/**
	 * Marks a segment as a dirty.
	 * 
	 * @param treeId
	 * @param segId
	 * @return previousValue
	 */
	boolean setDirtySegment(long treeId, int segId) throws IOException;

	/**
	 * Clears the segments, which are passed as an argument.
	 * 
	 * @param treeId
	 * @param segId
	 * @return previousValue
	 */
	boolean clearDirtySegment(long treeId, int segId) throws IOException;

	/**
	 * Gets the dirty segments without clearing those bits.
	 * 
	 * @param treeId
	 * @return
	 */
	List<Integer> getDirtySegments(long treeId) throws IOException;

	/**
	 * Sets flags for the given segments. Used during rebuild process by
	 * {@link HashTrees#rebuildHashTree(long, boolean)}. If the process crashes
	 * in the middle of rebuilding we don't want to loose track. This has to
	 * persist that information, so that we can reuse it after the process
	 * recovery.
	 * 
	 * @param segIds
	 */
	void markSegments(long treeId, List<Integer> segIds) throws IOException;

	/**
	 * Gets the marked segments for the given treeId.
	 * 
	 * @param treeId
	 * @return
	 */
	List<Integer> getMarkedSegments(long treeId) throws IOException;

	/**
	 * Unsets flags for the given segments. Used during rebuild process by
	 * {@link HashTrees#rebuildHashTree(long, boolean)}. After rebuilding is
	 * done, this will be cleared.
	 * 
	 * @param segIds
	 */
	void unmarkSegments(long treeId, List<Integer> segIds) throws IOException;

	/**
	 * Deletes the segment hashes, and segment data for the given treeId.
	 * 
	 * @param treeId
	 */
	void deleteTree(long treeId) throws IOException;

	/**
	 * Stores the timestamp at which the complete HashTree was rebuilt. This
	 * method updates the value in store only if the given value is higher than
	 * the existing timestamp, otherwise a noop.
	 * 
	 * @param timestamp
	 */
	void setCompleteRebuiltTimestamp(long treeId, long timestamp)
			throws IOException;

	/**
	 * Returns the timestamp at which the complete HashTree was rebuilt.
	 * 
	 * @return
	 */
	long getCompleteRebuiltTimestamp(long treeId) throws IOException;
} public interface Store {

	boolean contains(byte[] key) throws IOException;

	/**
	 * Adds the key and value to the local store. Also this call should forward
	 * the request to {@link HashTrees#hPut(ByteBuffer, ByteBuffer)} with the
	 * HashTrees object (If there is a hash trees registered already through
	 * {@link #registerHashTrees(HashTrees)}.
	 * 
	 * @param key
	 * @param value
	 * @throws IOException
	 */
	void put(byte[] key, byte[] value) throws IOException;

	byte[] get(byte[] key) throws IOException;

	/**
	 * Removes the key from the local store. Also this call should forward the
	 * request to {@link HashTrees#hRemove(ByteBuffer)} with the HashTrees
	 * object (If there is a hash trees registered already through
	 * {@link #registerHashTrees(HashTrees)}.
	 * 
	 * @param key
	 * @return
	 * @throws IOException
	 */
	void delete(byte[] key) throws IOException;

	/**
	 * This should return only (key,value) that are belonging to treeId. Used by
	 * {@link HashTrees#rebuildHashTree(long, boolean)} while rebuilding a
	 * specific hashtree.
	 * 
	 * Note: Iterator implementations should throw
	 * {@link HashTreesCustomRuntimeException} so that failure cases can be
	 * handled properly by {@link HashTrees}
	 * 
	 * @param treeId
	 * @return
	 * @throws IOException
	 */
	Iterator<Map.Entry<byte[], byte[]>> iterator(long treeId)
			throws IOException;

	/**
	 * Register hashtrees with storage implementation. On modifications to the
	 * storage, it should forward the calls to HashTrees.
	 * 
	 * @param hashTrees
	 */
	void registerHashTrees(HashTrees hashTrees);
} public class KeyValue implements org.apache.thrift.TBase<KeyValue, KeyValue._Fields>, java.io.Serializable, Cloneable {
  private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("KeyValue");

  private static final org.apache.thrift.protocol.TField KEY_FIELD_DESC = new org.apache.thrift.protocol.TField("key", org.apache.thrift.protocol.TType.STRING, (short)1);
  private static final org.apache.thrift.protocol.TField VALUE_FIELD_DESC = new org.apache.thrift.protocol.TField("value", org.apache.thrift.protocol.TType.STRING, (short)2);

  private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
  static {
    schemes.put(StandardScheme.class, new KeyValueStandardSchemeFactory());
    schemes.put(TupleScheme.class, new KeyValueTupleSchemeFactory());
  }

  public ByteBuffer key; // required
  public ByteBuffer value; // required

  /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
  public enum _Fields implements org.apache.thrift.TFieldIdEnum {
    KEY((short)1, "key"),
    VALUE((short)2, "value");

    private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();

    static {
      for (_Fields field : EnumSet.allOf(_Fields.class)) {
        byName.put(field.getFieldName(), field);
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, or null if its not found.
     */
    public static _Fields findByThriftId(int fieldId) {
      switch(fieldId) {
        case 1: // KEY
          return KEY;
        case 2: // VALUE
          return VALUE;
        default:
          return null;
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, throwing an exception
     * if it is not found.
     */
    public static _Fields findByThriftIdOrThrow(int fieldId) {
      _Fields fields = findByThriftId(fieldId);
      if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
      return fields;
    }

    /**
     * Find the _Fields constant that matches name, or null if its not found.
     */
    public static _Fields findByName(String name) {
      return byName.get(name);
    }

    private final short _thriftId;
    private final String _fieldName;

    _Fields(short thriftId, String fieldName) {
      _thriftId = thriftId;
      _fieldName = fieldName;
    }

    public short getThriftFieldId() {
      return _thriftId;
    }

    public String getFieldName() {
      return _fieldName;
    }
  }

  // isset id assignments
  public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
  static {
    Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
    tmpMap.put(_Fields.KEY, new org.apache.thrift.meta_data.FieldMetaData("key", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , true)));
    tmpMap.put(_Fields.VALUE, new org.apache.thrift.meta_data.FieldMetaData("value", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , true)));
    metaDataMap = Collections.unmodifiableMap(tmpMap);
    org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(KeyValue.class, metaDataMap);
  }

  public KeyValue() {
  }

  public KeyValue(
    ByteBuffer key,
    ByteBuffer value)
  {
    this();
    this.key = key;
    this.value = value;
  }

  /**
   * Performs a deep copy on <i>other</i>.
   */
  public KeyValue(KeyValue other) {
    if (other.isSetKey()) {
      this.key = org.apache.thrift.TBaseHelper.copyBinary(other.key);
;
    }
    if (other.isSetValue()) {
      this.value = org.apache.thrift.TBaseHelper.copyBinary(other.value);
;
    }
  }

  public KeyValue deepCopy() {
    return new KeyValue(this);
  }

  @Override
  public void clear() {
    this.key = null;
    this.value = null;
  }

  public byte[] getKey() {
    setKey(org.apache.thrift.TBaseHelper.rightSize(key));
    return key == null ? null : key.array();
  }

  public ByteBuffer bufferForKey() {
    return key;
  }

  public KeyValue setKey(byte[] key) {
    setKey(key == null ? (ByteBuffer)null : ByteBuffer.wrap(key));
    return this;
  }

  public KeyValue setKey(ByteBuffer key) {
    this.key = key;
    return this;
  }

  public void unsetKey() {
    this.key = null;
  }

  /** Returns true if field key is set (has been assigned a value) and false otherwise */
  public boolean isSetKey() {
    return this.key != null;
  }

  public void setKeyIsSet(boolean value) {
    if (!value) {
      this.key = null;
    }
  }

  public byte[] getValue() {
    setValue(org.apache.thrift.TBaseHelper.rightSize(value));
    return value == null ? null : value.array();
  }

  public ByteBuffer bufferForValue() {
    return value;
  }

  public KeyValue setValue(byte[] value) {
    setValue(value == null ? (ByteBuffer)null : ByteBuffer.wrap(value));
    return this;
  }

  public KeyValue setValue(ByteBuffer value) {
    this.value = value;
    return this;
  }

  public void unsetValue() {
    this.value = null;
  }

  /** Returns true if field value is set (has been assigned a value) and false otherwise */
  public boolean isSetValue() {
    return this.value != null;
  }

  public void setValueIsSet(boolean value) {
    if (!value) {
      this.value = null;
    }
  }

  public void setFieldValue(_Fields field, Object value) {
    switch (field) {
    case KEY:
      if (value == null) {
        unsetKey();
      } else {
        setKey((ByteBuffer)value);
      }
      break;

    case VALUE:
      if (value == null) {
        unsetValue();
      } else {
        setValue((ByteBuffer)value);
      }
      break;

    }
  }

  public Object getFieldValue(_Fields field) {
    switch (field) {
    case KEY:
      return getKey();

    case VALUE:
      return getValue();

    }
    throw new IllegalStateException();
  }

  /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
  public boolean isSet(_Fields field) {
    if (field == null) {
      throw new IllegalArgumentException();
    }

    switch (field) {
    case KEY:
      return isSetKey();
    case VALUE:
      return isSetValue();
    }
    throw new IllegalStateException();
  }

  @Override
  public boolean equals(Object that) {
    if (that == null)
      return false;
    if (that instanceof KeyValue)
      return this.equals((KeyValue)that);
    return false;
  }

  public boolean equals(KeyValue that) {
    if (that == null)
      return false;

    boolean this_present_key = true && this.isSetKey();
    boolean that_present_key = true && that.isSetKey();
    if (this_present_key || that_present_key) {
      if (!(this_present_key && that_present_key))
        return false;
      if (!this.key.equals(that.key))
        return false;
    }

    boolean this_present_value = true && this.isSetValue();
    boolean that_present_value = true && that.isSetValue();
    if (this_present_value || that_present_value) {
      if (!(this_present_value && that_present_value))
        return false;
      if (!this.value.equals(that.value))
        return false;
    }

    return true;
  }

  @Override
  public int hashCode() {
    return 0;
  }

  public int compareTo(KeyValue other) {
    if (!getClass().equals(other.getClass())) {
      return getClass().getName().compareTo(other.getClass().getName());
    }

    int lastComparison = 0;
    KeyValue typedOther = (KeyValue)other;

    lastComparison = Boolean.valueOf(isSetKey()).compareTo(typedOther.isSetKey());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetKey()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.key, typedOther.key);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetValue()).compareTo(typedOther.isSetValue());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetValue()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.value, typedOther.value);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    return 0;
  }

  public _Fields fieldForId(int fieldId) {
    return _Fields.findByThriftId(fieldId);
  }

  public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
    schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
  }

  public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
    schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder("KeyValue(");
    boolean first = true;

    sb.append("key:");
    if (this.key == null) {
      sb.append("null");
    } else {
      org.apache.thrift.TBaseHelper.toString(this.key, sb);
    }
    first = false;
    if (!first) sb.append(", ");
    sb.append("value:");
    if (this.value == null) {
      sb.append("null");
    } else {
      org.apache.thrift.TBaseHelper.toString(this.value, sb);
    }
    first = false;
    sb.append(")");
    return sb.toString();
  }

  public void validate() throws org.apache.thrift.TException {
    // check for required fields
    if (key == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'key' was not present! Struct: " + toString());
    }
    if (value == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'value' was not present! Struct: " + toString());
    }
  }

  private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
    try {
      write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
    try {
      read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private static class KeyValueStandardSchemeFactory implements SchemeFactory {
    public KeyValueStandardScheme getScheme() {
      return new KeyValueStandardScheme();
    }
  }

  private static class KeyValueStandardScheme extends StandardScheme<KeyValue> {

    public void read(org.apache.thrift.protocol.TProtocol iprot, KeyValue struct) throws org.apache.thrift.TException {
      org.apache.thrift.protocol.TField schemeField;
      iprot.readStructBegin();
      while (true)
      {
        schemeField = iprot.readFieldBegin();
        if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
          break;
        }
        switch (schemeField.id) {
          case 1: // KEY
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.key = iprot.readBinary();
              struct.setKeyIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          case 2: // VALUE
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.value = iprot.readBinary();
              struct.setValueIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          default:
            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
        }
        iprot.readFieldEnd();
      }
      iprot.readStructEnd();

      // check for required fields of primitive type, which can't be checked in the validate method
      struct.validate();
    }

    public void write(org.apache.thrift.protocol.TProtocol oprot, KeyValue struct) throws org.apache.thrift.TException {
      struct.validate();

      oprot.writeStructBegin(STRUCT_DESC);
      if (struct.key != null) {
        oprot.writeFieldBegin(KEY_FIELD_DESC);
        oprot.writeBinary(struct.key);
        oprot.writeFieldEnd();
      }
      if (struct.value != null) {
        oprot.writeFieldBegin(VALUE_FIELD_DESC);
        oprot.writeBinary(struct.value);
        oprot.writeFieldEnd();
      }
      oprot.writeFieldStop();
      oprot.writeStructEnd();
    }

  }

  private static class KeyValueTupleSchemeFactory implements SchemeFactory {
    public KeyValueTupleScheme getScheme() {
      return new KeyValueTupleScheme();
    }
  }

  private static class KeyValueTupleScheme extends TupleScheme<KeyValue> {

    @Override
    public void write(org.apache.thrift.protocol.TProtocol prot, KeyValue struct) throws org.apache.thrift.TException {
      TTupleProtocol oprot = (TTupleProtocol) prot;
      oprot.writeBinary(struct.key);
      oprot.writeBinary(struct.value);
    }

    @Override
    public void read(org.apache.thrift.protocol.TProtocol prot, KeyValue struct) throws org.apache.thrift.TException {
      TTupleProtocol iprot = (TTupleProtocol) prot;
      struct.key = iprot.readBinary();
      struct.setKeyIsSet(true);
      struct.value = iprot.readBinary();
      struct.setValueIsSet(true);
    }
  }

}
import org.hashtrees.SyncType;
import org.hashtrees.store.HashTreesStore;
import org.hashtrees.store.Store;
import org.hashtrees.thrift.generated.KeyValue;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.TREE_ID_PROVIDER;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import org.hashtrees.HTReentrantLockProvider;
import org.hashtrees.HashTrees;
import org.hashtrees.HashTreesImpl;
import org.hashtrees.ModuloSegIdProvider;
import org.hashtrees.SyncDiffResult;
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you 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.
 */
package org.hashtrees.test.utils;




public class HashTreesImplTestObj extends HashTreesImpl {

	private final BlockingQueue<HTSynchEvent> events;

	public static enum HTSynchEvent {
		UPDATE_SEGMENT, UPDATE_FULL_TREE, SYNCH, SYNCH_INITIATED
	}

	public HashTreesImplTestObj(final int noOfSegments,
			boolean enabledNonBlockingCalls, int nbQueSize,
			final HashTreesStore htStore, final Store store,
			BlockingQueue<HTSynchEvent> events) {
		super(noOfSegments, enabledNonBlockingCalls, nbQueSize,
				TREE_ID_PROVIDER, new ModuloSegIdProvider(noOfSegments),
				htStore, store, new HTReentrantLockProvider());
		this.events = events;
	}

	@Override
	public void sPut(List<KeyValue> keyValuePairs) throws IOException {
		super.sPut(keyValuePairs);
		events.add(HTSynchEvent.SYNCH_INITIATED);
	}

	@Override
	public void sRemove(List<ByteBuffer> keys) throws IOException {
		super.sRemove(keys);
		events.add(HTSynchEvent.SYNCH_INITIATED);
	}

	@Override
	public SyncDiffResult synch(long treeId, HashTrees remoteTree)