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