//
using System;
using System.Collections.Generic;
+using System.Collections.ObjectModel;
using System.IO;
+using System.Linq;
using System.Net;
using System.Net.Security;
+using System.Net.Sockets;
using System.ServiceModel;
using System.ServiceModel.Description;
+using System.ServiceModel.PeerResolvers;
using System.ServiceModel.Security;
using System.Threading;
+using System.Xml;
namespace System.ServiceModel.Channels
{
+
+ // PeerDuplexChannel can be created either from PeerChannelFactory
+ // (as IOutputChannel) or PeerChannelListener (as IInputChannel).
+ //
+ // PeerNode has to be created before Open() (at least at client side).
+ // On open, it tries to resolve the nodes in the mesh (and do something
+ // - but what?). Then registers itself to the mesh and refreshes it.
+
internal class PeerDuplexChannel : DuplexChannelBase
{
- PeerChannelFactory<IDuplexChannel> factory;
- PeerChannelListener<IDuplexChannel> listener;
- EndpointAddress local_address;
+ enum RemotePeerStatus
+ {
+ None,
+ Connected,
+ Error,
+ }
+
+ class RemotePeerConnection
+ {
+ public RemotePeerConnection (PeerNodeAddress address)
+ {
+ Address = address;
+ }
+
+ public PeerNodeAddress Address { get; private set; }
+ public RemotePeerStatus Status { get; set; }
+ public LocalPeerReceiver Instance { get; set; }
+ public IPeerConnectorClient Channel { get; set; }
+ public ulong NodeId { get; set; }
+ }
+
+ class LocalPeerReceiver : IPeerConnectorContract
+ {
+ List<PeerNodeAddress> connections = new List<PeerNodeAddress> ();
+ AutoResetEvent connect_handle = new AutoResetEvent (false);
+ public event Action<WelcomeInfo> WelcomeReceived;
+
+ public LocalPeerReceiver (PeerDuplexChannel owner)
+ {
+ this.owner = owner;
+ }
+
+ PeerDuplexChannel owner;
+
+ public void Connect (ConnectInfo connect)
+ {
+ if (connect == null)
+ throw new ArgumentNullException ("connect");
+ var ch = OperationContext.Current.GetCallbackChannel<IPeerConnectorContract> ();
+
+ connections.Add (connect.Address);
+ // FIXME: check and reject if inappropriate. For example, maximum connection exceeded.
+ using (var octx = new OperationContextScope ((IContextChannel) ch)) {
+ OperationContext.Current.OutgoingMessageHeaders.To = new Uri (Constants.WsaAnonymousUri);
+ if (!owner.peers.Any (p => p.Address.EndpointAddress.Equals (connect.Address.EndpointAddress)))
+ owner.peers.Add (new RemotePeerConnection (connect.Address));
+ ch.Welcome (new WelcomeInfo () { NodeId = owner.node.NodeId });
+ }
+ }
+
+ internal void WaitForConnectResponse (TimeSpan timeout)
+ {
+ if (!connect_handle.WaitOne (timeout))
+ throw new TimeoutException ();
+ }
+
+ public void Disconnect (DisconnectInfo disconnect)
+ {
+ if (disconnect == null)
+ throw new ArgumentNullException ("disconnect");
+ // Console.WriteLine ("DisconnectInfo.Reason: " + disconnect.Reason);
+ // FIXME: handle disconnection in practice. So far I see nothing to do.
+ }
+
+ public void Welcome (WelcomeInfo welcome)
+ {
+ if (WelcomeReceived != null)
+ WelcomeReceived (welcome);
+ connect_handle.Set ();
+ }
+
+ public void Refuse (RefuseInfo refuse)
+ {
+ // FIXME: it should not probably actually throw an error.
+ connect_handle.Set ();
+ throw new InvalidOperationException ("Peer connection was refused");
+ }
+
+ public void LinkUtility (LinkUtilityInfo linkUtility)
+ {
+ throw new NotImplementedException ();
+ }
+
+ public void Ping ()
+ {
+ throw new NotImplementedException ();
+ }
+
+ public void SendMessage (Message msg)
+ {
+ int idx = msg.Headers.FindHeader ("PeerTo", Constants.NetPeer);
+ if (idx >= 0)
+ msg.Headers.To = msg.Headers.GetHeader<Uri> (idx);
+ // FIXME: anything to do for PeerVia?
+
+ owner.EnqueueMessage (msg);
+ }
+ }
+
+ interface IPeerConnectorClient : IClientChannel, IPeerConnectorContract
+ {
+ }
+
+ IChannelFactory<IDuplexSessionChannel> client_factory;
+ PeerTransportBindingElement binding;
PeerResolver resolver;
PeerNode node;
+ ServiceHost listener_host;
+ TcpChannelInfo info;
+ List<RemotePeerConnection> peers = new List<RemotePeerConnection> ();
+ PeerNodeAddress local_node_address;
- public PeerDuplexChannel (PeerChannelFactory<IDuplexChannel> factory, EndpointAddress address, Uri via, PeerResolver resolver)
- : base (factory, address, via)
+ public PeerDuplexChannel (IPeerChannelManager factory, EndpointAddress address, Uri via, PeerResolver resolver)
+ : base ((ChannelFactoryBase) factory, address, via)
{
- this.factory = factory;
- this.resolver = resolver;
+ binding = factory.Source;
+ this.resolver = factory.Resolver;
+ info = new TcpChannelInfo (binding, factory.MessageEncoder, null); // FIXME: fill properties correctly.
- string mesh = RemoteAddress.Uri.Host;
// It could be opened even with empty list of PeerNodeAddresses.
// So, do not create PeerNode per PeerNodeAddress, but do it with PeerNodeAddress[].
- node = new PeerNodeImpl (resolver, mesh, factory.Source.Port);
+ node = new PeerNodeImpl (RemoteAddress.Uri.Host, factory.Source.ListenIPAddress, factory.Source.Port);
}
- // FIXME: receive local_address too
- public PeerDuplexChannel (PeerChannelListener<IDuplexChannel> listener)
- : base (listener)
+ public PeerDuplexChannel (IPeerChannelManager listener)
+ : base ((ChannelListenerBase) listener)
{
- this.listener = listener;
+ binding = listener.Source;
+ this.resolver = listener.Resolver;
+ info = new TcpChannelInfo (binding, listener.MessageEncoder, null); // FIXME: fill properties correctly.
- // FIXME: set resolver and node.
- }
-
- public override EndpointAddress LocalAddress {
- get { return local_address; }
+ node = new PeerNodeImpl (((ChannelListenerBase) listener).Uri.Host, listener.Source.ListenIPAddress, listener.Source.Port);
}
public override T GetProperty<T> ()
// DuplexChannelBase
- public override IAsyncResult BeginSend (Message message, TimeSpan timeout, AsyncCallback callback, object state)
+ IPeerConnectorClient CreateInnerClient (RemotePeerConnection conn)
{
- throw new NotImplementedException ();
- }
+ conn.Instance = new LocalPeerReceiver (this);
+ conn.Instance.WelcomeReceived += delegate (WelcomeInfo welcome) {
+ conn.NodeId = welcome.NodeId;
+ // FIXME: handle referrals
+ };
- public override void EndSend (IAsyncResult result)
- {
- throw new NotImplementedException ();
- }
+ // FIXME: pass more setup parameters
+ var binding = new NetTcpBinding ();
+ binding.Security.Mode = SecurityMode.None;
+ var channel_factory = new DuplexChannelFactory<IPeerConnectorClient> (conn.Instance, binding);
+ channel_factory.Open ();
- public override IAsyncResult BeginReceive (TimeSpan timeout, AsyncCallback callback, object state)
- {
- throw new NotImplementedException ();
+ var ch = channel_factory.CreateChannel (new EndpointAddress ("net.p2p://" + node.MeshId + "/"), conn.Address.EndpointAddress.Uri);
+ ch.Closed += delegate {
+ channel_factory.Close ();
+ };
+ return ch;
}
- public override IAsyncResult BeginTryReceive (TimeSpan timeout, AsyncCallback callback, object state)
+ public override void Send (Message message, TimeSpan timeout)
{
- throw new NotImplementedException ();
- }
+ ThrowIfDisposedOrNotOpen ();
- public override IAsyncResult BeginWaitForMessage (TimeSpan timeout, AsyncCallback callback, object state)
- {
- throw new NotImplementedException ();
+ DateTime start = DateTime.Now;
+
+ // FIXME: give max buffer size
+ var mb = message.CreateBufferedCopy (0x10000);
+
+ for (int i = 0; i < peers.Count; i++) {
+ var pc = peers [i];
+ message = mb.CreateMessage ();
+
+ if (pc.Status == RemotePeerStatus.None) {
+ pc.Status = RemotePeerStatus.Error; // prepare for cases that it resulted in an error in the middle.
+ var inner = CreateInnerClient (pc);
+ pc.Channel = inner;
+ inner.Open (timeout - (DateTime.Now - start));
+ inner.OperationTimeout = timeout - (DateTime.Now - start);
+ inner.Connect (new ConnectInfo () { Address = local_node_address, NodeId = (uint) node.NodeId });
+ pc.Instance.WaitForConnectResponse (timeout - (DateTime.Now - start));
+ pc.Status = RemotePeerStatus.Connected;
+ }
+
+ pc.Channel.OperationTimeout = timeout - (DateTime.Now - start);
+
+ // see [MC-PRCH] 3.2.4.1
+ if (message.Headers.MessageId == null)
+ message.Headers.MessageId = new UniqueId ();
+ message.Headers.Add (MessageHeader.CreateHeader ("PeerTo", Constants.NetPeer, RemoteAddress.Uri));
+ message.Headers.Add (MessageHeader.CreateHeader ("PeerVia", Constants.NetPeer, RemoteAddress.Uri));
+ message.Headers.Add (MessageHeader.CreateHeader ("FloodMessage", Constants.NetPeer, "PeerFlooder"));
+ pc.Channel.SendMessage (message);
+ }
}
- public override Message EndReceive (IAsyncResult result)
+ internal void EnqueueMessage (Message message)
{
- throw new NotImplementedException ();
+ queue.Enqueue (message);
+ receive_handle.Set ();
}
- public override bool EndTryReceive (IAsyncResult result, out Message message)
+ Queue<Message> queue = new Queue<Message> ();
+ AutoResetEvent receive_handle = new AutoResetEvent (false);
+
+ public override bool TryReceive (TimeSpan timeout, out Message message)
{
- throw new NotImplementedException ();
+ ThrowIfDisposedOrNotOpen ();
+ DateTime start = DateTime.Now;
+
+ if (queue.Count > 0 || receive_handle.WaitOne (timeout)) {
+ message = queue.Dequeue ();
+ return message == null;
+ } else {
+ message = null;
+ return false;
+ }
}
- public override bool EndWaitForMessage (IAsyncResult result)
+ public override bool WaitForMessage (TimeSpan timeout)
{
+ ThrowIfDisposedOrNotOpen ();
+
throw new NotImplementedException ();
}
// CommunicationObject
- [MonoTODO]
protected override void OnAbort ()
{
- throw new NotImplementedException ();
- }
-
- [MonoTODO]
- protected override IAsyncResult OnBeginClose (TimeSpan timeout,
- AsyncCallback callback, object state)
- {
- throw new NotImplementedException ();
- }
-
- [MonoTODO]
- protected override IAsyncResult OnBeginOpen (TimeSpan timeout,
- AsyncCallback callback, object state)
- {
- throw new NotImplementedException ();
+ if (client_factory != null) {
+ client_factory.Abort ();
+ client_factory = null;
+ }
+ OnClose (TimeSpan.Zero);
}
- [MonoTODO]
protected override void OnClose (TimeSpan timeout)
{
- throw new NotImplementedException ();
- }
-
- [MonoTODO]
- protected override void OnEndClose (IAsyncResult result)
- {
- throw new NotImplementedException ();
+ DateTime start = DateTime.Now;
+ if (client_factory != null)
+ client_factory.Close (timeout - (DateTime.Now - start));
+ peers.Clear ();
+ resolver.Unregister (node.RegisteredId, timeout - (DateTime.Now - start));
+ node.SetOffline ();
+ if (listener_host != null)
+ listener_host.Close (timeout - (DateTime.Now - start));
+ node.RegisteredId = null;
}
- [MonoTODO]
- protected override void OnEndOpen (IAsyncResult result)
- {
- throw new NotImplementedException ();
- }
- // At some stage I should unify this class with PeerOutputChannel (and probably PeerInputChannel). Too much duplicate.
protected override void OnOpen (TimeSpan timeout)
{
- node.Open (timeout);
+ DateTime start = DateTime.Now;
+
+ // FIXME: supply maxAddresses
+ foreach (var a in resolver.Resolve (node.MeshId, 3, timeout))
+ peers.Add (new RemotePeerConnection (a));
+
+ // FIXME: pass more configuration
+ var binding = new NetTcpBinding ();
+ binding.Security.Mode = SecurityMode.None;
+
+ int port = 0;
+ var rnd = new Random ();
+ for (int i = 0; i < 1000; i++) {
+ if (DateTime.Now - start > timeout)
+ throw new TimeoutException ();
+ try {
+ port = rnd.Next (50000, 51000);
+ var t = new TcpListener (port);
+ t.Start ();
+ t.Stop ();
+ break;
+ } catch (SocketException) {
+ continue;
+ }
+ }
+
+ string name = Dns.GetHostName ();
+ var uri = new Uri ("net.tcp://" + name + ":" + port + "/PeerChannelEndpoints/" + Guid.NewGuid ());
+
+ var peer_receiver = new LocalPeerReceiver (this);
+ listener_host = new ServiceHost (peer_receiver);
+ var sba = listener_host.Description.Behaviors.Find<ServiceBehaviorAttribute> ();
+ sba.InstanceContextMode = InstanceContextMode.Single;
+ sba.IncludeExceptionDetailInFaults = true;
+
+ var se = listener_host.AddServiceEndpoint (typeof (IPeerConnectorContract).FullName, binding, "net.p2p://" + node.MeshId + "/");
+ se.ListenUri = uri;
+
+ // FIXME: remove debugging code
+ listener_host.UnknownMessageReceived += delegate (object obj, UnknownMessageReceivedEventArgs earg) { Console.WriteLine ("%%%%% UNKOWN MESSAGE " + earg.Message); };
+
+ listener_host.Open (timeout - (DateTime.Now - start));
+
+ var nid = (ulong) new Random ().Next (0, int.MaxValue);
+ var ea = new EndpointAddress (uri);
+ var pna = new PeerNodeAddress (ea, new ReadOnlyCollection<IPAddress> (Dns.GetHostEntry (name).AddressList));
+ local_node_address = pna;
+ node.RegisteredId = resolver.Register (node.MeshId, pna, timeout - (DateTime.Now - start));
+ node.NodeId = nid;
+
+ // Add itself to the local list as well.
+ // FIXME: it might become unnecessary once it implemented new node registration from peer resolver service.
+ peers.Add (new RemotePeerConnection (pna));
+
+ node.SetOnline ();
}
}
}
projects / mono.git / blobdiff