RSocketWebSocketClient-test.js

/** Copyright (c) Facebook, Inc. and its affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

'use strict';

/* eslint-env browser */

jest.useFakeTimers();

import {FRAME_TYPES, deserializeFrame, serializeFrame} from 'rsocket-core';
import {genMockPublisher} from 'MockFlowableSubscription';
import {genMockSubscriber} from 'MockFlowableSubscriber';
import RSocketWebSocketClient from '../RSocketWebSocketClient';

describe('RSocketWebSocketClient', () => {
  beforeEach(() => {
    jest.clearAllTimers();
  });

  describe('connectionStatus() and connect()', () => {
    let client;
    let status;

    beforeEach(() => {
      client = new RSocketWebSocketClient({url: 'wss://...'});
      client.connectionStatus().subscribe({
        onNext: _status => (status = _status),
        onSubscribe: subscription =>
          subscription.request(Number.MAX_SAFE_INTEGER),
      });
    });

    it('initially returns NOT_CONNECTED', () => {
      expect(status.kind).toBe('NOT_CONNECTED');
    });

    it('returns CONNECTING while connecting', () => {
      client.connect();
      expect(status.kind).toBe('CONNECTING');
    });

    it('returns CONNECTED once connected', () => {
      client.connect();
      WebSocket.socket.mock.open();
      expect(status.kind).toBe('CONNECTED');
    });

    it('returns ERROR if the socket errors', () => {
      client.connect();
      WebSocket.socket.mock.error(new Error('oopsie'));
      expect(status.kind).toBe('ERROR');
      expect(status.error.message).toBe('oopsie');
    });

    it('returns CLOSED if explicitly closed', () => {
      client.connect();
      client.close();
      expect(status.kind).toBe('CLOSED');
    });
  });

  describe('post-connect() APIs', () => {
    let client;
    let socket;

    const frame = {
      data: null,
      flags: 0,
      lastReceivedPosition: 0,
      streamId: 0,
      type: FRAME_TYPES.KEEPALIVE,
    };

    beforeEach(() => {
      client = new RSocketWebSocketClient({url: 'wss://...'});
      client.connect();
      jest.runAllTimers();
      socket = WebSocket.socket;
      socket.mock.open();
      jest.runAllTimers();
    });

    describe('close()', () => {
      describe('given an error', () => {
        it('closes the socket', () => {
          client.close(new Error());
          expect(socket.close.mock.calls.length).toBe(1);
        });

        it('sets the status to ERROR with the given error', () => {
          let status;
          client.connectionStatus().subscribe({
            onNext: _status => (status = _status),
            onSubscribe: subscription =>
              subscription.request(Number.MAX_SAFE_INTEGER),
          });
          const error = new Error();
          client.close(error);
          expect(status.kind).toBe('ERROR');
          expect(status.error).toBe(error);
        });

        it('calls receive.onError with the given error', () => {
          const onError = jest.fn();
          const onSubscribe = subscription =>
            subscription.request(Number.MAX_SAFE_INTEGER);
          client.receive().subscribe({onError, onSubscribe});
          const error = new Error();
          client.close(error);
          expect(onError.mock.calls.length).toBe(1);
          expect(onError.mock.calls[0][0]).toBe(error);
        });
      });

      describe('not given an error', () => {
        it('closes the socket', () => {
          client.close();
          expect(socket.close.mock.calls.length).toBe(1);
        });

        it('sets the status to CLOSED', () => {
          let status;
          client.connectionStatus().subscribe({
            onNext: _status => (status = _status),
            onSubscribe: subscription =>
              subscription.request(Number.MAX_SAFE_INTEGER),
          });
          client.close();
          expect(status.kind).toBe('CLOSED');
        });

        it('calls receive.onComplete', () => {
          const onComplete = jest.fn();
          const onSubscribe = subscription =>
            subscription.request(Number.MAX_SAFE_INTEGER);
          client.receive().subscribe({onComplete, onSubscribe});
          client.close();
          expect(onComplete.mock.calls.length).toBe(1);
        });
      });
    });

    describe('sendOne()', () => {
      it('sends a frame', () => {
        client.sendOne(frame);
        expect(socket.send.mock.calls.length).toBe(1);
        const buffer = socket.send.mock.calls[0][0];
        expect(deserializeFrame(buffer)).toEqual({
          ...frame,
          length: buffer.length,
        });
      });

      it('calls receive.onError if the frame cannot be sent', () => {
        const onError = jest.fn();
        const onSubscribe = subscription =>
          subscription.request(Number.MAX_SAFE_INTEGER);
        client.receive().subscribe({onError, onSubscribe});
        socket.send = () => {
          throw new Error('wtf');
        };
        client.sendOne(frame);
        expect(onError.mock.calls.length).toBe(1);
      });
    });

    describe('send()', () => {
      it('sends frames', () => {
        const frame2 = {...frame, flags: 1};
        const publisher = genMockPublisher();
        client.send(publisher);
        publisher.onNext(frame);
        publisher.onNext(frame2);
        expect(socket.send.mock.calls.length).toBe(2);
        const buffer = socket.send.mock.calls[0][0];
        expect(deserializeFrame(buffer)).toEqual({
          ...frame,
          length: buffer.length,
        });
        const buffer2 = socket.send.mock.calls[1][0];
        expect(deserializeFrame(buffer2)).toEqual({
          ...frame2,
          length: buffer2.length,
        });
      });

      it('calls receive.onError if frames cannot be sent', () => {
        const onError = jest.fn();
        const onSubscribe = subscription =>
          subscription.request(Number.MAX_SAFE_INTEGER);
        client.receive().subscribe({onError, onSubscribe});
        socket.send = () => {
          throw new Error('wtf');
        };
        const publisher = genMockPublisher();
        client.send(publisher);
        publisher.onNext(frame);
        expect(onError.mock.calls.length).toBe(1);
      });

      it('unsubscribes when closed', () => {
        const publisher = genMockPublisher();
        client.send(publisher);
        client.close();
        expect(publisher.cancel).toBeCalled();
      });
    });

    describe('receive()', () => {
      it('calls onNext with deserialized frames', () => {
        const subscriber = genMockSubscriber({
          onSubscribe(subscription) {
            subscription.request(Number.MAX_SAFE_INTEGER);
          },
        });
        client.receive().subscribe(subscriber);
        expect(subscriber.onNext.mock.calls.length).toBe(0);

        socket.mock.message(serializeFrame(frame));
        expect(subscriber.onNext.mock.calls.length).toBe(1);
        const nextFrame = subscriber.onNext.mock.calls[0][0];
        expect(nextFrame).toEqual({...frame, length: nextFrame.length});

        expect(subscriber.onComplete.mock.calls.length).toBe(0);
        expect(subscriber.onError.mock.calls.length).toBe(0);
      });

      it('calls onError when partial frames are received', () => {
        const subscriber = genMockSubscriber({
          onSubscribe(subscription) {
            subscription.request(Number.MAX_SAFE_INTEGER);
          },
        });
        client.receive().subscribe(subscriber);

        const buffer = serializeFrame(frame);
        socket.mock.message(buffer.slice(0, 2));
        expect(subscriber.onComplete.mock.calls.length).toBe(0);
        expect(subscriber.onNext.mock.calls.length).toBe(0);
        expect(subscriber.onError.mock.calls.length).toBe(1);
      });

      it('calls onComplete when intentionally close()-ed', () => {
        const subscriber = genMockSubscriber({
          onSubscribe(subscription) {
            subscription.request(Number.MAX_SAFE_INTEGER);
          },
        });
        client.receive().subscribe(subscriber);
        client.close();
        expect(subscriber.onComplete.mock.calls.length).toBe(1);
        expect(subscriber.onError.mock.calls.length).toBe(0);
        expect(subscriber.onNext.mock.calls.length).toBe(0);
      });

      it('calls onError when the socket is closed by the peer', () => {
        const subscriber = genMockSubscriber({
          onSubscribe(subscription) {
            subscription.request(Number.MAX_SAFE_INTEGER);
          },
        });
        client.receive().subscribe(subscriber);
        socket.mock.close();
        expect(subscriber.onComplete.mock.calls.length).toBe(0);
        expect(subscriber.onError.mock.calls.length).toBe(1);
        const error = subscriber.onError.mock.calls[0][0];
        expect(error.message).toBe(
          'RSocketWebSocketClient: Socket closed unexpectedly.',
        );
        expect(subscriber.onNext.mock.calls.length).toBe(0);
      });

      it('calls onError when invalid frames are received', () => {
        const subscriber = genMockSubscriber({
          onSubscribe(subscription) {
            subscription.request(Number.MAX_SAFE_INTEGER);
          },
        });
        client.receive().subscribe(subscriber);
        // Emit a frame of length one, which is shorter than the smallest
        // possible frame (3 bytes of length, 1 byte of payload).
        const buffer = new Buffer([0x00, 0x00, 0x01, 0x00]);
        socket.mock.message(buffer);

        expect(subscriber.onComplete.mock.calls.length).toBe(0);
        expect(subscriber.onError.mock.calls.length).toBe(1);
        expect(subscriber.onNext.mock.calls.length).toBe(0);
      });

      it('calls onError when a socket error occurs', () => {
        const subscriber = genMockSubscriber({
          onSubscribe(subscription) {
            subscription.request(Number.MAX_SAFE_INTEGER);
          },
        });
        client.receive().subscribe(subscriber);
        socket.mock.error(new Error('oops'));
        expect(subscriber.onComplete.mock.calls.length).toBe(0);
        expect(subscriber.onError.mock.calls.length).toBe(1);
        expect(subscriber.onError.mock.calls[0][0].message).toBe('oops');
        expect(subscriber.onNext.mock.calls.length).toBe(0);
      });
    });
  });
});