feat(webSocket): Add binaryType to config object

Add binaryType to config object, so that it is possible
to set that parameter on underlying socket before any
data emits happen.

Closes ReactiveX#2353

spec/observables/dom/webSocket-spec.ts

@@ -261,6 +261,20 @@ describe('Observable.webSocket', () => {
  subject.unsubscribe();
  });
 
+ it('should take a binaryType and set it properly on the web socket', () => {
+ const subject = Observable.webSocket({
+ url: 'ws://mysocket',
+ binaryType: 'blob'
+ });
+
+ subject.subscribe();
+
+ const socket = MockWebSocket.lastSocket;
+ expect(socket.binaryType).to.equal('blob');
+
+ subject.unsubscribe();
+ });
+
  it('should take a resultSelector', () => {
  const results = [];
 
@@ -632,6 +646,7 @@ class MockWebSocket {
  readyState: number = 0;
  closeCode: any;
  closeReason: any;
+ binaryType?: string;
 
  constructor(public url: string, public protocol: string) {
  MockWebSocket.sockets.push(this);

src/observable/dom/WebSocketSubject.ts

@@ -18,6 +18,7 @@ export interface WebSocketSubjectConfig {
  closeObserver?: NextObserver<CloseEvent>;
  closingObserver?: NextObserver<void>;
  WebSocketCtor?: { new(url: string, protocol?: string|Array<string>): WebSocket };
+ binaryType?: 'blob' | 'arraybuffer';
 }
 
 /**
@@ -34,6 +35,7 @@ export class WebSocketSubject<T> extends AnonymousSubject<T> {
  closeObserver: NextObserver<CloseEvent>;
  closingObserver: NextObserver<void>;
  WebSocketCtor: { new(url: string, protocol?: string|Array<string>): WebSocket };
+ binaryType?: 'blob' | 'arraybuffer';
 
  private _output: Subject<T>;
 
@@ -159,6 +161,9 @@ export class WebSocketSubject<T> extends AnonymousSubject<T> {
  new WebSocketCtor(this.url, this.protocol) :
  new WebSocketCtor(this.url);
  this.socket = socket;
+ if (this.binaryType) {
+ this.socket.binaryType = this.binaryType;
+ }
  } catch (e) {
  observer.error(e);
  return;