?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
2009-07-17 Atsushi Enomoto <atsushi@ximian.com>
[mcs.git] / class / System.ServiceModel / System.ServiceModel.Channels / TcpDuplexSessionChannel.cs
blob1d597894eb9196e37c9e8b280cd3c9028f9489dc
1 //
2 // TcpDuplexSessionChannel.cs
3 //
4 // Author:
5 // Marcos Cobena (marcoscobena@gmail.com)
6 // Atsushi Enomoto <atsushi@ximian.com>
7 //
8 // Copyright 2007 Marcos Cobena (http://www.youcannoteatbits.org/)
9 //
10 // Copyright (C) 2009 Novell, Inc (http://www.novell.com)
11 //
12 // Permission is hereby granted, free of charge, to any person obtaining
13 // a copy of this software and associated documentation files (the
14 // "Software"), to deal in the Software without restriction, including
15 // without limitation the rights to use, copy, modify, merge, publish,
16 // distribute, sublicense, and/or sell copies of the Software, and to
17 // permit persons to whom the Software is furnished to do so, subject to
18 // the following conditions:
19 //
20 // The above copyright notice and this permission notice shall be
21 // included in all copies or substantial portions of the Software.
22 //
23 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
24 // EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
25 // MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
26 // NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
27 // LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
28 // OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
29 // WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
30 //
31
32 using System;
33 using System.Collections.Generic;
34 using System.IO;
35 using System.Net;
36 using System.Net.Sockets;
37 using System.Runtime.Serialization;
38 using System.Runtime.Serialization.Formatters.Binary;
39 using System.ServiceModel.Channels;
40 using System.Text;
41 using System.Threading;
42 using System.Xml;
43
44 namespace System.ServiceModel.Channels
45 {
46 internal class TcpDuplexSessionChannel : DuplexChannelBase, IDuplexSessionChannel
47 {
48 class TcpDuplexSession : DuplexSessionBase
49 {
50 TcpDuplexSessionChannel owner;
51
52 internal TcpDuplexSession (TcpDuplexSessionChannel owner)
53 {
54 this.owner = owner;
55 }
56
57 public override TimeSpan DefaultCloseTimeout {
58 get { return owner.DefaultCloseTimeout; }
59 }
60
61 public override void Close (TimeSpan timeout)
62 {
63 owner.DiscardSession ();
64 }
65 }
66
67 TcpChannelInfo info;
68 TcpClient client;
69 bool is_service_side;
70 EndpointAddress local_address;
71 TcpBinaryFrameManager frame;
72 TcpDuplexSession session; // do not use this directly. Use Session instead.
73
74 public TcpDuplexSessionChannel (ChannelFactoryBase factory, TcpChannelInfo info, EndpointAddress address, Uri via)
75 : base (factory, address, via)
76 {
77 is_service_side = false;
78 this.info = info;
79 }
80
81 public TcpDuplexSessionChannel (ChannelListenerBase listener, TcpChannelInfo info, TcpClient client)
82 : base (listener)
83 {
84 is_service_side = true;
85 this.client = client;
86 this.info = info;
87 }
88
89 public MessageEncoder Encoder {
90 get { return info.MessageEncoder; }
91 }
92
93 public override EndpointAddress LocalAddress {
94 get { return local_address; }
95 }
96
97 public IDuplexSession Session {
98 get {
99 if (session == null)
100 session = new TcpDuplexSession (this);
101 return session;
102 }
103 }
104
105 void DiscardSession ()
106 {
107 frame.ProcessEndRecordInitiator ();
108 session = null;
109 }
110
111 public override void Send (Message message)
112 {
113 Send (message, DefaultSendTimeout);
114 }
115
116 public override void Send (Message message, TimeSpan timeout)
117 {
118 if (timeout <= TimeSpan.Zero)
119 throw new ArgumentException (String.Format ("Timeout value must be positive value. It was {0}", timeout));
120
121 if (!is_service_side) {
122 if (message.Headers.To == null)
123 message.Headers.To = RemoteAddress.Uri;
124 if (message.Headers.ReplyTo == null)
125 message.Headers.ReplyTo = new EndpointAddress (Constants.WsaAnonymousUri);
126 } else {
127 if (message.Headers.RelatesTo == null)
128 message.Headers.RelatesTo = OperationContext.Current.IncomingMessageHeaders.MessageId;
129 }
130
131 client.SendTimeout = (int) timeout.TotalMilliseconds;
132 frame.WriteSizedMessage (message);
133 // FIXME: should EndRecord be sent here?
134 //if (is_service_side && client.Available > 0)
135 // frame.ProcessEndRecordRecipient ();
136 }
137
138 public override Message Receive ()
139 {
140 return Receive (DefaultReceiveTimeout);
141 }
142
143 public override Message Receive (TimeSpan timeout)
144 {
145 if (timeout <= TimeSpan.Zero)
146 throw new ArgumentException (String.Format ("Timeout value must be positive value. It was {0}", timeout));
147 client.ReceiveTimeout = (int) timeout.TotalMilliseconds;
148 return frame.ReadSizedMessage ();
149 }
150
151 public override bool TryReceive (TimeSpan timeout, out Message message)
152 {
153 try {
154 DateTime start = DateTime.Now;
155 message = Receive (timeout);
156 if (message != null)
157 return true;
158 // received EndRecord, so close the session and return false instead.
159 // (Closing channel here might not be a good idea, but right now I have no better way.)
160 Close (timeout - (DateTime.Now - start));
161 return false;
162 } catch (TimeoutException) {
163 message = null;
164 return false;
165 }
166 }
167
168 public override bool WaitForMessage (TimeSpan timeout)
169 {
170 if (client.Available > 0)
171 return true;
172
173 DateTime start = DateTime.Now;
174 do {
175 Thread.Sleep (50);
176 if (client.Available > 0)
177 return true;
178 } while (DateTime.Now - start < timeout);
179 return false;
180 }
181
182 // CommunicationObject
183
184 [MonoTODO]
185 protected override void OnAbort ()
186 {
187 Close (TimeSpan.FromTicks (1));
188 }
189
190 [MonoTODO]
191 protected override IAsyncResult OnBeginClose (TimeSpan timeout,
192 AsyncCallback callback, object state)
193 {
194 throw new NotImplementedException ();
195 }
196
197 [MonoTODO]
198 protected override IAsyncResult OnBeginOpen (TimeSpan timeout,
199 AsyncCallback callback, object state)
200 {
201 throw new NotImplementedException ();
202 }
203
204 [MonoTODO]
205 protected override void OnClose (TimeSpan timeout)
206 {
207 if (!is_service_side)
208 if (session != null)
209 session.Close (timeout);
210
211 if (client != null)
212 client.Close ();
213 }
214
215 [MonoTODO]
216 protected override void OnEndClose (IAsyncResult result)
217 {
218 throw new NotImplementedException ();
219 }
220
221 [MonoTODO]
222 protected override void OnEndOpen (IAsyncResult result)
223 {
224 throw new NotImplementedException ();
225 }
226
227 [MonoTODO]
228 protected override void OnOpen (TimeSpan timeout)
229 {
230 if (! is_service_side) {
231 int explicitPort = RemoteAddress.Uri.Port;
232 client = new TcpClient (RemoteAddress.Uri.Host, explicitPort <= 0 ? TcpTransportBindingElement.DefaultPort : explicitPort);
233 //RemoteAddress.Uri.Port);
234
235 NetworkStream ns = client.GetStream ();
236 frame = new TcpBinaryFrameManager (TcpBinaryFrameManager.DuplexMode, ns, is_service_side) {
237 Encoder = this.Encoder,
238 Via = RemoteAddress.Uri };
239 frame.ProcessPreambleInitiator ();
240 frame.ProcessPreambleAckInitiator ();
241 } else {
242 // server side
243 Stream s = client.GetStream ();
244
245 frame = new TcpBinaryFrameManager (TcpBinaryFrameManager.DuplexMode, s, is_service_side) { Encoder = this.Encoder };
246
247 // FIXME: use retrieved record properties in the request processing.
248
249 frame.ProcessPreambleRecipient ();
250 frame.ProcessPreambleAckRecipient ();
251 }
252 }
253
254 class MyBinaryWriter : BinaryWriter
255 {
256 public MyBinaryWriter (Stream s)
257 : base (s)
258 {
259 }
260
261 public void WriteBytes (byte [] bytes)
262 {
263 Write7BitEncodedInt (bytes.Length);
264 Write (bytes);
265 }
266 }
267 }
268
269 // seealso: [MC-NMF] Windows Protocol document.
270 class TcpBinaryFrameManager
271 {
272 class MyBinaryReader : BinaryReader
273 {
274 public MyBinaryReader (Stream s)
275 : base (s)
276 {
277 }
278
279 public int ReadVariableInt ()
280 {
281 return Read7BitEncodedInt ();
282 }
283 }
284
285 class MyBinaryWriter : BinaryWriter
286 {
287 public MyBinaryWriter (Stream s)
288 : base (s)
289 {
290 }
291
292 public void WriteVariableInt (int value)
293 {
294 Write7BitEncodedInt (value);
295 }
296
297 public int GetSizeOfLength (int value)
298 {
299 int x = 0;
300 do {
301 value /= 0x100;
302 x++;
303 } while (value != 0);
304 return x;
305 }
306 }
307
308 class MyXmlBinaryWriterSession : XmlBinaryWriterSession
309 {
310 public override bool TryAdd (XmlDictionaryString value, out int key)
311 {
312 if (!base.TryAdd (value, out key))
313 return false;
314 List.Add (value);
315 return true;
316 }
317
318 public List<XmlDictionaryString> List = new List<XmlDictionaryString> ();
319 }
320
321 public const byte VersionRecord = 0;
322 public const byte ModeRecord = 1;
323 public const byte ViaRecord = 2;
324 public const byte KnownEncodingRecord = 3;
325 public const byte ExtendingEncodingRecord = 4;
326 public const byte UnsizedEnvelopeRecord = 5;
327 public const byte SizedEnvelopeRecord = 6;
328 public const byte EndRecord = 7;
329 public const byte FaultRecord = 8;
330 public const byte UpgradeRequestRecord = 9;
331 public const byte UpgradeResponseRecord = 0xA;
332 public const byte PreambleAckRecord = 0xB;
333 public const byte PreambleEndRecord = 0xC;
334
335 public const byte UnsizedMessageTerminator = 0;
336 public const byte SingletonUnsizedMode = 1;
337 public const byte DuplexMode = 2;
338 public const byte SimplexMode = 3;
339 public const byte SingletonSizedMode = 4;
340 MyBinaryReader reader;
341 MyBinaryWriter writer;
342
343 public TcpBinaryFrameManager (int mode, Stream s, bool isServiceSide)
344 {
345 this.mode = mode;
346 this.s = s;
347 this.is_service_side = isServiceSide;
348 reader = new MyBinaryReader (s);
349 ResetWriteBuffer ();
350
351 EncodingRecord = 8; // FIXME: it should depend on mode.
352 }
353
354 Stream s;
355 MemoryStream buffer;
356 bool is_service_side;
357
358 int mode;
359
360 public byte EncodingRecord { get; set; }
361
362 public Uri Via { get; set; }
363
364 public MessageEncoder Encoder { get; set; }
365
366 void ResetWriteBuffer ()
367 {
368 this.buffer = new MemoryStream ();
369 writer = new MyBinaryWriter (buffer);
370 }
371
372 public byte [] ReadSizedChunk ()
373 {
374 int length = reader.ReadVariableInt ();
375
376 if (length > 65536)
377 throw new InvalidOperationException ("The message is too large.");
378
379 byte [] buffer = new byte [length];
380 for (int readSize = 0; readSize < length; )
381 readSize += reader.Read (buffer, readSize, length - readSize);
382 return buffer;
383 }
384
385 public void WriteSizedChunk (byte [] data)
386 {
387 writer.WriteVariableInt (data.Length);
388 writer.Write (data, 0, data.Length);
389 }
390
391 public void ProcessPreambleInitiator ()
392 {
393 ResetWriteBuffer ();
394
395 buffer.WriteByte (VersionRecord);
396 buffer.WriteByte (1);
397 buffer.WriteByte (0);
398 buffer.WriteByte (ModeRecord);
399 buffer.WriteByte ((byte) mode);
400 buffer.WriteByte (ViaRecord);
401 writer.Write (Via.ToString ());
402 buffer.WriteByte (KnownEncodingRecord); // FIXME
403 buffer.WriteByte ((byte) EncodingRecord);
404 buffer.WriteByte (PreambleEndRecord);
405 buffer.Flush ();
406 s.Write (buffer.GetBuffer (), 0, (int) buffer.Position);
407 s.Flush ();
408 }
409
410 public void ProcessPreambleAckInitiator ()
411 {
412 int b = s.ReadByte ();
413 switch (b) {
414 case PreambleAckRecord:
415 return; // success
416 case FaultRecord:
417 throw new FaultException (reader.ReadString ());
418 default:
419 throw new ProtocolException (String.Format ("Preamble Ack Record is expected, got {0:X}", b));
420 }
421 }
422
423 public void ProcessPreambleAckRecipient ()
424 {
425 s.WriteByte (PreambleAckRecord);
426 }
427
428 public void ProcessPreambleRecipient ()
429 {
430 bool preambleEnd = false;
431 while (!preambleEnd) {
432 int b = s.ReadByte ();
433 switch (b) {
434 case VersionRecord:
435 if (s.ReadByte () != 1)
436 throw new ProtocolException ("Major version must be 1");
437 if (s.ReadByte () != 0)
438 throw new ProtocolException ("Minor version must be 0");
439 break;
440 case ModeRecord:
441 if (s.ReadByte () != mode)
442 throw new ProtocolException (String.Format ("Duplex mode is expected to be {0:X}", mode));
443 break;
444 case ViaRecord:
445 Via = new Uri (reader.ReadString ());
446 break;
447 case KnownEncodingRecord:
448 EncodingRecord = (byte) s.ReadByte ();
449 break;
450 case ExtendingEncodingRecord:
451 throw new NotImplementedException ("ExtendingEncodingRecord");
452 case UpgradeRequestRecord:
453 throw new NotImplementedException ("UpgradeRequetRecord");
454 case UpgradeResponseRecord:
455 throw new NotImplementedException ("UpgradeResponseRecord");
456 case PreambleEndRecord:
457 preambleEnd = true;
458 break;
459 default:
460 throw new ProtocolException (String.Format ("Unexpected record type {0:X2}", b));
461 }
462 }
463 }
464
465 XmlBinaryReaderSession reader_session;
466 int reader_session_items;
467
468 public Message ReadSizedMessage ()
469 {
470 // FIXME: implement full [MC-NMF].
471
472 var packetType = s.ReadByte ();
473 if (packetType == EndRecord)
474 return null;
475 if (packetType != SizedEnvelopeRecord)
476 throw new NotImplementedException (String.Format ("Packet type {0:X} is not implemented", packetType));
477
478 byte [] buffer = ReadSizedChunk ();
479
480 var ms = new MemoryStream (buffer, 0, buffer.Length);
481
482 // FIXME: turned out that it could be either in-band dictionary ([MC-NBFSE]), or a mere xml body ([MC-NBFS]).
483 if (EncodingRecord != 8)
484 throw new NotImplementedException (String.Format ("Message encoding {0:X} is not implemented yet", EncodingRecord));
485
486 // Encoding type 8:
487 // the returned buffer consists of a serialized reader
488 // session and the binary xml body.
489
490 var session = reader_session ?? new XmlBinaryReaderSession ();
491 reader_session = session;
492 byte [] rsbuf = new TcpBinaryFrameManager (0, ms, is_service_side).ReadSizedChunk ();
493 using (var rms = new MemoryStream (rsbuf, 0, rsbuf.Length)) {
494 var rbr = new BinaryReader (rms, Encoding.UTF8);
495 while (rms.Position < rms.Length)
496 session.Add (reader_session_items++, rbr.ReadString ());
497 }
498 var benc = Encoder as BinaryMessageEncoder;
499 if (benc != null)
500 benc.CurrentReaderSession = session;
501
502 // FIXME: supply maxSizeOfHeaders.
503 Message msg = Encoder.ReadMessage (ms, 0x10000);
504 if (benc != null)
505 benc.CurrentReaderSession = null;
506
507 return msg;
508 }
509
510 // FIXME: support timeout
511 public Message ReadUnsizedMessage (TimeSpan timeout)
512 {
513 var packetType = s.ReadByte ();
514 if (packetType == EndRecord)
515 return null;
516 if (packetType != UnsizedEnvelopeRecord)
517 throw new NotImplementedException (String.Format ("Packet type {0:X} is not implemented", packetType));
518
519 // FIXME: turned out that it could be either in-band dictionary ([MC-NBFSE]), or a mere xml body ([MC-NBFS]).
520 if (EncodingRecord != 8)
521 throw new NotImplementedException (String.Format ("Message encoding {0:X} is not implemented yet", EncodingRecord));
522
523 // Encoding type 8:
524 // the returned buffer consists of a serialized reader
525 // session and the binary xml body.
526
527 var session = reader_session ?? new XmlBinaryReaderSession ();
528 reader_session = session;
529 byte [] rsbuf = new TcpBinaryFrameManager (0, s, is_service_side).ReadSizedChunk ();
530
531 using (var rms = new MemoryStream (rsbuf, 0, rsbuf.Length)) {
532 var rbr = new BinaryReader (rms, Encoding.UTF8);
533 while (rms.Position < rms.Length)
534 session.Add (reader_session_items++, rbr.ReadString ());
535 }
536 var benc = Encoder as BinaryMessageEncoder;
537 if (benc != null)
538 benc.CurrentReaderSession = session;
539
540 // FIXME: supply maxSizeOfHeaders.
541 Message msg = Encoder.ReadMessage (s, 0x10000);
542 if (benc != null)
543 benc.CurrentReaderSession = null;
544 if (s.ReadByte () != UnsizedMessageTerminator)
545 throw new InvalidOperationException ("Unsized message terminator is expected");
546
547 return msg;
548 }
549
550 byte [] eof_buffer = new byte [1];
551 MyXmlBinaryWriterSession writer_session;
552
553 public void WriteSizedMessage (Message message)
554 {
555 ResetWriteBuffer ();
556
557 if (EncodingRecord != 8)
558 throw new NotImplementedException (String.Format ("Message encoding {0:X} is not implemented yet", EncodingRecord));
559
560 buffer.WriteByte (SizedEnvelopeRecord);
561
562 MemoryStream ms = new MemoryStream ();
563 var session = writer_session ?? new MyXmlBinaryWriterSession ();
564 writer_session = session;
565 int writer_session_count = session.List.Count;
566 var benc = Encoder as BinaryMessageEncoder;
567 try {
568 if (benc != null)
569 benc.CurrentWriterSession = session;
570 Encoder.WriteMessage (message, ms);
571 } finally {
572 benc.CurrentWriterSession = null;
573 }
574
575 // dictionary
576 MemoryStream msd = new MemoryStream ();
577 BinaryWriter dw = new BinaryWriter (msd);
578 for (int i = writer_session_count; i < session.List.Count; i++)
579 dw.Write (session.List [i].Value);
580 dw.Flush ();
581
582 int length = (int) (msd.Position + ms.Position);
583 var msda = msd.ToArray ();
584 int sizeOfLength = writer.GetSizeOfLength (msda.Length);
585
586 writer.WriteVariableInt (length + sizeOfLength); // dictionary array also involves the size of itself.
587 WriteSizedChunk (msda);
588 // message body
589 var arr = ms.GetBuffer ();
590 writer.Write (arr, 0, (int) ms.Position);
591
592 writer.Flush ();
593
594 s.Write (buffer.GetBuffer (), 0, (int) buffer.Position);
595 s.Flush ();
596 }
597
598 // FIXME: support timeout
599 public void WriteUnsizedMessage (Message message, TimeSpan timeout)
600 {
601 ResetWriteBuffer ();
602
603 if (EncodingRecord != 8)
604 throw new NotImplementedException (String.Format ("Message encoding {0:X} is not implemented yet", EncodingRecord));
605
606 buffer.WriteByte (UnsizedEnvelopeRecord);
607
608 MemoryStream ms = new MemoryStream ();
609 var session = writer_session ?? new MyXmlBinaryWriterSession ();
610 writer_session = session;
611 int writer_session_count = session.List.Count;
612 var benc = Encoder as BinaryMessageEncoder;
613 try {
614 if (benc != null)
615 benc.CurrentWriterSession = session;
616 Encoder.WriteMessage (message, ms);
617 } finally {
618 benc.CurrentWriterSession = null;
619 }
620
621 // dictionary
622 MemoryStream msd = new MemoryStream ();
623 BinaryWriter dw = new BinaryWriter (msd);
624 for (int i = writer_session_count; i < session.List.Count; i++)
625 dw.Write (session.List [i].Value);
626 dw.Flush ();
627
628 int length = (int) (msd.Position + ms.Position);
629 var msda = msd.ToArray ();
630
631 writer.Write (msda, 0, msda.Length);
632 // message body
633 var arr = ms.GetBuffer ();
634 writer.Write (arr, 0, (int) ms.Position);
635
636 writer.Write (UnsizedMessageTerminator); // terminator
637 writer.Flush ();
638
639 // FIXME: it should be rewritten to directly write to the stream.
640 s.Write (buffer.GetBuffer (), 0, (int) buffer.Position);
641 s.Flush ();
642 }
643
644 public void ProcessEndRecordInitiator ()
645 {
646 s.WriteByte (EndRecord); // it is required
647 s.Flush ();
648 }
649
650 public void ProcessEndRecordRecipient ()
651 {
652 int b;
653 if ((b = s.ReadByte ()) != EndRecord)
654 throw new ProtocolException (String.Format ("EndRecord message was expected, got {0:X}", b));
655 }
656 }
657 }
git-svn mirror of mcs