mcs/class/System.ServiceModel/System.ServiceModel.Channels.NetTcp/PeerDuplexChannel.cs

   1 //
   2 // PeerDuplexChannel.cs
   3 //
   4 // Author:
   5 //      Atsushi Enomoto <atsushi@ximian.com>
   6 //
   7 // Copyright (C) 2009 Novell, Inc.  http://www.novell.com
   8 //
   9 // Permission is hereby granted, free of charge, to any person obtaining
  10 // a copy of this software and associated documentation files (the
  11 // "Software"), to deal in the Software without restriction, including
  12 // without limitation the rights to use, copy, modify, merge, publish,
  13 // distribute, sublicense, and/or sell copies of the Software, and to
  14 // permit persons to whom the Software is furnished to do so, subject to
  15 // the following conditions:
  16 //
  17 // The above copyright notice and this permission notice shall be
  18 // included in all copies or substantial portions of the Software.
  19 //
  20 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  21 // EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  22 // MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  23 // NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  24 // LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  25 // OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  26 // WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  27 //
  28 using System;
  29 using System.Collections.Generic;
  30 using System.Collections.ObjectModel;
  31 using System.IO;
  32 using System.Linq;
  33 using System.Net;
  34 using System.Net.Security;
  35 using System.Net.Sockets;
  36 using System.ServiceModel;
  37 using System.ServiceModel.Channels.NetTcp;
  38 using System.ServiceModel.Description;
  39 using System.ServiceModel.PeerResolvers;
  40 using System.ServiceModel.Security;
  41 using System.Threading;
  42 using System.Xml;
  43
  44 namespace System.ServiceModel.Channels
  45 {
  46
  47         // PeerDuplexChannel can be created either from PeerChannelFactory
  48         // (as IOutputChannel) or PeerChannelListener (as IInputChannel).
  49         //
  50         // PeerNode has to be created before Open() (at least at client side).
  51         // On open, it tries to resolve the nodes in the mesh (and do something
  52         // - but what?). Then registers itself to the mesh and refreshes it.
  53
  54         internal class PeerDuplexChannel : DuplexChannelBase
  55         {
  56                 enum RemotePeerStatus
  57                 {
  58                         None,
  59                         Connected,
  60                         Error,
  61                 }
  62
  63                 class RemotePeerConnection
  64                 {
  65                         public RemotePeerConnection (PeerNodeAddress address)
  66                         {
  67                                 Address = address;
  68                         }
  69
  70                         public PeerNodeAddress Address { get; private set; }
  71                         public RemotePeerStatus Status { get; set; }
  72                         public LocalPeerReceiver Instance { get; set; }
  73                         public IPeerConnectorClient Channel { get; set; }
  74                         public ulong NodeId { get; set; }
  75                 }
  76
  77                 class LocalPeerReceiver : IPeerConnectorContract
  78                 {
  79                         List<PeerNodeAddress> connections = new List<PeerNodeAddress> ();
  80                         AutoResetEvent connect_handle = new AutoResetEvent (false);
  81                         public event Action<WelcomeInfo> WelcomeReceived;
  82
  83                         public LocalPeerReceiver (PeerDuplexChannel owner)
  84                         {
  85                                 this.owner = owner;
  86                         }
  87
  88                         PeerDuplexChannel owner;
  89
  90                         public void Connect (ConnectInfo connect)
  91                         {
  92                                 if (connect == null)
  93                                         throw new ArgumentNullException ("connect");
  94                                 var ch = OperationContext.Current.GetCallbackChannel<IPeerConnectorContract> ();
  95
  96                                 connections.Add (connect.Address);
  97                                 // FIXME: check and reject if inappropriate. For example, maximum connection exceeded.
  98                                 using (var octx = new OperationContextScope ((IContextChannel) ch)) {
  99                                         OperationContext.Current.OutgoingMessageHeaders.To = new Uri (Constants.WsaAnonymousUri);
 100                                         if (!owner.peers.Any (p => p.Address.EndpointAddress.Equals (connect.Address.EndpointAddress)))
 101                                                 owner.peers.Add (new RemotePeerConnection (connect.Address));
 102                                         ch.Welcome (new WelcomeInfo () { NodeId = owner.node.NodeId });
 103                                 }
 104                         }
 105
 106                         internal void WaitForConnectResponse (TimeSpan timeout)
 107                         {
 108                                 if (!connect_handle.WaitOne (timeout))
 109                                         throw new TimeoutException ();
 110                         }
 111
 112                         public void Disconnect (DisconnectInfo disconnect)
 113                         {
 114                                 if (disconnect == null)
 115                                         throw new ArgumentNullException ("disconnect");
 116                                 // Console.WriteLine ("DisconnectInfo.Reason: " + disconnect.Reason);
 117                                 // FIXME: handle disconnection in practice. So far I see nothing to do.
 118                         }
 119
 120                         public void Welcome (WelcomeInfo welcome)
 121                         {
 122                                 if (WelcomeReceived != null)
 123                                         WelcomeReceived (welcome);
 124                                 connect_handle.Set ();
 125                         }
 126
 127                         public void Refuse (RefuseInfo refuse)
 128                         {
 129                                 // FIXME: it should not probably actually throw an error.
 130                                 connect_handle.Set ();
 131                                 throw new InvalidOperationException ("Peer connection was refused");
 132                         }
 133
 134                         public void LinkUtility (LinkUtilityInfo linkUtility)
 135                         {
 136                                 throw new NotImplementedException ();
 137                         }
 138
 139                         public void Ping ()
 140                         {
 141                                 throw new NotImplementedException ();
 142                         }
 143
 144                         public void SendMessage (Message msg)
 145                         {
 146                                 int idx = msg.Headers.FindHeader ("PeerTo", Constants.NetPeer);
 147                                 if (idx >= 0)
 148                                         msg.Headers.To = msg.Headers.GetHeader<Uri> (idx);
 149                                 // FIXME: anything to do for PeerVia?
 150
 151                                 owner.EnqueueMessage (msg);
 152                         }
 153                 }
 154
 155                 interface IPeerConnectorClient : IClientChannel, IPeerConnectorContract
 156                 {
 157                 }
 158
 159                 IChannelFactory<IDuplexSessionChannel> client_factory;
 160                 PeerTransportBindingElement binding;
 161                 PeerResolver resolver;
 162                 PeerNode node;
 163                 ServiceHost listener_host;
 164                 TcpChannelInfo info;
 165                 List<RemotePeerConnection> peers = new List<RemotePeerConnection> ();
 166                 PeerNodeAddress local_node_address;
 167
 168                 public PeerDuplexChannel (IPeerChannelManager factory, EndpointAddress address, Uri via, PeerResolver resolver)
 169                         : base ((ChannelFactoryBase) factory, address, via)
 170                 {
 171                         binding = factory.Source;
 172                         this.resolver = factory.Resolver;
 173                         info = new TcpChannelInfo (binding, factory.MessageEncoder, null); // FIXME: fill properties correctly.
 174
 175                         // It could be opened even with empty list of PeerNodeAddresses.
 176                         // So, do not create PeerNode per PeerNodeAddress, but do it with PeerNodeAddress[].
 177                         node = new PeerNodeImpl (RemoteAddress.Uri.Host, factory.Source.ListenIPAddress, factory.Source.Port);
 178                 }
 179
 180                 public PeerDuplexChannel (IPeerChannelManager listener)
 181                         : base ((ChannelListenerBase) listener)
 182                 {
 183                         binding = listener.Source;
 184                         this.resolver = listener.Resolver;
 185                         info = new TcpChannelInfo (binding, listener.MessageEncoder, null); // FIXME: fill properties correctly.
 186
 187                         node = new PeerNodeImpl (((ChannelListenerBase) listener).Uri.Host, listener.Source.ListenIPAddress, listener.Source.Port);
 188                 }
 189
 190                 public override T GetProperty<T> ()
 191                 {
 192                         if (typeof (T).IsInstanceOfType (node))
 193                                 return (T) (object) node;
 194                         return base.GetProperty<T> ();
 195                 }
 196
 197                 // DuplexChannelBase
 198
 199                 IPeerConnectorClient CreateInnerClient (RemotePeerConnection conn)
 200                 {
 201                         conn.Instance = new LocalPeerReceiver (this);
 202                         conn.Instance.WelcomeReceived += delegate (WelcomeInfo welcome) {
 203                                 conn.NodeId = welcome.NodeId;
 204                                 // FIXME: handle referrals
 205                                 };
 206
 207                         // FIXME: pass more setup parameters
 208                         var binding = new NetTcpBinding ();
 209                         binding.Security.Mode = SecurityMode.None;
 210                         var channel_factory = new DuplexChannelFactory<IPeerConnectorClient> (conn.Instance, binding);
 211                         channel_factory.Open ();
 212
 213                         var ch = channel_factory.CreateChannel (new EndpointAddress ("net.p2p://" + node.MeshId + "/"), conn.Address.EndpointAddress.Uri);
 214                         ch.Closed += delegate {
 215                                 channel_factory.Close ();
 216                                 };
 217                         return ch;
 218                 }
 219
 220                 public override void Send (Message message, TimeSpan timeout)
 221                 {
 222                         ThrowIfDisposedOrNotOpen ();
 223
 224                         DateTime start = DateTime.UtcNow;
 225
 226                         // FIXME: give max buffer size
 227                         var mb = message.CreateBufferedCopy (0x10000);
 228
 229                         for (int i = 0; i < peers.Count; i++) {
 230                                 var pc = peers [i];
 231                                 message = mb.CreateMessage ();
 232
 233                                 if (pc.Status == RemotePeerStatus.None) {
 234                                         pc.Status = RemotePeerStatus.Error; // prepare for cases that it resulted in an error in the middle.
 235                                         var inner = CreateInnerClient (pc);
 236                                         pc.Channel = inner;
 237                                         inner.Open (timeout - (DateTime.UtcNow - start));
 238                                         inner.OperationTimeout = timeout - (DateTime.UtcNow - start);
 239                                         inner.Connect (new ConnectInfo () { Address = local_node_address, NodeId = (uint) node.NodeId });
 240                                         pc.Instance.WaitForConnectResponse (timeout - (DateTime.UtcNow - start));
 241                                         pc.Status = RemotePeerStatus.Connected;
 242                                 }
 243
 244                                 pc.Channel.OperationTimeout = timeout - (DateTime.UtcNow - start);
 245
 246                                 // see [MC-PRCH] 3.2.4.1
 247                                 if (message.Headers.MessageId == null)
 248                                         message.Headers.MessageId = new UniqueId ();
 249                                 message.Headers.Add (MessageHeader.CreateHeader ("PeerTo", Constants.NetPeer, RemoteAddress.Uri));
 250                                 message.Headers.Add (MessageHeader.CreateHeader ("PeerVia", Constants.NetPeer, RemoteAddress.Uri));
 251                                 message.Headers.Add (MessageHeader.CreateHeader ("FloodMessage", Constants.NetPeer, "PeerFlooder"));
 252                                 pc.Channel.SendMessage (message);
 253                         }
 254                 }
 255
 256                 internal void EnqueueMessage (Message message)
 257                 {
 258                         queue.Enqueue (message);
 259                         receive_handle.Set ();
 260                 }
 261
 262                 Queue<Message> queue = new Queue<Message> ();
 263                 AutoResetEvent receive_handle = new AutoResetEvent (false);
 264
 265                 public override bool TryReceive (TimeSpan timeout, out Message message)
 266                 {
 267                         ThrowIfDisposedOrNotOpen ();
 268
 269                         if (queue.Count > 0 || receive_handle.WaitOne (timeout)) {
 270                                 message = queue.Dequeue ();
 271                                 return message == null;
 272                         } else {
 273                                 message = null;
 274                                 return false;
 275                         }
 276                 }
 277
 278                 public override bool WaitForMessage (TimeSpan timeout)
 279                 {
 280                         ThrowIfDisposedOrNotOpen ();
 281
 282                         throw new NotImplementedException ();
 283                 }
 284
 285                 // CommunicationObject
 286
 287                 protected override void OnAbort ()
 288                 {
 289                         if (client_factory != null) {
 290                                 client_factory.Abort ();
 291                                 client_factory = null;
 292                         }
 293                         OnClose (TimeSpan.Zero);
 294                 }
 295
 296                 protected override void OnClose (TimeSpan timeout)
 297                 {
 298                         DateTime start = DateTime.UtcNow;
 299                         if (client_factory != null)
 300                                 client_factory.Close (timeout - (DateTime.UtcNow - start));
 301                         peers.Clear ();
 302                         resolver.Unregister (node.RegisteredId, timeout - (DateTime.UtcNow - start));
 303                         node.SetOffline ();
 304                         if (listener_host != null)
 305                                 listener_host.Close (timeout - (DateTime.UtcNow - start));
 306                         node.RegisteredId = null;
 307                 }
 308
 309
 310                 protected override void OnOpen (TimeSpan timeout)
 311                 {
 312                         DateTime start = DateTime.UtcNow;
 313
 314                         // FIXME: supply maxAddresses
 315                         foreach (var a in resolver.Resolve (node.MeshId, 3, timeout))
 316                                 peers.Add (new RemotePeerConnection (a));
 317
 318                         // FIXME: pass more configuration
 319                         var binding = new NetTcpBinding ();
 320                         binding.Security.Mode = SecurityMode.None;
 321
 322                         int port = 0;
 323                         var rnd = new Random ();
 324                         for (int i = 0; i < 1000; i++) {
 325                                 if (DateTime.UtcNow - start > timeout)
 326                                         throw new TimeoutException ();
 327                                 try {
 328                                         port = rnd.Next (50000, 51000);
 329                                         var t = new TcpListener (port);
 330                                         t.Start ();
 331                                         t.Stop ();
 332                                         break;
 333                                 } catch (SocketException) {
 334                                         continue;
 335                                 }
 336                         }
 337
 338                         string name = Dns.GetHostName ();
 339                         var uri = new Uri ("net.tcp://" + name + ":" + port + "/PeerChannelEndpoints/" + Guid.NewGuid ());
 340
 341                         var peer_receiver = new LocalPeerReceiver (this);
 342                         listener_host = new ServiceHost (peer_receiver);
 343                         var sba = listener_host.Description.Behaviors.Find<ServiceBehaviorAttribute> ();
 344                         sba.InstanceContextMode = InstanceContextMode.Single;
 345                         sba.IncludeExceptionDetailInFaults = true;
 346
 347                         var se = listener_host.AddServiceEndpoint (typeof (IPeerConnectorContract).FullName, binding, "net.p2p://" + node.MeshId + "/");
 348                         se.ListenUri = uri;
 349
 350                         // FIXME: remove debugging code
 351                         listener_host.UnknownMessageReceived += delegate (object obj, UnknownMessageReceivedEventArgs earg) { Console.WriteLine ("%%%%% UNKOWN MESSAGE " + earg.Message); };
 352
 353                         listener_host.Open (timeout - (DateTime.UtcNow - start));
 354
 355                         var nid = (ulong) new Random ().Next (0, int.MaxValue);
 356                         var ea = new EndpointAddress (uri);
 357                         var pna = new PeerNodeAddress (ea, new ReadOnlyCollection<IPAddress> (Dns.GetHostEntry (name).AddressList));
 358                         local_node_address = pna;
 359                         node.RegisteredId = resolver.Register (node.MeshId, pna, timeout - (DateTime.UtcNow - start));
 360                         node.NodeId = nid;
 361
 362                         // Add itself to the local list as well.
 363                         // FIXME: it might become unnecessary once it implemented new node registration from peer resolver service.
 364                         peers.Add (new RemotePeerConnection (pna));
 365
 366                         node.SetOnline ();
 367                 }
 368         }
 369 }