using System;
|
using System.IO;
|
using System.Threading;
|
namespace OpenTap
|
{
|
/// <summary> TeeStream allows many reader-streams to be created from one.
|
/// If one reader is slower than the others, it will block until the slowest catches up. </summary>
|
class TeeStream
|
{
|
/// <summary> Represents a client stream that reads from a shared TeeStream. </summary>
|
class TeeStreamClient : Stream
|
{
|
/// <summary> Keeps track of the current position in the stream for this client. </summary>
|
long globalOffset;
|
|
/// <summary> Reference to the host TeeStream that this client reads from. </summary>
|
readonly TeeStream streamHost;
|
public TeeStreamClient(TeeStream streamHost) => this.streamHost = streamHost;
|
|
/// <summary> This stream is read-only. Flush does nothing. </summary>
|
public override void Flush()
|
{
|
|
}
|
|
/// <summary> Reads a sequence of bytes from the current stream and advances the position within the stream. </summary>
|
public override int Read(byte[] buffer, int offset, int count)
|
{
|
int read = streamHost.Read(buffer, globalOffset, offset, count);
|
globalOffset += read;
|
return read;
|
}
|
public override long Seek(long offset, SeekOrigin origin)
|
{
|
throw new NotSupportedException();
|
}
|
public override void SetLength(long value)
|
{
|
throw new NotSupportedException();
|
}
|
public override void Write(byte[] buffer, int offset, int count)
|
{
|
throw new NotSupportedException();
|
}
|
public override bool CanRead => true;
|
public override bool CanSeek => false;
|
public override bool CanWrite => false;
|
public override long Length => streamHost.Length;
|
public override long Position
|
{
|
get => globalOffset;
|
set { }
|
}
|
|
protected override void Dispose(bool disposing)
|
{
|
base.Dispose(disposing);
|
|
// flush everything. There may be other peers so doing this makes sure that nobody waits for this client to read.
|
this.CopyTo(Stream.Null);
|
}
|
}
|
|
public long Length => mainStream.Length;
|
public long Position => mainStreamPosition - blockLength;
|
public long blockLength;
|
|
readonly Stream mainStream;
|
byte[] currentBlock;
|
|
public TeeStream(Stream mainStream) => this.mainStream = mainStream;
|
|
Stream CreateClientStream() => new TeeStreamClient(this);
|
|
public Stream[] CreateClientStreams(int count)
|
{
|
if (count == 0)
|
{
|
mainStream.Dispose();
|
return Array.Empty<Stream>();
|
}
|
currentBlock = new byte[4096 * count];
|
clientCount = count;
|
var result = new Stream[count];
|
for (int i = 0; i < count; i++)
|
{
|
result[i] = CreateClientStream();
|
}
|
return result;
|
}
|
bool done;
|
Exception readException = null;
|
|
void ReadNextBlock()
|
{
|
// at this point everyone is waiting for the next block.
|
int len;
|
try
|
{
|
len = mainStream.Read(currentBlock, 0, currentBlock.Length);
|
}
|
catch(Exception exception)
|
{
|
this.readException = exception;
|
len = 0;
|
}
|
|
// user interlocked.add to ensure that other threads will get the updated value.
|
Interlocked.Add(ref mainStreamPosition, len);
|
|
if (len == 0)
|
{
|
// We are done. let's stop.
|
done = true;
|
mainStream.Close();
|
mainStream.Dispose();
|
}
|
blockLength = len;
|
|
var oldEvt = evt;
|
var w2 = waiting;
|
evt = new SemaphoreSlim(0);
|
Interlocked.Exchange(ref waiting, 0);
|
oldEvt.Release(w2);
|
}
|
|
SemaphoreSlim evt = new SemaphoreSlim(0);
|
int waiting;
|
int clientCount;
|
long mainStreamPosition;
|
public int Read(byte[] buffer, long subStreamPosition, int bufferOffset, int count)
|
{
|
if (done) return 0;
|
if (readException != null)
|
throw readException;
|
|
// Offset into the current block.
|
long blockOffset = subStreamPosition - (mainStreamPosition - blockLength);
|
if (blockOffset < 0)
|
throw new InvalidOperationException("Unexpected position calculated");
|
|
var waitEvent = evt;
|
|
// if the block offset is greater than the size of the block, we need to get/wait for the next block.
|
if (blockOffset >= blockLength)
|
{
|
if (Interlocked.Increment(ref waiting) == clientCount)
|
{
|
// All clients are waiting - read the next block.
|
ReadNextBlock();
|
}
|
else
|
{
|
// wait for a new block.
|
waitEvent.Wait();
|
}
|
// new blocks released. start over.
|
return Read(buffer, subStreamPosition, bufferOffset, count);
|
}
|
|
// read the block byte-by-byte.
|
for (int i = 0; i < count; i++)
|
{
|
long o2 = subStreamPosition - (mainStreamPosition - blockLength) + i;
|
if (o2 >= blockLength)
|
{
|
// End of the block reached.
|
// start Read over with new args.
|
int r = Read(buffer, subStreamPosition + i, bufferOffset + i, count - i);
|
if (r == 0) return i;
|
return r + i;
|
}
|
buffer[i + bufferOffset] = currentBlock[o2];
|
}
|
return count;
|
}
|
}
|
}
|
