rudpsocket.cs

rudp可靠保障得udp传输

		public void Listen(int max)
		{
			// Nothing to do !
		}

		#endregion

		#region Connect

		public void Connect(IPEndPoint endPoint)
		{
			IAsyncResult result = BeginConnect(endPoint, null, null);
			EndConnect(result);

			if (_status == RUDPSocketStatus.Closed)
				throw new SocketException();
		}

		public IAsyncResult BeginConnect(IPEndPoint remoteEP,
											AsyncCallback callback,
											Object state)
		{
			if (_physical == null)
				AutomaticBind();

			RUDPConnectIAsyncResult asyncResult = new RUDPConnectIAsyncResult(this, callback, state);
			Interlocked.Exchange<RUDPConnectIAsyncResult>(ref _asyncResultConnect, asyncResult);

			_remoteEndPoint = remoteEP;
			RUDPSocketError result = _physical.BeginConnect(this, DefaultConnectionTimeOut);
			if (result != RUDPSocketError.Success)
			{
				Interlocked.Exchange<RUDPConnectIAsyncResult>(ref _asyncResultConnect, null);
				_remoteEndPoint = null;
				throw new RUDPSocketException(result);
			}

			return asyncResult;
		}

		public void EndConnect(IAsyncResult asyncResult)
		{
			((RUDPConnectIAsyncResult)asyncResult).EndInvoke(true);
		}

		#endregion

		#region Accept

		public RUDPSocket Accept()
		{
			IAsyncResult result = BeginAccept(null, null);
			return EndAccept(result);
		}

		public IAsyncResult BeginAccept(AsyncCallback callback, Object state)
		{
			RUDPAcceptIAsyncResult asyncResult = new RUDPAcceptIAsyncResult(this, callback, state);
			Interlocked.Exchange<RUDPAcceptIAsyncResult>(ref _asyncResultAccept, asyncResult);

			//---- Check if we do not already have a socket
			if (_acceptedRUDPSockets.Count > 0)
			{
				RUDPSocket rudp = _acceptedRUDPSockets[0];
				lock (_acceptedRUDPSockets)
					_acceptedRUDPSockets.RemoveAt(0);

				OnEndAccept(rudp);
			}
			else
			{
				//-- Request an accept
				_physical.BeginAccept(this, callback, state);
			}

			return asyncResult;
		}

		public RUDPSocket EndAccept(IAsyncResult asyncResult)
		{
			RUDPAcceptIAsyncResult result = asyncResult as RUDPAcceptIAsyncResult;
			result.EndInvoke(true);
			return result.AcceptedSocket;
		}

		#endregion

		#region Send

		public int Send(byte[] buffer,
						int offset,
						int size)
		{
			RUDPSocketError errorCode = RUDPSocketError.Success;
			return Send(buffer, offset, size, out errorCode, true);
		}

		public int Send(byte[] buffer,
						int offset,
						int size,
						out RUDPSocketError errorCode)
		{
			return Send(buffer, offset, size, out errorCode, true);
		}

		public int Send(byte[] buffer,
						int offset,
						int size,
						out RUDPSocketError errorCode,
						bool reliable)
		{
			errorCode = RUDPStack.SendPayload(this, buffer, offset, size, reliable, null);

			if (errorCode != RUDPSocketError.Success)
				return -1;

			return size;
		}

		public IAsyncResult BeginSend(byte[] buffer,
										int offset,
										int size,
										out RUDPSocketError errorCode,
										AsyncCallback callback,
										Object state)
		{
			return BeginSend(buffer, offset, size, out errorCode, callback, state, true);
		}

		public IAsyncResult BeginSend(byte[] buffer,
										int offset,
										int size,
										out RUDPSocketError errorCode,
										AsyncCallback callback,
										Object state,
										bool reliable)
		{
			RUDPSendIAsyncResult asyncResult = new RUDPSendIAsyncResult(this, callback, state, size);

			errorCode = RUDPStack.SendPayload(this, buffer, offset, size, reliable, asyncResult);

			if (errorCode != RUDPSocketError.Success)
				return null;

			return asyncResult;
		}

		public int EndSend(IAsyncResult asyncResult)
		{
			RUDPSendIAsyncResult result = asyncResult as RUDPSendIAsyncResult;
			result.EndInvoke(false);

			if (result.SocketError != RUDPSocketError.Success)
				return -1;

			return result._size;
		}

		#endregion

		#region Receive

		public byte[] Receive()
		{
			SocketError error = SocketError.Success;
			IAsyncResult result = BeginReceive(out error, null, null);
			return EndReceive(result);
		}

		public byte[] Receive(out SocketError errorCode)
		{
			IAsyncResult result = BeginReceive(out errorCode, null, null);
			return EndReceive(result);
		}

		public IAsyncResult BeginReceive(AsyncCallback callback,
									Object state)
		{
			SocketError errorCode = SocketError.Success;
			return BeginReceive(out errorCode, callback, state);
		}

		public IAsyncResult BeginReceive(out SocketError errorCode,
										AsyncCallback callback,
										Object state)
		{
			RUDPReceiveIAsyncResult asyncResult = new RUDPReceiveIAsyncResult(this, callback, state);
			Interlocked.Exchange<RUDPReceiveIAsyncResult>(ref _asyncResultReceive, asyncResult);

			errorCode = SocketError.Success;

			//---- Check if we do not already have packets
			HandleNextUserPacket(true);

			return asyncResult;
		}

		/// <summary>
		/// We directly return the payload to avoid memory copy.
		/// </summary>
		public byte[] EndReceive(IAsyncResult asyncResult)
		{
			RUDPReceiveIAsyncResult result = asyncResult as RUDPReceiveIAsyncResult;
			result.EndInvoke(false);

			if (result.SocketError != RUDPSocketError.Success)
				return null;

			return result.Packet.Payload;
		}

		#endregion

		#region Close

		public void Close()
		{
			_physical.Close(this);
		}

		#endregion

		#region Shutdown

		public void Shutdown()
		{
			_physical.Shutdown(this);
		}

		#endregion

		#region Properties

		/// <summary>
		/// Gets a value that indicates whether a ServerSocket is connected to a remote host as
		/// of the last Send or Receive operation.
		/// </summary>
		public bool Connected
		{
			get
			{
				return (_status == RUDPSocketStatus.Connected);
			}
		}

		public bool IsRendezVousMode
		{
			get
			{
				return _isRendezVousMode;
			}
			set
			{
				_isRendezVousMode = value;
			}
		}

		public int MTU
		{
			get
			{
				return _mtu;
			}
			set
			{
				_mtu = value;
			}
		}

		public bool UsePMTUDiscovery
		{
			get
			{
				return _usePMTUDiscovery;
			}
			set
			{
				_usePMTUDiscovery = value;
			}
		}

		public int Handle
		{
			get
			{
				return _handle;
			}
		}

		public RUDPSocketNetworkInformation NetworkInformation
		{
			get
			{
				return _networkInformation;
			}
		}

		/// <summary>
		/// Gets or sets a value that specifies the size of the receive buffer of the Socket.
		/// </summary>
		public int ReceiveBufferSize
		{
			get
			{
				return _receiveSize;
			}
			set
			{
				_receiveSize = value;
			}
		}

		/// <summary>
		/// Gets or sets a value that specifies the size of the send buffer of the Socket.
		/// </summary>
		public int SendBufferSize
		{
			get
			{
				return _sendSize;
			}
			set
			{
				_sendSize = value;
			}
		}

		#endregion

		#region OnEnd...

		internal void OnEndReceive(RUDPSocketError error, RUDPIngoingPacket packet, bool forceAsyncCall, RUDPReceiveIAsyncResult asyncResult)
		{
			asyncResult.Packet = packet;
			asyncResult.ForceAsyncCall = forceAsyncCall;
			asyncResult.SetAsCompleted(error, false);
		}

		internal void OnEndSend(RUDPSocketError error, RUDPSendIAsyncResult asyncResult)
		{
			if (asyncResult == null)
				return;

			asyncResult.SetAsCompleted(error, false);
		}

		internal void OnEndConnect(RUDPSocketError error)
		{
			RUDPConnectIAsyncResult result = null;
			Interlocked.Exchange<RUDPConnectIAsyncResult>(ref result, _asyncResultConnect);
			if (result == null)
				return;

			Interlocked.Exchange<RUDPConnectIAsyncResult>(ref _asyncResultConnect, null);

			result.Connected = (error == RUDPSocketError.Success);
			result.SetAsCompleted(error, false);
		}

		internal void OnEndAccept(RUDPSocket acceptedSocket)
		{
			RUDPAcceptIAsyncResult result = null;
			Interlocked.Exchange<RUDPAcceptIAsyncResult>(ref result, _asyncResultAccept);
			if (result == null)
				return;

			Interlocked.Exchange<RUDPAcceptIAsyncResult>(ref _asyncResultAccept, null);

			result.AcceptedSocket = acceptedSocket;
			result.ForceAsyncCall = true;
			result.SetAsCompleted(RUDPSocketError.Success, false);
		}

		internal void OnDisconnected(RUDPSocketError error)
		{
			OnEndConnect(error);

			RUDPReceiveIAsyncResult asyncResult = null;
			Interlocked.Exchange<RUDPReceiveIAsyncResult>(ref asyncResult, _asyncResultReceive);
			if (asyncResult != null)
			{
				Interlocked.Exchange<RUDPReceiveIAsyncResult>(ref _asyncResultReceive, null);
				OnEndReceive(error, null, false, asyncResult);
			}
		}

		#endregion

		#region HandleNextUserPacket

		internal void HandleNextUserPacket(bool forceAsyncCall)
		{
			if (_asyncResultReceive == null)
				return;

			RUDPReceiveIAsyncResult asyncResult;
			RUDPIngoingPacket packet;

			//---- Non reliable packets
			if (_incomingNonReliablePackets.TryDequeue(out packet))
			{
				asyncResult = Interlocked.Exchange<RUDPReceiveIAsyncResult>(ref _asyncResultReceive, null);

				if (asyncResult != null)
					OnEndReceive(RUDPSocketError.Success, packet, forceAsyncCall, asyncResult);
				else
					_incomingNonReliablePackets.Enqueue(packet);

				return;
			}

			//---- Reliable packets
			lock (_incomingPacketsLock)
			{
				if (_incomingPackets.Count < 1)
					return;

				packet = TryCleanIncomingPackets();

				if (packet == null)
					return;

				// Try to receive
				asyncResult = Interlocked.Exchange<RUDPReceiveIAsyncResult>(ref _asyncResultReceive, null);
				if (asyncResult != null)
				{
					_incomingPackets.Remove(packet.PacketId);
					Interlocked.Increment(ref _incomingPacketId);
				}
				//else throw new Exception("Sync problem");
			}

			//---- Receive
			if (asyncResult != null)
			{
				_controlWindow.OnReceiveProcessed(packet);
				OnEndReceive(RUDPSocketError.Success, packet, forceAsyncCall, asyncResult);
			}
		}

		private RUDPIngoingPacket TryCleanIncomingPackets()
		{
			while (_incomingPackets.Count > 0)
			{
				// Get the current packet
				RUDPIngoingPacket packet = (RUDPIngoingPacket)_incomingPackets[_incomingPacketId + 1];

				// No more packet
				if (packet == null)
					return null;

				// Ordered packet
				if (packet.Channel != RUDPPacketChannel.UserPacket)
				{
					_controlWindow.OnReceiveProcessed(packet);
					_incomingPackets.Remove(packet.PacketId);
					Interlocked.Increment(ref _incomingPacketId);
				}
				else return packet; // It is the next message to process.
			}

			return null;
		}

		#endregion

		#region For debug purpose only
		/*
		public PeerInformationStatus Status
		{
			get
			{
				return _status;
			}
			set
			{
				_status = value;
				BruNetStream.Trace("Update PeerInformationStatus:" + _status);
			}
		}
		*/
		#endregion

	}

	#region RecursionPoint

	public sealed class RecursionPoint
	{

		#region Variables

		private int _isExecuting = 0;

		static private readonly int TRUE = 1;
		static private readonly int FALSE = 0;

		#endregion

		#region Enter

		/// <summary>
		/// Returns true if we are already calling this method (We are in a recursion case)
		/// Returns false if we are not in recursion.
		/// </summary>
		public bool Enter()
		{
			// Avoid recursion for the same thread only (will not work with multiple threads).
			//return (Interlocked.Exchange(ref _isExecuting, Thread.CurrentThread.ManagedThreadId) != FALSE);
			return (Interlocked.Exchange(ref _isExecuting, TRUE) == TRUE);
		}

		#endregion

		#region Exit

		public void Exit()
		{
			Interlocked.Exchange(ref _isExecuting, FALSE);
		}

		#endregion

	}

	#endregion

}