SystemNetWebSocketsClientWebSocket.cs 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. using System;
  2. using System.Net.WebSockets;
  3. using System.Reactive.Linq;
  4. using System.Reactive.Subjects;
  5. using System.Text;
  6. using System.Threading;
  7. using System.Threading.Tasks;
  8. namespace SocketIOClient.Transport
  9. {
  10. public class SystemNetWebSocketsClientWebSocket : IClientWebSocket
  11. {
  12. public SystemNetWebSocketsClientWebSocket(int eio)
  13. {
  14. _eio = eio;
  15. _textSubject = new Subject<string>();
  16. _bytesSubject = new Subject<byte[]>();
  17. TextObservable = _textSubject.AsObservable();
  18. BytesObservable = _bytesSubject.AsObservable();
  19. _ws = new ClientWebSocket();
  20. _listenCancellation = new CancellationTokenSource();
  21. _sendLock = new SemaphoreSlim(1, 1);
  22. }
  23. const int ReceiveChunkSize = 1024 * 8;
  24. readonly int _eio;
  25. readonly ClientWebSocket _ws;
  26. readonly Subject<string> _textSubject;
  27. readonly Subject<byte[]> _bytesSubject;
  28. readonly CancellationTokenSource _listenCancellation;
  29. readonly SemaphoreSlim _sendLock;
  30. public IObservable<string> TextObservable { get; }
  31. public IObservable<byte[]> BytesObservable { get; }
  32. private void Listen()
  33. {
  34. Task.Factory.StartNew(async() =>
  35. {
  36. while (true)
  37. {
  38. if (_listenCancellation.IsCancellationRequested)
  39. {
  40. break;
  41. }
  42. var buffer = new byte[ReceiveChunkSize];
  43. int count = 0;
  44. WebSocketReceiveResult result = null;
  45. while (_ws.State == WebSocketState.Open)
  46. {
  47. var subBuffer = new byte[ReceiveChunkSize];
  48. try
  49. {
  50. result = await _ws.ReceiveAsync(new ArraySegment<byte>(subBuffer), CancellationToken.None).ConfigureAwait(false);
  51. // resize
  52. if (buffer.Length - count < result.Count)
  53. {
  54. Array.Resize(ref buffer, buffer.Length + result.Count);
  55. }
  56. Buffer.BlockCopy(subBuffer, 0, buffer, count, result.Count);
  57. count += result.Count;
  58. if (result.EndOfMessage)
  59. {
  60. break;
  61. }
  62. }
  63. catch (Exception e)
  64. {
  65. _textSubject.OnError(e);
  66. break;
  67. }
  68. }
  69. if (result == null)
  70. {
  71. break;
  72. }
  73. switch (result.MessageType)
  74. {
  75. case WebSocketMessageType.Text:
  76. string text = Encoding.UTF8.GetString(buffer, 0, count);
  77. _textSubject.OnNext(text);
  78. break;
  79. case WebSocketMessageType.Binary:
  80. byte[] bytes;
  81. if (_eio == 3)
  82. {
  83. bytes = new byte[count - 1];
  84. Buffer.BlockCopy(buffer, 1, bytes, 0, bytes.Length);
  85. }
  86. else
  87. {
  88. bytes = new byte[count];
  89. Buffer.BlockCopy(buffer, 0, bytes, 0, bytes.Length);
  90. }
  91. _bytesSubject.OnNext(bytes);
  92. break;
  93. case WebSocketMessageType.Close:
  94. _textSubject.OnError(new WebSocketException("Received a Close message"));
  95. break;
  96. }
  97. }
  98. });
  99. }
  100. public async Task ConnectAsync(Uri uri, CancellationToken cancellationToken)
  101. {
  102. await _ws.ConnectAsync(uri, cancellationToken);
  103. Listen();
  104. }
  105. public async Task DisconnectAsync(CancellationToken cancellationToken)
  106. {
  107. await _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken);
  108. }
  109. public async Task SendAsync(byte[] bytes, TransportMessageType type, bool endOfMessage, CancellationToken cancellationToken)
  110. {
  111. var msgType = WebSocketMessageType.Text;
  112. if (type == TransportMessageType.Binary)
  113. {
  114. msgType = WebSocketMessageType.Binary;
  115. }
  116. await _ws.SendAsync(new ArraySegment<byte>(bytes), msgType, endOfMessage, cancellationToken).ConfigureAwait(false);
  117. }
  118. public void AddHeader(string key, string val)
  119. {
  120. _ws.Options.SetRequestHeader(key, val);
  121. }
  122. public void Dispose()
  123. {
  124. _textSubject.Dispose();
  125. _bytesSubject.Dispose();
  126. _ws.Dispose();
  127. }
  128. }
  129. }