sdks/wasm/Mono.WebAssembly.DebuggerProxy/WsClient.cs

   1 using System;
   2 using System.Linq;
   3 using System.Threading.Tasks;
   4 using Newtonsoft.Json.Linq;
   5
   6 using System.Net.WebSockets;
   7 using System.Threading;
   8 using System.IO;
   9 using System.Text;
  10 using System.Collections.Generic;
  11
  12 namespace WsProxy {
  13         public class WsClient: IDisposable {
  14                 ClientWebSocket socket;
  15                 List<Task> pending_ops = new List<Task> ();
  16                 TaskCompletionSource<bool> side_exit = new TaskCompletionSource<bool> ();
  17                 List<byte []> pending_writes = new List<byte []> ();
  18                 Task current_write;
  19
  20                 public WsClient () {
  21                 }
  22
  23                 ~WsClient() {
  24                         Dispose(false);
  25                 }
  26
  27                 public void Dispose() {
  28                         Dispose(true);
  29                 }
  30
  31                 protected virtual void Dispose (bool disposing) {
  32                         if (disposing)
  33                                 socket.Dispose ();
  34                 }
  35
  36                 Task Pump (Task task, CancellationToken token)
  37                 {
  38                         if (task != current_write)
  39                                 return null;
  40                         current_write = null;
  41
  42                         pending_writes.RemoveAt (0);
  43
  44                         if (pending_writes.Count > 0) {
  45                                 current_write = socket.SendAsync (new ArraySegment<byte> (pending_writes [0]), WebSocketMessageType.Text, true, token);
  46                                 return current_write;
  47                         }
  48                         return null;
  49                 }
  50
  51                 async Task<string> ReadOne (CancellationToken token)
  52                 {
  53                         byte [] buff = new byte [4000];
  54                         var mem = new MemoryStream ();
  55                         while (true) {
  56                                 var result = await this.socket.ReceiveAsync (new ArraySegment<byte> (buff), token);
  57                                 if (result.MessageType == WebSocketMessageType.Close) {
  58                                         return null;
  59                                 }
  60
  61                                 if (result.EndOfMessage) {
  62                                         mem.Write (buff, 0, result.Count);
  63                                         return Encoding.UTF8.GetString (mem.GetBuffer (), 0, (int)mem.Length);
  64                                 } else {
  65                                         mem.Write (buff, 0, result.Count);
  66                                 }
  67                         }
  68                 }
  69
  70                 protected void Send (byte [] bytes, CancellationToken token)
  71                 {
  72                         pending_writes.Add (bytes);
  73                         if (pending_writes.Count == 1) {
  74                                 if (current_write != null)
  75                                         throw new Exception ("Internal state is bad. current_write must be null if there are no pending writes");
  76
  77                                 current_write = socket.SendAsync (new ArraySegment<byte> (bytes), WebSocketMessageType.Text, true, token);
  78                                 pending_ops.Add (current_write);
  79                         }
  80                 }
  81
  82                 async Task MarkCompleteAfterward (Func<CancellationToken, Task> send, CancellationToken token)
  83                 {
  84                         try {
  85                                 await send(token);
  86                                 side_exit.SetResult (true);
  87                         } catch (Exception e) {
  88                                 side_exit.SetException (e);
  89                         }
  90                 }
  91                 protected async Task<bool> ConnectWithMainLoops(
  92                         Uri uri,
  93                         Func<string, CancellationToken, Task> receive,
  94                         Func<CancellationToken, Task> send,
  95                         CancellationToken token) {
  96
  97                         Console.WriteLine ("connecting to {0}", uri);
  98                         this.socket = new ClientWebSocket ();
  99                         this.socket.Options.KeepAliveInterval = Timeout.InfiniteTimeSpan;
 100
 101                         await this.socket.ConnectAsync (uri, token);
 102                         pending_ops.Add (ReadOne (token));
 103                         pending_ops.Add (side_exit.Task);
 104                         pending_ops.Add (MarkCompleteAfterward (send, token));
 105
 106                         while (!token.IsCancellationRequested) {
 107                                 var task = await Task.WhenAny (pending_ops);
 108                                 if (task == pending_ops [0]) { //pending_ops[0] is for message reading
 109                                         var msg = ((Task<string>)task).Result;
 110                                         pending_ops [0] = ReadOne (token);
 111                                         Task tsk = receive (msg, token);
 112                                         if (tsk != null)
 113                                                 pending_ops.Add (tsk);
 114                                 } else if (task == pending_ops [1]) {
 115                                         var res = ((Task<bool>)task).Result;
 116                                         //it might not throw if exiting successfull
 117                                         return res;
 118                                 } else { //must be a background task
 119                                         pending_ops.Remove (task);
 120                                         var tsk = Pump (task, token);
 121                                         if (tsk != null)
 122                                                 pending_ops.Add (tsk);
 123                                 }
 124                         }
 125
 126                         return false;
 127                 }
 128         }
 129 }