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.Net;
  33 using System.Net.Security;
  34 using System.Net.Sockets;
  35 using System.ServiceModel;
  36 using System.ServiceModel.Description;
  37 using System.ServiceModel.PeerResolvers;
  38 using System.ServiceModel.Security;
  39 using System.Threading;
  40
  41 namespace System.ServiceModel.Channels
  42 {
  43
  44         // PeerDuplexChannel can be created either from PeerChannelFactory
  45         // (as IOutputChannel) or PeerChannelListener (as IInputChannel).
  46         //
  47         // PeerNode has to be created before Open() (at least at client side).
  48         // On open, it tries to resolve the nodes in the mesh (and do something
  49         // - but what?). Then registers itself to the mesh and refreshes it.
  50
  51         internal class PeerDuplexChannel : DuplexChannelBase
  52         {
  53                 enum RemotePeerStatus
  54                 {
  55                         None,
  56                         Connected,
  57                         Error,
  58                 }
  59
  60                 class RemotePeerConnection
  61                 {
  62                         public RemotePeerConnection (PeerNodeAddress address)
  63                         {
  64                                 Address = address;
  65                         }
  66
  67                         public PeerNodeAddress Address { get; private set; }
  68                         public RemotePeerStatus Status { get; set; }
  69                         public IPeerConnectorClient Channel { get; set; }
  70                 }
  71
  72                 class LocalPeerReceiver : IPeerReceiverContract
  73                 {
  74                         public LocalPeerReceiver (PeerDuplexChannel owner)
  75                         {
  76                                 this.owner = owner;
  77                         }
  78
  79                         PeerDuplexChannel owner;
  80
  81                         public void Connect (ConnectInfo connect)
  82                         {
  83                                 if (connect == null)
  84                                         throw new ArgumentNullException ("connect");
  85 try {
  86                                 var ch = OperationContext.Current.GetCallbackChannel<IPeerConnectorContract> ();
  87 // FIXME: so, this duplex channel, when created by a listener, lacks RemoteAddress to send callback. Get it from somewhere.
  88 Console.WriteLine ("FIXME FIXME:" + ((IContextChannel) ch).RemoteAddress);
  89                                 // FIXME: check and reject if inappropriate.
  90                                 ch.Welcome (new WelcomeInfo () { NodeId = connect.NodeId });
  91
  92 } catch (Exception ex) {
  93 Console.WriteLine ("Exception during Connect()");
  94 Console.WriteLine (ex);
  95 throw;
  96 }
  97
  98                         }
  99
 100                         public void Welcome (WelcomeInfo welcome)
 101                         {
 102                         }
 103
 104                         public void Refuse (RefuseInfo refuse)
 105                         {
 106                         }
 107
 108                         public void SendMessage (Message msg)
 109                         {
 110                                 owner.EnqueueMessage (msg);
 111                         }
 112                 }
 113
 114                 interface IPeerConnectorClient : IClientChannel, IPeerConnectorContract
 115                 {
 116                 }
 117
 118                 IChannelFactory<IDuplexSessionChannel> client_factory;
 119                 ChannelFactory<IPeerConnectorClient> channel_factory;
 120                 PeerTransportBindingElement binding;
 121                 PeerResolver resolver;
 122                 PeerNode node;
 123                 ServiceHost listener_host;
 124                 TcpChannelInfo info;
 125                 List<RemotePeerConnection> peers = new List<RemotePeerConnection> ();
 126
 127                 public PeerDuplexChannel (IPeerChannelManager factory, EndpointAddress address, Uri via, PeerResolver resolver)
 128                         : base ((ChannelFactoryBase) factory, address, via)
 129                 {
 130                         binding = factory.Source;
 131                         this.resolver = factory.Resolver;
 132                         info = new TcpChannelInfo (binding, factory.MessageEncoder, null); // FIXME: fill properties correctly.
 133
 134                         // It could be opened even with empty list of PeerNodeAddresses.
 135                         // So, do not create PeerNode per PeerNodeAddress, but do it with PeerNodeAddress[].
 136                         node = new PeerNodeImpl (RemoteAddress.Uri.Host, factory.Source.ListenIPAddress, factory.Source.Port);
 137                 }
 138
 139                 public PeerDuplexChannel (IPeerChannelManager listener)
 140                         : base ((ChannelListenerBase) listener)
 141                 {
 142                         binding = listener.Source;
 143                         this.resolver = listener.Resolver;
 144                         info = new TcpChannelInfo (binding, listener.MessageEncoder, null); // FIXME: fill properties correctly.
 145
 146                         node = new PeerNodeImpl (((ChannelListenerBase) listener).Uri.Host, listener.Source.ListenIPAddress, listener.Source.Port);
 147                 }
 148
 149                 public override T GetProperty<T> ()
 150                 {
 151                         if (typeof (T).IsInstanceOfType (node))
 152                                 return (T) (object) node;
 153                         return base.GetProperty<T> ();
 154                 }
 155
 156                 // DuplexChannelBase
 157
 158                 IPeerConnectorClient CreateInnerClient (PeerNodeAddress pna)
 159                 {
 160                         // FIXME: pass more setup parameters
 161                         if (channel_factory == null) {
 162                                 var binding = new NetTcpBinding ();
 163                                 binding.Security.Mode = SecurityMode.None;
 164                                 channel_factory = new ChannelFactory<IPeerConnectorClient> (binding);
 165                         }
 166
 167                         return channel_factory.CreateChannel (new EndpointAddress ("net.p2p://" + node.MeshId), pna.EndpointAddress.Uri);
 168                 }
 169
 170                 public override void Send (Message message, TimeSpan timeout)
 171                 {
 172                         ThrowIfDisposedOrNotOpen ();
 173
 174                         DateTime start = DateTime.Now;
 175
 176                         foreach (var pc in peers) {
 177                                 if (pc.Status == RemotePeerStatus.None) {
 178                                         var inner = CreateInnerClient (pc.Address);
 179                                         pc.Channel = inner;
 180                                         inner.Open (timeout - (DateTime.Now - start));
 181                                         inner.OperationTimeout = timeout - (DateTime.Now - start);
 182                                         inner.Connect (new ConnectInfo () { PeerNodeAddress = pc.Address, NodeId = (uint) node.NodeId });
 183
 184                                         // FIXME: wait for Welcome or Reject and take further action.
 185                                         throw new NotImplementedException ();
 186                                 }
 187
 188                                 pc.Channel.OperationTimeout = timeout - (DateTime.Now - start);
 189                                 pc.Channel.SendMessage (message);
 190                         }
 191                 }
 192
 193                 internal void EnqueueMessage (Message message)
 194                 {
 195 Console.WriteLine ("###########################");
 196 var mb = message.CreateBufferedCopy (0x10000);
 197 Console.WriteLine (mb.CreateMessage ());
 198 message = mb.CreateMessage ();
 199                         queue.Enqueue (message);
 200                         receive_handle.Set ();
 201                 }
 202
 203                 Queue<Message> queue = new Queue<Message> ();
 204                 AutoResetEvent receive_handle = new AutoResetEvent (false);
 205
 206                 public override Message Receive (TimeSpan timeout)
 207                 {
 208                         ThrowIfDisposedOrNotOpen ();
 209                         DateTime start = DateTime.Now;
 210
 211                         if (queue.Count > 0)
 212                                 return queue.Dequeue ();
 213                         receive_handle.WaitOne ();
 214                         return queue.Dequeue ();
 215                 }
 216
 217                 public override bool WaitForMessage (TimeSpan timeout)
 218                 {
 219                         ThrowIfDisposedOrNotOpen ();
 220
 221                         throw new NotImplementedException ();
 222                 }
 223
 224                 // CommunicationObject
 225
 226                 protected override void OnAbort ()
 227                 {
 228                         if (client_factory != null) {
 229                                 client_factory.Abort ();
 230                                 client_factory = null;
 231                         }
 232                         OnClose (TimeSpan.Zero);
 233                 }
 234
 235                 protected override void OnClose (TimeSpan timeout)
 236                 {
 237                         DateTime start = DateTime.Now;
 238                         if (client_factory != null)
 239                                 client_factory.Close (timeout - (DateTime.Now - start));
 240                         peers.Clear ();
 241                         resolver.Unregister (node.RegisteredId, timeout - (DateTime.Now - start));
 242                         node.SetOffline ();
 243                         if (listener_host != null)
 244                                 listener_host.Close (timeout - (DateTime.Now - start));
 245                         node.RegisteredId = null;
 246                 }
 247
 248
 249                 protected override void OnOpen (TimeSpan timeout)
 250                 {
 251                         DateTime start = DateTime.Now;
 252
 253                         // FIXME: supply maxAddresses
 254                         foreach (var a in resolver.Resolve (node.MeshId, 3, timeout))
 255                                 peers.Add (new RemotePeerConnection (a));
 256
 257                         // FIXME: pass more configuration
 258                         var binding = new NetTcpBinding ();
 259                         binding.Security.Mode = SecurityMode.None;
 260
 261                         int port = 0;
 262                         var rnd = new Random ();
 263                         for (int i = 0; i < 1000; i++) {
 264                                 if (DateTime.Now - start > timeout)
 265                                         throw new TimeoutException ();
 266                                 try {
 267                                         port = rnd.Next (50000, 51000);
 268                                         var t = new TcpListener (port);
 269                                         t.Start ();
 270                                         t.Stop ();
 271                                         break;
 272                                 } catch (SocketException) {
 273                                         continue;
 274                                 }
 275                         }
 276
 277                         string name = Dns.GetHostName ();
 278                         var uri = new Uri ("net.tcp://" + name + ":" + port + "/PeerChannelEndpoints/" + Guid.NewGuid ());
 279
 280                         var peer_receiver = new LocalPeerReceiver (this);
 281                         listener_host = new ServiceHost (peer_receiver);
 282                         var sba = listener_host.Description.Behaviors.Find<ServiceBehaviorAttribute> ();
 283                         sba.InstanceContextMode = InstanceContextMode.Single;
 284                         sba.IncludeExceptionDetailInFaults = true;
 285
 286                         var se = listener_host.AddServiceEndpoint (typeof (IPeerReceiverContract), binding, "net.p2p://" + node.MeshId);
 287                         se.ListenUri = uri;
 288                         listener_host.Open (timeout - (DateTime.Now - start));
 289
 290                         var nid = new Random ().Next (0, int.MaxValue);
 291                         var ea = new EndpointAddress (uri);
 292                         var pna = new PeerNodeAddress (ea, new ReadOnlyCollection<IPAddress> (Dns.GetHostEntry (name).AddressList));
 293                         node.RegisteredId = resolver.Register (node.MeshId, pna, timeout - (DateTime.Now - start));
 294                         node.NodeId = nid;
 295
 296                         // Add itself to the local list as well.
 297                         // FIXME: it might become unnecessary once it implemented new node registration from peer resolver service.
 298                         peers.Add (new RemotePeerConnection (pna));
 299
 300                         node.SetOnline ();
 301                 }
 302         }
 303 }