public void Flush()
        {
            _bufferSize = 0;

            using (var tx = _storageEnvironment.NewTransaction(TransactionFlags.ReadWrite))
            {
                foreach (var kvp in _buffer)
                {
                    var data = _storageEnvironment.CreateTree(tx, "channel:" + kvp.Key);
                    var buffer = new byte[16];
                    var key = new Slice(buffer);
                    var ms = new MemoryStream();
                    var bw = new BinaryWriter(ms);
                    foreach (var item in kvp.Value)
                    {
                        var date = item.Timestamp;
                        EndianBitConverter.Big.CopyBytes(date.Ticks, buffer, 0);
                        EndianBitConverter.Big.CopyBytes(_last++, buffer, 8);
                        ms.SetLength(0);
                        bw.Write(item.Value);
                        ms.Position = 0;

                        data.Add(tx, key, ms);
                    }
                }

                tx.State.Root.Add(tx, _lastKey, new MemoryStream(BitConverter.GetBytes(_last)));
                tx.Commit();
            }
            _buffer.Clear();
        }

		public override void Accept(Disk d)
		{
			var ms = new MemoryStream();
			_serializer.Serialize(new JsonTextWriter(new StreamWriter(ms)), d);
			ms.Position = 0;
			var key = new Slice(EndianBitConverter.Big.GetBytes(counter++));
			_currentBatch.Add(key, ms, "albums");

			foreach (var diskId in d.DiskIds)
			{
				_currentBatch.MultiAdd(diskId, key, "ix_diskids");
			}

			if(d.Artist != null)
				_currentBatch.MultiAdd(d.Artist.ToLower(), key, "ix_artists");
			if (d.Title != null)
				_currentBatch.MultiAdd(d.Title.ToLower(), key, "ix_titles");

			if (counter%1000 == 0)
			{
				_storageEnvironment.Writer.Write(_currentBatch);
				_currentBatch = new WriteBatch();
			}

		}

 public static void SetInline(Slice slice, NodeHeader* node)
 {
     slice.Pointer = (byte*)node + Constants.NodeHeaderSize;
     slice.Size = node->KeySize;
     slice.KeyLength = node->KeySize;
     slice.Array = null;
 }

        public virtual void Add(WriteBatch writeBatch, Slice key, byte[] value, ushort? expectedVersion = null)
        {
            var stream = new BufferPoolMemoryStream();
            stream.Write(value, 0, value.Length);
            stream.Position = 0;

            writeBatch.Add(key, stream, TableName, expectedVersion);
        }

 public PrefixedSlice()
 {
     Options = SliceOptions.Key;
     Size = 0;
     KeyLength = 0;
     Header = new PrefixedSliceHeader();
     NonPrefixedData = new Slice(SliceOptions.Key);
 }

		public virtual void Add(WriteBatch writeBatch, Slice key, RavenJToken value, ushort? expectedVersion = null)
		{
            var stream = new BufferPoolMemoryStream(BufferPool);
            value.WriteTo(stream);
            stream.Position = 0;

			writeBatch.Add(key, stream, TableName, expectedVersion);
		}

		public static int Compare(Slice x, PrefixedSlice y, SliceComparer cmp, ushort size)
		{
			fixed (byte* p1 = x.Array)
			fixed (byte* p2 = y.NonPrefixedData.Array)
			{
				var xPtr = p1 != null ? p1 : x.Pointer;
				var yPtr = p2 != null ? p2 : y.NonPrefixedData.Pointer;

				if (y.Header.PrefixId == PrefixedSlice.NonPrefixedId)
					return Compare(null, 0, null, 0, xPtr, x.KeyLength, yPtr, y.Header.NonPrefixedDataSize, cmp, size);

				if (x.PrefixComparisonCache == null)
				{
					if(y.Prefix == null)
						return Compare(null, 0, null, 0, xPtr, x.KeyLength, yPtr, y.Header.NonPrefixedDataSize, cmp, size);
					else if (y.Prefix.Value == null)
						return Compare(null, 0, y.Prefix.ValuePtr, y.Header.PrefixUsage, xPtr, x.KeyLength, yPtr, y.Header.NonPrefixedDataSize, cmp, size);
					else
					{
						fixed (byte* prefixVal = y.Prefix.Value)
							return Compare(null, 0, prefixVal, y.Header.PrefixUsage, xPtr, x.KeyLength, yPtr, y.Header.NonPrefixedDataSize, cmp, size);
					}
				}

				var prefixBytesToCompare = Math.Min(y.Header.PrefixUsage, x.KeyLength);

				int r;

				if (x.PrefixComparisonCache.TryGetCachedResult(y.Header.PrefixId, y.Prefix.PageNumber, prefixBytesToCompare, out r) == false)
				{
					if (y.Prefix == null)
						r = Compare(null, 0, null, 0, xPtr, x.KeyLength, null, 0, cmp,
							prefixBytesToCompare);

					else if (y.Prefix.Value == null)
						r = Compare(null, 0, y.Prefix.ValuePtr, y.Header.PrefixUsage, xPtr, x.KeyLength, null, 0, cmp,
							prefixBytesToCompare);
					else
					{
						fixed (byte* prefixVal = y.Prefix.Value)
							r = Compare(null, 0, prefixVal, y.Header.PrefixUsage, xPtr, x.KeyLength, null, 0, cmp,
								prefixBytesToCompare);
					}

					x.PrefixComparisonCache.SetPrefixComparisonResult(y.Header.PrefixId, y.Prefix.PageNumber, prefixBytesToCompare, r);
				}

				if (r != 0)
					return r;

				size -= prefixBytesToCompare;

				return Compare(null, 0, null, 0, xPtr + prefixBytesToCompare, (ushort)(x.KeyLength - prefixBytesToCompare), yPtr, y.Header.NonPrefixedDataSize, cmp, size);
			}
		}

		public Slice(Slice other, ushort size)
		{
			if (other.Array != null)
				Array = other.Array;
			else
				Pointer = other.Pointer;

			Options = other.Options;
			Size = size;
			KeyLength = size;
		}

		private void Initialize()
		{
			using (var tx = storageEnvironment.NewTransaction(TransactionFlags.ReadWrite))
			{
				var serverNamesToIds = storageEnvironment.CreateTree(tx, "serverNames->Ids");
				var serverIdsToNames = storageEnvironment.CreateTree(tx, "Ids->serverNames");
				storageEnvironment.CreateTree(tx, "servers->lastEtag");
				storageEnvironment.CreateTree(tx, "counters");
				storageEnvironment.CreateTree(tx, "countersGroups");
				var etags = storageEnvironment.CreateTree(tx, "etags->counters");
				storageEnvironment.CreateTree(tx, "counters->etags");
				
				var metadata = tx.State.GetTree(tx, "$metadata");
				var id = metadata.Read("id");

				if (id == null) // new counter db
				{
					var serverIdBytes = EndianBitConverter.Big.GetBytes(0); // local is always 0
					var serverIdSlice = new Slice(serverIdBytes);
					serverNamesToIds.Add(CounterStorageUrl, serverIdSlice);
					serverIdsToNames.Add(serverIdSlice, CounterStorageUrl);

					Id = Guid.NewGuid();
					metadata.Add("id", Id.ToByteArray());
					metadata.Add("name", Encoding.UTF8.GetBytes(CounterStorageUrl));

					tx.Commit();
				}
				else // existing counter db
				{
					int used;
					Id = new Guid(id.Reader.ReadBytes(16, out used));
					var nameResult = metadata.Read("name");
					if (nameResult == null)
						throw new InvalidOperationException("Could not read name from the store, something bad happened");
					var storedName = new StreamReader(nameResult.Reader.AsStream()).ReadToEnd();

					if (storedName != CounterStorageUrl)
						throw new InvalidOperationException("The stored name " + storedName + " does not match the given name " + CounterStorageUrl);

					using (var it = etags.Iterate())
					{
						if (it.Seek(Slice.AfterAllKeys))
						{
							LastEtag = it.CurrentKey.CreateReader().ReadBigEndianInt64();
						}
					}
				}

                ReplicationTask.StartReplication();
			}
		}

        protected override void Init()
        {
            base.Init();

            _positionsTree = Transaction.ReadTree("TermPositions");
            _docsTree = Transaction.ReadTree("Docs");

            _fieldId = Index.GetFieldNumber(Field);
            _prefix = new byte[FullTextIndex.FieldDocumentSize];
            _prefixSlice = new Slice(_prefix);

            _maxKey = new byte[FullTextIndex.FieldDocumentSize];
            _maxKeySlice = new Slice(_maxKey);
        }

 protected RavenJObject LoadJson(Table table, Slice key, WriteBatch writeBatch, out ushort version)
 {
     var read = table.Read(Snapshot, key, writeBatch);
     if (read == null)
     {
         version = table.ReadVersion(Snapshot, key, writeBatch) ?? 0;
         return null;
     } 
     
     using (var stream = read.Reader.AsStream())
     {
         version = read.Version;
         return stream.ToJObject();
     }
 }

		public void SetupDatabaseIdAndSchemaVersion()
		{
			using (var snapshot = storage.CreateSnapshot())
			{
                var idSlice = new Slice("id");
                var schemaVersionSlice = new Slice("schema_version");

				Guid id;
				string schemaVersion;

                var read = storage.Details.Read(snapshot, idSlice, null);
				if (read == null) // new db
				{
					id = Guid.NewGuid();
					schemaVersion = SchemaVersion;
					using (var writeIdBatch = new WriteBatch())
					{
                        storage.Details.Add(writeIdBatch, idSlice, id.ToByteArray());
                        storage.Details.Add(writeIdBatch, schemaVersionSlice, schemaVersion);
						storage.Write(writeIdBatch);
					}
				}
				else
				{
					if (read.Reader.Length != 16) //precaution - might prevent NRE in edge cases
						throw new InvalidDataException("Failed to initialize Voron transactional storage. Possible data corruption. (no db id)");

					using (var stream = read.Reader.AsStream())
					using (var reader = new BinaryReader(stream))
					{
						id = new Guid(reader.ReadBytes((int)stream.Length));
					}

                    var schemaRead = storage.Details.Read(snapshot, schemaVersionSlice, null);
					if (schemaRead == null)
						throw new InvalidDataException("Failed to initialize Voron transactional storage. Possible data corruption. (no schema version)");

					schemaVersion = schemaRead.Reader.ToStringValue();
				}

				storage.SetDatabaseIdAndSchemaVersion(id, schemaVersion);
			}
		}

		public PrefixedSlice(NodeHeader* node)
		{
			if (node->KeySize > 0)
			{
				var prefixHeaderPtr = (PrefixedSliceHeader*)((byte*)node + Constants.NodeHeaderSize);
				Header = *prefixHeaderPtr;

				NonPrefixedData = new Slice((byte*)prefixHeaderPtr + Constants.PrefixedSliceHeaderSize, Header.NonPrefixedDataSize);

				Size = node->KeySize;
				KeyLength = (ushort) (Header.PrefixUsage + Header.NonPrefixedDataSize);
			}
			else
			{
				Size = 0;
				KeyLength = 0;
			}

			Options = SliceOptions.Key;
		}

        public override int Accept(string d)
        {
            var disk = JObject.Parse(d);

            var ms = new MemoryStream();
            var writer = new StreamWriter(ms);
            writer.Write(d);
            writer.Flush();
            ms.Position = 0;
            var key = new Slice(EndianBitConverter.Big.GetBytes(counter++));
            _currentBatch.Add(key, ms, "albums");
            int count = 1;

            foreach (var diskId in disk.Value<JArray>("DiskIds"))
            {
                count++;
                _currentBatch.MultiAdd(diskId.Value<string>(), key, "ix_diskids");
            }

            var artist = disk.Value<string>("Artist");
            if (artist != null)
            {
                count++;
                _currentBatch.MultiAdd(artist.ToLower(), key, "ix_artists");
            }
            var title = disk.Value<string>("Title");
            if (title != null)
            {
                count++;
                _currentBatch.MultiAdd(title.ToLower(), key, "ix_titles");
            }

            if (counter % 500 == 0)
            {
                _storageEnvironment.Writer.Write(_currentBatch);
                _currentBatch = new WriteBatch();
            }
            return count;
        }

        //returns true if it was update operation
        private bool WriteDocumentMetadata(JsonDocumentMetadata metadata, Slice key, bool shouldIgnoreConcurrencyExceptions = false)
        {
            var metadataStream = CreateStream();

            metadataStream.Write(metadata.Etag);
            metadataStream.Write(metadata.Key);

            if (metadata.LastModified.HasValue)
                metadataStream.Write(metadata.LastModified.Value.ToBinary());
            else
                metadataStream.Write((long)0);

            metadata.Metadata.WriteTo(metadataStream);

            metadataStream.Position = 0;

            ushort? existingVersion;
            var isUpdate = metadataIndex.Contains(Snapshot, key, writeBatch.Value, out existingVersion);
            metadataIndex.Add(writeBatch.Value, key, metadataStream, existingVersion, shouldIgnoreConcurrencyExceptions);

            return isUpdate;
        }

 private bool PutDocumentMetadataInternal(string key, Slice normalizedKey, RavenJObject metadata, Etag newEtag, DateTime savedAt)
 {
     return WriteDocumentMetadata(new JsonDocumentMetadata
     {
         Key = key,
         Etag = newEtag,
         Metadata = metadata,
         LastModified = savedAt
     }, normalizedKey);
 }

		public bool Contains(SnapshotReader snapshot, Slice key, WriteBatch writeBatch, out ushort? version)
		{
			return snapshot.Contains(TableName, key, out version, writeBatch);
		}

		public virtual IIterator MultiRead(SnapshotReader snapshot, Slice key)
		{
			return snapshot.MultiRead(TableName, key);
		}

		public virtual ReadResult Read(SnapshotReader snapshot, Slice key, WriteBatch writeBatch)
		{
			return snapshot.Read(TableName, key, writeBatch);
		}

		public virtual void Increment(WriteBatch writeBatch, Slice key, long delta, ushort? expectedVersion = null)
		{
			writeBatch.Increment(key, delta, TableName, expectedVersion);
		}