Web-Cam SecureChat

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using Pfz.Caching;
using Pfz.Threading;

namespace Pfz.Remoting
{
	/// <summary>
	/// Represents a "Channel" inside a StreamChanneller. This is used by the remoting
	/// mechanism to separate each thread communication channel inside a single tcp/ip
	/// connection.
	/// </summary>
	public sealed class Channel:
		ExceptionAwareStream
	{
		#region Private and internal fields
			internal int fId;
			internal int fRemoteId;
			
			internal Queue<byte[]> fInMessages = new Queue<byte[]>();
			private byte[] fActualMessage;
			private int fPositionInActualMessage;
			
			internal ManualResetEvent fWaitEvent = new ManualResetEvent(false);
			private byte[] fSendBuffer;
			private int fSendBufferPosition;
			
			private bool fCanFlush = true;
		#endregion
		
		#region Constructor
			internal Channel(StreamChanneller channeller)
			{
				fChanneller = channeller;
				fSendBuffer = new byte[channeller.fChannelBufferSize];
				
				GCUtils.Collected += p_Collected;
			}
		#endregion
		#region Dispose
			/// <summary>
			/// Frees all needed resources and informs the remote side.
			/// </summary>
			/// <param name="disposing">True if called from Dispose()</param>
			protected override void OnDispose(bool disposing)
			{
				GCUtils.Collected -= p_Collected;

				var waitEvent = fWaitEvent;
				if (waitEvent != null)
				{
					fWaitEvent = null;
					waitEvent.Set();
				}
				
				var channeller = fChanneller;
				if (channeller != null)
				{
					fChanneller = null;
					
					var runRemoveChannel = channeller.fRunRemoveChannel;
					if (runRemoveChannel != null)
						runRemoveChannel.Run(channeller.i_RemoveChannel, new KeyValuePair<int, int>(fId, fRemoteId));
				}
			
				base.OnDispose(disposing);
			}
		#endregion
		#region p_Collected
			private void p_Collected()
			{
				try
				{
					var inMessages = fInMessages;

					if (WasDisposed || inMessages == null)
					{
						GCUtils.Collected -= p_Collected;
						return;
					}
				
					AbortSafe.UnabortableLock
					(
						inMessages,
						() => inMessages.TrimExcess()
					);
				}
				catch
				{
				}
			}
		#endregion
		
		#region Properties
			#region Id
				/// <summary>
				/// Gets the Id given to this channel locally.
				/// </summary>
				public int Id
				{
					get
					{
						return fId;
					}
				}
			#endregion
			#region RemoteId
				/// <summary>
				/// Gets the Id given to this channel by the remote host.
				/// </summary>
				public int RemoteId
				{
					get
					{
						return fRemoteId;
					}
				}
			#endregion
		
			#region Channeller
				internal StreamChanneller fChanneller;
				
				/// <summary>
				/// Gets the channeller to which this channel belongs to.
				/// </summary>
				public StreamChanneller Channeller
				{
					get
					{
						return fChanneller;
					}
				}
			#endregion
			
			#region Length
				/// <summary>
				/// Property from Stream. Always returns -1.
				/// </summary>
				public override long Length
				{
					get { return -1; }
				}
			#endregion
			#region Position
				/// <summary>
				/// Property from Stream. Always returns -1 and throws a NotSupportedException
				/// if set.
				/// </summary>
				public override long Position
				{
					get
					{
						return -1;
					}
					set
					{
						throw new NotSupportedException();
					}
				}
			#endregion
			
			#region CanRead
				/// <summary>
				/// Property from Stream. Always return true.
				/// </summary>
				public override bool CanRead
				{
					get { return true; }
				}
			#endregion
			#region CanSeek
				/// <summary>
				/// Property from Stream. Always return false.
				/// </summary>
				public override bool CanSeek
				{
					get { return false; }
				}
			#endregion
			#region CanWrite
				/// <summary>
				/// Property from Stream. Always return true.
				/// </summary>
				public override bool CanWrite
				{
					get { return true; }
				}
			#endregion
		#endregion
		#region Methods
			#region Flush
				/// <summary>
				/// Sends all buffered data to the stream.
				/// </summary>
				/// 
				public override void Flush()
				{
					if (fCanFlush)
						p_Flush();
				}
			#endregion
			#region p_Flush
				private void p_Flush()
				{
					int count = fSendBufferPosition;
					
					if (count == 0)
						return;
						
					try
					{
						fSendBufferPosition = 0;
							
						byte[] bufferCopy = new byte[count + 8];
						BitConverter.GetBytes(fId).CopyTo(bufferCopy, 0);
						BitConverter.GetBytes(count).CopyTo(bufferCopy, 4);
						
						Buffer.BlockCopy(fSendBuffer, 0, bufferCopy, 8, count);
						
						var buffersToSend = fChanneller.fBuffersToSend;
						
						AbortSafe.UnabortableLock
						(
							buffersToSend,
							() => buffersToSend.Enqueue(bufferCopy)
						);
							
						fChanneller.fWriterEvent.Set();
					}
					catch(Exception exception)
					{
						if (!fChanneller.WasDisposed)
							fChanneller.Dispose(exception);
							
						throw;
					}
				}
			#endregion
			#region Read
				/// <summary>
				/// Reads bytes from the channel.
				/// </summary>
				/// <param name="buffer">The buffer to store the read data.</param>
				/// <param name="offset">The initial position to store data in the buffer.</param>
				/// <param name="count">The number of bytes expected to read.</param>
				/// <returns>The number of bytes actually read.</returns>
				public override int Read(byte[] buffer, int offset, int count)
				{
					CheckUndisposed();

					if (count == 0)
						return 0;
				
					byte[] actualMessage = fActualMessage;
					if (actualMessage == null)
					{
						bool mustBreak = false;

						while (true)
						{
							AbortSafe.UnabortableLock
							(
								fInMessages,
								delegate
								{
									if (fInMessages.Count > 0)
									{
										actualMessage = fInMessages.Dequeue();
										fActualMessage = actualMessage;
										fPositionInActualMessage = 0;
										mustBreak = true;
									}
								}
							);
							
							if (mustBreak)
								break;
							
							fWaitEvent.WaitOne();
							
							CheckUndisposed();
							
							fWaitEvent.Reset();
						}
					}
					
					int messageLength = actualMessage.Length;
					int positionInActualMessage = fPositionInActualMessage;
					int remainingLength = messageLength - positionInActualMessage;
					
					if (remainingLength <= count)
					{
						count = remainingLength;
						fActualMessage = null;
					}
					else
						fPositionInActualMessage += count;
					
					Buffer.BlockCopy(actualMessage, positionInActualMessage, buffer, offset, count);
					
					return count;
				}
			#endregion
			#region Write
				/// <summary>
				/// Writes bytes into this channel.
				/// </summary>
				/// <param name="buffer">The buffer to get bytes to write.</param>
				/// <param name="offset">The initial position in the buffer to send.</param>
				/// <param name="count">The number of bytes from the buffer to send.</param>
				public override void Write(byte[] buffer, int offset, int count)
				{
					int bufferSize = fSendBuffer.Length;
				
					int lastValue = offset + count;
					for (int i=offset; i<lastValue; i++)
					{
						fSendBuffer[fSendBufferPosition] = buffer[i];
						fSendBufferPosition++;
						
						if (fSendBufferPosition == bufferSize)
							p_Flush();
					}
				}
			#endregion

			#region Seek
				/// <summary>
				/// Method from Stream. Throws a NotSupportedException.
				/// </summary>
				public override long Seek(long offset, SeekOrigin origin)
				{
					throw new NotSupportedException();
				}
			#endregion
			#region SetLength
				/// <summary>
				/// Method from Stream. Throws a NotSupportedException.
				/// </summary>
				public override void SetLength(long value)
				{
					throw new NotSupportedException();
				}
			#endregion
		#endregion
	}
}