fix(webSocket): manage serializer throwing exceptions

spec/observables/dom/webSocket-spec.ts

@@ -416,6 +416,52 @@ describe('webSocket', () => {
       subject.unsubscribe();
     });
 
+    it('should take a serializer', () => {
+      const subject = webSocket<{ data: string}>({
+        url: 'ws://mysocket',
+        serializer: (e: any) => {
+          return e.data + '!';
+        }
+      });
+
+      subject.subscribe();
+
+      const socket = MockWebSocket.lastSocket;
+      socket.open();
+
+
+      ['ahoy', 'yarr', 'shove off'].forEach(x => {
+        subject.next({ data: x });
+        expect(socket.lastMessageSent).to.equal(x + '!');
+      });
+
+      subject.unsubscribe();
+    });
+
+    it('if the serializer fails it should close the connection and rethrow the error', () => {
+      const subject = webSocket<string>({
+        url: 'ws://mysocket',
+        serializer: (e: any) => {
+          throw new Error('I am a bad error');
+        }
+      });
+
+      subject.subscribe({ next: (x: any) => {
+        expect(x).to.equal('this should not happen');
+      }, error: (err: any) => {
+        expect(err).to.equal('this should not happen');
+      } });
+
+      const socket = MockWebSocket.lastSocket;
+      socket.open();
+
+      expect(() => subject.next('weee!')).to.throw('I am a bad error');
+      expect(socket.readyState).to.equal(WebSocketState.CLOSING);
+      expect(socket.closeCode).to.equal(1000);
+
+      subject.unsubscribe();
+    });
+
     it('should accept a closingObserver', () => {
       let calls = 0;
       const subject = webSocket<string>(<any>{

src/internal/observable/dom/WebSocketSubject.ts

@@ -296,18 +296,25 @@ export class WebSocketSubject<T> extends AnonymousSubject<T> {
 
       const queue = this.destination;
 
-      this.destination = Subscriber.create<T>(
-        (x) => {
+      this.destination = new Subscriber<T>({
+        next: (x: T) => {
           if (socket!.readyState === 1) {
             try {
               const { serializer } = this._config;
               socket!.send(serializer!(x!));
-            } catch (e) {
-              this.destination!.error(e);
+            } catch (e: any) {
+              if (e && e.code) {
+                this.destination!.error(e);
+              } else {
+                this.destination!.error({
+                  code: 1000,
+                });
+                throw e;
+              }
             }
           }
         },
-        (err) => {
+        error: (err: any) => {
           const { closingObserver } = this._config;
           if (closingObserver) {
             closingObserver.next(undefined);
@@ -319,15 +326,15 @@ export class WebSocketSubject<T> extends AnonymousSubject<T> {
           }
           this._resetState();
         },
-        () => {
+        complete: () => {
           const { closingObserver } = this._config;
           if (closingObserver) {
             closingObserver.next(undefined);
           }
           socket!.close();
           this._resetState();
-        }
-      ) as Subscriber<any>;
+        },
+      }) as Subscriber<any>;
 
       if (queue && queue instanceof ReplaySubject) {
         subscription.add((queue as ReplaySubject<T>).subscribe(this.destination));