Adding error handling

Fixing SSE transport on the server

samples/SocketsSample/wwwroot/hubs.html

@@ -25,22 +25,37 @@
 
             document.getElementById('head1').innerHTML = transports ? transports.join(', ') : "auto (WebSockets)";
 
-            let connectButton = document.getElementById('connect');
             let connection = new RpcConnection(`http://${document.location.host}/hubs`, 'formatType=json&format=text');
             connection.on('Send', msg => {
                 addLine(msg); });
-            let isConnected = false;
+            connection.connectionClosed = e => {
+                if (e) {
+                    addLine('Connection closed with error: ' + e, 'red');
+                }
+                else {
+                    addLine('Disconnected', 'green');
+                }
+            }
 
+            let isConnected = false;
+            let connectButton = document.getElementById('connect');
             connectButton.addEventListener('click', () => {
                 connection.start(transports)
                 .then(() => {
                     isConnected = true;
+                    addLine('Connected successfully', 'green');
                 })
                 .catch(err => {
-                    addLine(err, true);
+                    addLine(err, 'red');
                 });
             });
 
+            let disconnectButton = document.getElementById('disconnect');
+            disconnectButton.addEventListener('click', () => {
+                connection.stop();
+                isConnected = false;
+            });
+
             document.getElementById('sendmessage').addEventListener('submit', event => {
                 let data = document.getElementById('data').value;
 
@@ -54,18 +69,18 @@
                         }
                     })
                     .catch(err => {
-                        addLine(err, true);
+                        addLine(err, 'red');
                     });
                 }
 
                 event.preventDefault();
             });
         });
 
-        function addLine(line, isError) {
+        function addLine(line, color) {
             var child = document.createElement('li');
-            if (isError === true) {
-                child.style.color = 'red';
+            if (color) {
+                child.style.color = color;
             }
             child.innerText = line;
             document.getElementById('messages').appendChild(child);
@@ -81,6 +96,7 @@ <h1 id="head1"></h1>
         </select>
 
         <input type="button" id="connect" value="Connect" />
+        <input type="button" id="disconnect" value="Disconnect" />
     </div>
 
     <form id="sendmessage">

src/Microsoft.AspNetCore.SignalR.Client.TS/Common.ts

@@ -1,2 +1,3 @@
 declare type DataReceived = (data: any) => void;
-declare type ErrorHandler = (e: any) => void;
+declare type ErrorHandler = (e: any) => void;
+declare type ConnectionClosed = (e?: any) => void;

src/Microsoft.AspNetCore.SignalR.Client.TS/Connection.ts

@@ -11,13 +11,12 @@ class Connection {
     private queryString: string;
     private connectionId: string;
     private transport: ITransport;
-    private dataReceivedCallback: DataReceived;
-    private errorHandler: ErrorHandler;
+    private dataReceivedCallback: DataReceived = (data: any) => { };
+    private connectionClosedCallback: ConnectionClosed = (error?: any) => { };
 
     constructor(url: string, queryString: string = "") {
         this.url = url;
         this.queryString = queryString;
-
         this.connectionState = ConnectionState.Disconnected;
     }
 
@@ -75,7 +74,7 @@ class Connection {
     private tryStartTransport(transports: ITransport[], index: number): Promise<ITransport> {
         let thisConnection = this;
         transports[index].onDataReceived = data => thisConnection.dataReceivedCallback(data);
-        transports[index].onError = e => thisConnection.errorHandler(e);
+        transports[index].onError = e => thisConnection.stopConnection(e);
 
         return transports[index].connect(this.url, this.queryString)
             .then(() => {
@@ -102,18 +101,23 @@ class Connection {
 
     stop(): void {
         if (this.connectionState != ConnectionState.Connected) {
-            throw new Error("Cannot stop the connection if is not in the 'Connected' State");
+            throw new Error("Cannot stop the connection if it is not in the 'Connected' State");
         }
 
+        this.stopConnection();
+    }
+
+    private stopConnection(error?: any) {
         this.transport.stop();
         this.connectionState = ConnectionState.Disconnected;
+        this.connectionClosedCallback(error);
     }
 
     set dataReceived(callback: DataReceived) {
         this.dataReceivedCallback = callback;
     }
 
-    set onError(callback: ErrorHandler) {
-        this.errorHandler = callback;
+    set connectionClosed(callback: ConnectionClosed) {
+        this.connectionClosedCallback = callback;
     }
 }

src/Microsoft.AspNetCore.SignalR.Client.TS/LongPollingTransport.ts

@@ -6,48 +6,63 @@ class LongPollingTransport implements ITransport {
     connect(url: string, queryString: string): Promise<void> {
         this.url = url;
         this.queryString = queryString;
-        this.pollXhr = new XMLHttpRequest();
-        // TODO: resolve promise on open sending? + reject on error
         this.poll(url + "/poll?" + this.queryString)
         return Promise.resolve();
     }
 
     private poll(url: string): void {
-        //TODO: timeout
-        this.pollXhr.open("GET", url, true);
-        this.pollXhr.send();
-        this.pollXhr.onload = () => {
-            if (this.pollXhr.status >= 200 && this.pollXhr.status < 300) {
-                this.onDataReceived(this.pollXhr.response);
-                this.poll(url);
+        let thisLongPollingTransport = this;
+        let pollXhr = new XMLHttpRequest();
+
+        pollXhr.onload = () => {
+            if (pollXhr.status == 200) {
+                if (thisLongPollingTransport.onDataReceived) {
+                    thisLongPollingTransport.onDataReceived(pollXhr.response);
+                }
+                thisLongPollingTransport.poll(url);
+            }
+            else if (this.pollXhr.status == 204) {
+                // TODO: closed event?
             }
             else {
-                //TODO: handle error
-                /*
-                    {
-                        status: xhr.status,
-                        statusText: xhr.statusText
-                    };
-                }*/
-            };
-
-            this.pollXhr.onerror = () => {
-                /*
-                reject({
-                    status: xhr.status,
-                    statusText: xhr.statusText
-                });*/
-                //TODO: handle error
-            };
+                if (thisLongPollingTransport.onError) {
+                    thisLongPollingTransport.onError({
+                        status: pollXhr.status,
+                        statusText: pollXhr.statusText
+                    });
+                }
+            }
+        };
+
+        pollXhr.onerror = () => {
+            if (thisLongPollingTransport.onError) {
+                thisLongPollingTransport.onError({
+                    status: pollXhr.status,
+                    statusText: pollXhr.statusText
+                });
+            }
         };
+
+        pollXhr.ontimeout = () => {
+            thisLongPollingTransport.poll(url);
+        }
+
+        this.pollXhr = pollXhr;
+        this.pollXhr.open("GET", url, true);
+        // TODO: consider making timeout configurable
+        this.pollXhr.timeout = 110000;
+        this.pollXhr.send();
     }
 
     send(data: any): Promise<void> {
         return new HttpClient().post(this.url + "/send?" + this.queryString, data);
     }
 
     stop(): void {
-        this.pollXhr.abort();
+        if (this.pollXhr) {
+            this.pollXhr.abort();
+            this.pollXhr = null;
+        }
     }
 
     onDataReceived: DataReceived;

src/Microsoft.AspNetCore.SignalR.Client.TS/RpcConnection.ts

@@ -93,4 +93,8 @@ class RpcConnection {
     on(methodName: string, method: (...args: any[]) => void) {
         this.methods[methodName] = method;
     }
+
+    set connectionClosed(callback: ConnectionClosed) {
+        this.connection.connectionClosed = callback;
+    }
 }

src/Microsoft.AspNetCore.SignalR.Client.TS/ServerSentEventsTransport.ts

@@ -13,29 +13,47 @@ class ServerSentEventsTransport implements ITransport {
         this.queryString = queryString;
         this.url = url;
         let tmp = `${this.url}/sse?${this.queryString}`;
-        try {
-            this.eventSource = new EventSource(`${this.url}/sse?${this.queryString}`);
-
-            this.eventSource.onmessage = (e: MessageEvent) => {
-                this.onDataReceived(e.data);
-            };
-            this.eventSource.onerror = (e: Event) => {
-                // todo: handle errors
-            }
-        }
-        catch (e) {
-            return Promise.reject(e);
-        }
 
-        return Promise.resolve();
+        return new Promise((resolve, reject) => {
+            let eventSource = new EventSource(`${this.url}/sse?${this.queryString}`);
+
+            try {
+                let thisEventSourceTransport = this;
+                eventSource.onmessage = (e: MessageEvent) => {
+                    if (thisEventSourceTransport.onDataReceived) {
+                        thisEventSourceTransport.onDataReceived(e.data);
+                    }
+                };
+
+                eventSource.onerror = (e: Event) => {
+                    reject();
+
+                    // don't report an error if the transport did not start successfully
+                    if (thisEventSourceTransport.eventSource && thisEventSourceTransport.onError) {
+                        thisEventSourceTransport.onError(e);
+                    }
+                }
+
+                eventSource.onopen = () => {
+                    thisEventSourceTransport.eventSource = eventSource;
+                    resolve();
+                }
+            }
+            catch (e) {
+                return Promise.reject(e);
+            }
+        });
     }
 
     send(data: any): Promise<void> {
         return new HttpClient().post(this.url + "/send?" + this.queryString, data);
     }
 
     stop(): void {
-        this.eventSource.close();
+        if (this.eventSource) {
+            this.eventSource.close();
+            this.eventSource = null;
+        }
     }
 
     onDataReceived: DataReceived;

src/Microsoft.AspNetCore.SignalR.Client.TS/WebSocketTransport.ts

@@ -5,33 +5,52 @@ class WebSocketTransport implements ITransport {
         return new Promise((resolve, reject) => {
             url = url.replace(/^http/, "ws");
             let connectUrl = url + "/ws?" + queryString;
-            this.webSocket = new WebSocket(connectUrl);
-            this.webSocket.onopen = (event: Event) => {
+
+            let webSocket = new WebSocket(connectUrl);
+            let thisWebSocketTransport = this;
+
+            webSocket.onopen = (event: Event) => {
                 console.log(`WebSocket connected to ${connectUrl}`);
+                thisWebSocketTransport.webSocket = webSocket;
                 resolve();
             };
 
-            this.webSocket.onerror = (event: Event) => {
-                // TODO: also handle when connection was opened successfully
+            webSocket.onerror = (event: Event) => {
                 reject();
             };
 
-            this.webSocket.onmessage = (message: MessageEvent) => {
+            webSocket.onmessage = (message: MessageEvent) => {
                 console.log(`(WebSockets transport) data received: ${message.data}`);
-                if (this.onDataReceived) {
-                    this.onDataReceived(message.data);
+                if (thisWebSocketTransport.onDataReceived) {
+                    thisWebSocketTransport.onDataReceived(message.data);
+                }
+            }
+
+            webSocket.onclose = (event: CloseEvent) => {
+                // webSocket will be null if the transport did not start successfully
+                if (thisWebSocketTransport.webSocket && event.wasClean === false) {
+                    if (thisWebSocketTransport.onError) {
+                        thisWebSocketTransport.onError(event);
+                    }
                 }
             }
         });
     }
 
     send(data: any): Promise<void> {
-        this.webSocket.send(data);
-        return Promise.resolve();
+        if (this.webSocket && this.webSocket.readyState === WebSocket.OPEN) {
+            this.webSocket.send(data);
+            return Promise.resolve();
+        }
+
+        return Promise.reject("WebSocket is not in OPEN state");
     }
 
     stop(): void {
-        this.webSocket.close();
+        if (this.webSocket) {
+            this.webSocket.close();
+            this.webSocket = null;
+        }
     }
 
     onDataReceived: DataReceived;

src/Microsoft.AspNetCore.Sockets/ServerSentEvents.cs

@@ -21,6 +21,7 @@ public async Task ProcessRequest(HttpContext context)
             context.Response.ContentType = "text/event-stream";
             context.Response.Headers["Cache-Control"] = "no-cache";
             context.Response.Headers["Content-Encoding"] = "identity";
+            await context.Response.Body.FlushAsync();
 
             while (true)
             {