BlockReader.cs

using System;
using System.IO;
using System.Runtime.ExceptionServices;
using System.Security.Cryptography;
using System.Text;
using System.Threading.Tasks;
using Waher.Persistence.Serialization;
using Waher.Runtime.Inventory;
 
namespace Waher.Persistence.NeuroLedger
{
    public delegate Task BinaryDataReadCallback(byte[] Data, bool More, object State);
 
    public class BlockReader : IDisposable
    {
        private const int BufferSize = 65536;
 
        private readonly byte[] buffer = new byte[BufferSize];
        private readonly byte[] signature;
        private readonly string collectionName;
        private BinaryDeserializer reader;
        private BlockHeader header = null;
        private Stream input;
        private readonly ulong bytes;
        private long bytesLeft;
        private int bufPos = 0;
        private int bufLen = 0;
 
        private BlockReader(Stream Input, long InputLength, string CollectionName, byte[] Signature, ulong Bytes)
        {
            this.input = Input;
            this.bytesLeft = InputLength;
            this.collectionName = CollectionName;
            this.signature = Signature;
            this.bytes = Bytes;
        }
 
        public static Stream GetStream(string FileName, NeuroLedgerProvider Provider)
        {
            FileStream fs = null;
            ICryptoTransform Aes = null;
 
            try
            {
                fs = File.OpenRead(FileName);
                Aes = Provider.GetAes(FileName, false);
                return new CryptoStream(fs, Aes, CryptoStreamMode.Read);
            }
            catch (Exception ex)
            {
                fs?.Dispose();
                Aes?.Dispose();
 
                ExceptionDispatchInfo.Capture(ex).Throw();
                return null;
            }
        }
 
        public static async Task<BlockReader> CreateAsync(string FileName, NeuroLedgerProvider Provider)
        {
            FileStream fs = null;
            ICryptoTransform Aes = null;
            CryptoStream cs = null;
            BlockReader Result = null;
 
            try
            {
                fs = File.OpenRead(FileName);
                Aes = Provider.GetAes(FileName, false);
                cs = new CryptoStream(fs, Aes, CryptoStreamMode.Read);
 
                Result = await CreateAsync(cs, Provider);
            }
            catch (Exception ex)
            {
                cs?.Dispose();
                fs?.Dispose();
                Aes?.Dispose();
 
                ExceptionDispatchInfo.Capture(ex).Throw();
            }
 
            return Result;
        }
 
        public static async Task<BlockReader> CreateAsync(Stream cs, NeuroLedgerProvider Provider)
        {
            int CollectionLen = cs.ReadByte();
            if (CollectionLen == 0)
                CollectionLen = 256;
 
            byte[] CollectionBin = new byte[CollectionLen];
            await cs.ReadAllAsync(CollectionBin, 0, CollectionLen);
 
            string CollectionName = Encoding.UTF8.GetString(CollectionBin);
 
            int Bytes = 1 + CollectionLen;
 
            int SigLen = cs.ReadByte();
            if (SigLen == 0)
                SigLen = 256;
 
            byte[] Signature = new byte[SigLen];
            await cs.ReadAllAsync(Signature, 0, SigLen);
 
            Bytes += 1 + SigLen;
 
            // TODO: Validate signature before processing contents of file.
 
            byte[] ContentLen = new byte[8];
            await cs.ReadAllAsync(ContentLen, 0, 8);
 
            long InputLength = BitConverter.ToInt64(ContentLen, 0);
 
            Bytes += 8;
 
            ObjectSerializer HeaderSerializer = await Provider.GetObjectSerializerEx(typeof(BlockHeader));
            BlockReader Result = null;
 
            try
            {
                Result = new BlockReader(cs, InputLength, CollectionName, Signature, (ulong)(InputLength + Bytes));
 
                int Len = (int)Math.Min(1024, Result.bytesLeft);
                byte[] Bin = await Result.ReadBytesAsync(Len);
 
                Result.reader = new BinaryDeserializer(CollectionName, Encoding.UTF8, Bin, 0);
                Result.header = (BlockHeader)await HeaderSerializer.Deserialize(Result.reader, ObjectSerializer.TYPE_OBJECT, false);
 
                Result.bufPos -= Bin.Length - Result.reader.Position;
            }
            catch (Exception ex)
            {
                Result?.Dispose();
                ExceptionDispatchInfo.Capture(ex).Throw();
            }
 
            return Result;
        }
 
        public ulong Bytes => this.bytes;
 
        private async Task<byte[]> ReadBytesAsync(int Bytes)
        {
            byte[] Result = new byte[Bytes];
            int Pos = 0;
            int i;
 
            while (Pos < Bytes)
            {
                if (this.bufPos >= this.bufLen)
                    await this.ReadMore();
 
                i = Math.Min(Bytes - Pos, this.bufLen - this.bufPos);
                Array.Copy(this.buffer, this.bufPos, Result, Pos, i);
                Pos += i;
                this.bufPos += i;
            }
 
            return Result;
        }
 
        private async Task ReadMore()
        {
            this.bufPos = 0;
            this.bufLen = await this.input.TryReadAllAsync(this.buffer, 0, BufferSize);
            if (this.bufLen <= 0)
                throw new IOException("Unable to read from block.");
 
            this.bytesLeft -= this.bufLen;
            if (this.bytesLeft < 0)
            {
                this.bufLen += (int)this.bytesLeft;
                this.bytesLeft = 0;
            }
        }
 
        private async Task<byte> ReadByteAsync()
        {
            if (this.bufPos >= this.bufLen)
                await this.ReadMore();
 
            return this.buffer[this.bufPos++];
        }
 
        public bool EOF
        {
            get
            {
                return (this.bytesLeft <= 0 && this.bufPos >= this.bufLen);
            }
        }
 
        public BlockHeader Header => this.header;
 
        public string CollectionName => this.collectionName;
 
        public byte[] Signature => this.signature;
 
        public byte[] Buffer => this.buffer;
 
        public int BufferPosition => this.bufPos;
 
        public void Dispose()
        {
            this.input?.Dispose();
            this.input = null;
        }
 
        public Entry ReadEntry()
        {
            return this.ReadEntryAsync().Result;
        }
 
        public async Task<Entry> ReadEntryAsync()
        {
            byte[] Buf = await this.ReadBytesAsync(10);
 
            EntryType EntryType = (EntryType)Buf[0];
            long Ticks = BitConverter.ToInt64(Buf, 1);
            DateTime Timestamp = new DateTime(Ticks, DateTimeKind.Utc);
            int b = Buf[9];
            int Len = b & 127;
            int Offset = 7;
 
            while ((b & 128) != 0)
            {
                b = await this.ReadByteAsync();
                Len |= (b & 127) << Offset;
                Offset += 7;
            }
 
            Buf = await this.ReadBytesAsync(Len);
 
            return new Entry(EntryType, Timestamp, Buf);
        }
 
    }
}