src/models/ws-stream.js
function openStream (makeUrl, authorization, endTimestamp, retries = MAX_RETRY_COUNT) {
  const endTs = endTimestamp || null;
  const s_websocket = openWebSocket(makeUrl(endTs), authorization);

  // Stream which contains only one element: the date at which the websocket closed
  const s_endTimestamp = s_websocket.filter(false).mapEnd(new Date());
  const s_interruption = s_endTimestamp.flatMapLatest((endTimestamp) => {
    Logger.warn('WebSocket connexion closed, reconnecting...');
    if (retries === 0) {
      return new Bacon.Error(`WebSocket connexion failed ${MAX_RETRY_COUNT} times!`);
    }
    return Bacon.later(RETRY_DELAY, null).flatMapLatest(() => {
      return openStream(makeUrl, authorization, endTimestamp, retries - 1);
    });
  });

  return Bacon.mergeAll(s_websocket, s_interruption);
}
function openStream (makeUrl, authorization, endTimestamp, retries = MAX_RETRY_COUNT) {
  const endTs = endTimestamp || null;
  const s_websocket = openWebSocket(makeUrl(endTs), authorization);

  // Stream which contains only one element: the date at which the websocket closed
  const s_endTimestamp = s_websocket.filter(false).mapEnd(new Date());
  const s_interruption = s_endTimestamp.flatMapLatest((endTimestamp) => {
    Logger.warn('WebSocket connexion closed, reconnecting...');
    if (retries === 0) {
      return new Bacon.Error(`WebSocket connexion failed ${MAX_RETRY_COUNT} times!`);
    }
    return Bacon.later(RETRY_DELAY, null).flatMapLatest(() => {
      return openStream(makeUrl, authorization, endTimestamp, retries - 1);
    });
  });

  return Bacon.mergeAll(s_websocket, s_interruption);
}
function openStream ( makeUrl , authorization , endTimestamp , retries = MAX_RETRY_COUNT ) { const endTs = endTimestamp || null ; const s_websocket = openWebSocket ( makeUrl ( endTs ) , authorization ) ; // Stream which contains only one element: the date at which the websocket closed const s_endTimestamp = s_websocket . filter ( false ) . mapEnd ( new Date ( ) ) ; const s_interruption = s_endTimestamp . flatMapLatest ( ( endTimestamp ) => { Logger . warn ( 'WebSocket connexion closed, reconnecting...' ) ; if ( retries === 0 ) { return new Bacon . Error ( ` ${ MAX_RETRY_COUNT } ` ) ; } return Bacon . later ( RETRY_DELAY , null ) . flatMapLatest ( ( ) => { return openStream ( makeUrl , authorization , endTimestamp , retries - 1 ) ; } ) ; } ) ; return Bacon . mergeAll ( s_websocket , s_interruption ) ; }
Open a never-ending stream of events, backed by a websocket.
If websocket connection is closed, it is automatically re-opened.
Ping messages are regularly sent to the server to keep the connection alive and
avoid disconnections.

makeUrl: Timestamp => String :
The WS URL can be constructed based on the closing date, in case the WS supports resuming.
On the first WS connection, this value will be null
authorization: The content of the authorization message sent to server
Open a never - ending stream of events backed by a websocket . If websocket connection is closed it is automatically re - opened . Ping messages are regularly sent to the server to keep the connection alive and avoid disconnections .