feat: replace upgrade with websocket+base mechanism

src/common/lib/client/realtimechannel.ts

@@ -753,9 +753,7 @@ class RealtimeChannel extends EventEmitter {
   checkPendingState(): void {
     /* if can't send events, do nothing */
     const cmState = this.connectionManager.state;
-    /* Allow attach messages to queue up when synchronizing, since this will be
-     * the state we'll be in when upgrade transport.active triggers a checkpendingstate */
-    if (!(cmState.sendEvents || cmState.forceQueueEvents)) {
+    if (!cmState.sendEvents) {
       Logger.logAction(
         Logger.LOG_MINOR,
         'RealtimeChannel.checkPendingState',

src/common/lib/transport/transport.ts

@@ -64,7 +64,6 @@ abstract class Transport extends EventEmitter {
       params.heartbeats = true;
     }
     this.connectionManager = connectionManager;
-    connectionManager.registerProposedTransport(this);
     this.auth = auth;
     this.params = params;
     this.timeouts = params.options.timeouts;
@@ -296,7 +295,7 @@ abstract class Transport extends EventEmitter {
     auth: Auth,
     transportParams: TransportParams,
     callback: TryConnectCallback,
-  ): void {
+  ): Transport {
     const transport = new transportCtor(connectionManager, auth, transportParams);
 
     let transportAttemptTimer: NodeJS.Timeout | number;
@@ -324,6 +323,7 @@ abstract class Transport extends EventEmitter {
       callback(null, transport);
     });
     transport.connect();
+    return transport;
   }
 
   onAuthUpdated?: (tokenDetails: API.TokenDetails) => void;

src/common/lib/transport/websockettransport.ts

@@ -29,7 +29,7 @@ class WebSocketTransport extends Transport {
     super(connectionManager, auth, params);
     /* If is a browser, can't detect pings, so request protocol heartbeats */
     params.heartbeats = Platform.Config.useProtocolHeartbeats;
-    this.wsHost = Defaults.getHost(params.options, params.host, true);
+    this.wsHost = params.host as string;
   }
 
   static isAvailable() {

src/common/lib/util/defaults.ts

@@ -28,8 +28,8 @@ type CompleteDefaults = IDefaults & {
     connectionStateTtl: number;
     realtimeRequestTimeout: number;
     recvTimeout: number;
-    preferenceConnectTimeout: number;
-    parallelUpgradeDelay: number;
+    webSocketConnectTimeout: number;
+    webSocketSlowTimeout: number;
   };
   httpMaxRetryCount: number;
   maxMessageSize: number;
@@ -41,7 +41,7 @@ type CompleteDefaults = IDefaults & {
   getHttpScheme(options: ClientOptions): string;
   environmentFallbackHosts(environment: string): string[];
   getFallbackHosts(options: NormalisedClientOptions): string[];
-  getHosts(options: NormalisedClientOptions): string[];
+  getHosts(options: NormalisedClientOptions, ws?: boolean): string[];
   checkHost(host: string): void;
   getRealtimeHost(options: ClientOptions, production: boolean, environment: string): string;
   objectifyOptions(
@@ -80,8 +80,8 @@ const Defaults = {
     connectionStateTtl: 120000,
     realtimeRequestTimeout: 10000,
     recvTimeout: 90000,
-    preferenceConnectTimeout: 6000,
-    parallelUpgradeDelay: 6000,
+    webSocketConnectTimeout: 10000,
+    webSocketSlowTimeout: 4000,
   },
   httpMaxRetryCount: 3,
   maxMessageSize: 65536,
@@ -136,8 +136,9 @@ export function getFallbackHosts(options: NormalisedClientOptions): string[] {
   return fallbackHosts ? Utils.arrChooseN(fallbackHosts, httpMaxRetryCount) : [];
 }
 
-export function getHosts(options: NormalisedClientOptions): string[] {
-  return [options.restHost].concat(getFallbackHosts(options));
+export function getHosts(options: NormalisedClientOptions, ws?: boolean): string[] {
+  const hosts = [options.restHost].concat(getFallbackHosts(options));
+  return ws ? hosts.map((host) => getHost(options, host, true)) : hosts;
 }
 
 function checkHost(host: string): void {

src/common/types/IDefaults.d.ts

@@ -3,9 +3,7 @@ import { RestAgentOptions } from './ClientOptions';
 
 export default interface IDefaults {
   connectivityCheckUrl: string;
+  wsConnectivityUrl: string;
   defaultTransports: TransportName[];
-  baseTransportOrder: TransportName[];
-  transportPreferenceOrder: TransportName[];
-  upgradeTransports: TransportName[];
   restAgentOptions?: RestAgentOptions;
 }

src/common/types/IPlatformConfig.d.ts

@@ -30,7 +30,6 @@ export interface ISpecificPlatformConfig {
   fetchSupported?: boolean;
   xhrSupported?: boolean;
   allowComet?: boolean;
-  streamingSupported?: boolean;
   ArrayBuffer?: typeof ArrayBuffer | false;
   atob?: typeof atob | null;
   TextEncoder?: typeof TextEncoder;

src/platform/nativescript/config.js

@@ -24,7 +24,6 @@ var Config = {
   WebSocket: WebSocket,
   xhrSupported: XMLHttpRequest,
   allowComet: true,
-  streamingSupported: false,
   useProtocolHeartbeats: true,
   supportsBinary: typeof TextDecoder !== 'undefined' && TextDecoder,
   preferBinary: false, // Motivation as on web; see `preferBinary` comment there.

src/platform/nodejs/lib/util/defaults.ts

@@ -3,12 +3,10 @@ import { TransportNames } from '../../../../common/constants/TransportName';
 
 const Defaults: IDefaults = {
   connectivityCheckUrl: 'https://internet-up.ably-realtime.com/is-the-internet-up.txt',
+  wsConnectivityUrl: 'wss://ws-up.ably-realtime.com',
   /* Note: order matters here: the base transport is the leftmost one in the
    * intersection of baseTransportOrder and the transports clientOption that's supported. */
   defaultTransports: [TransportNames.WebSocket],
-  baseTransportOrder: [TransportNames.Comet, TransportNames.WebSocket],
-  transportPreferenceOrder: [TransportNames.Comet, TransportNames.WebSocket],
-  upgradeTransports: [TransportNames.WebSocket],
   restAgentOptions: { maxSockets: 40, keepAlive: true },
 };
 

src/platform/react-native/config.ts

@@ -13,7 +13,6 @@ export default function (bufferUtils: typeof BufferUtils): IPlatformConfig {
     WebSocket: WebSocket,
     xhrSupported: true,
     allowComet: true,
-    streamingSupported: true,
     useProtocolHeartbeats: true,
     supportsBinary: !!(typeof TextDecoder !== 'undefined' && TextDecoder),
     preferBinary: false, // Motivation as on web; see `preferBinary` comment there.

src/platform/web/config.ts

@@ -41,7 +41,6 @@ const Config: IPlatformConfig = {
   fetchSupported: !!globalObject.fetch,
   xhrSupported: globalObject.XMLHttpRequest && 'withCredentials' in new XMLHttpRequest(),
   allowComet: allowComet(),
-  streamingSupported: true,
   useProtocolHeartbeats: true,
   supportsBinary: !!globalObject.TextDecoder,
   /* Per Paddy (https://ably-real-time.slack.com/archives/CURL4U2FP/p1705674537763479) web intentionally prefers JSON to MessagePack:

src/platform/web/lib/util/defaults.ts

@@ -3,13 +3,11 @@ import { TransportNames } from 'common/constants/TransportName';
 
 const Defaults: IDefaults = {
   connectivityCheckUrl: 'https://internet-up.ably-realtime.com/is-the-internet-up.txt',
+  wsConnectivityUrl: 'wss://ws-up.ably-realtime.com',
   /* Order matters here: the base transport is the leftmost one in the
    * intersection of baseTransportOrder and the transports clientOption that's
    * supported. */
   defaultTransports: [TransportNames.XhrPolling, TransportNames.WebSocket],
-  baseTransportOrder: [TransportNames.XhrPolling, TransportNames.WebSocket],
-  transportPreferenceOrder: [TransportNames.XhrPolling, TransportNames.WebSocket],
-  upgradeTransports: [TransportNames.WebSocket],
 };
 
 export default Defaults;

test/browser/connection.test.js

@@ -369,50 +369,11 @@ define(['shared_helper', 'chai'], function (helper, chai) {
         monitorConnection(done, realtime);
       });
 
-      it('use_persisted_transport0', function (done) {
-        var transportPreferenceName = 'ably-transport-preference';
-        window.localStorage.setItem(transportPreferenceName, JSON.stringify({ value: 'web_socket' }));
-
-        var realtime = helper.AblyRealtime();
-
-        realtime.connection.connectionManager.on(function (transport) {
-          if (this.event === 'transport.active') {
-            try {
-              expect(transport.shortName).to.equal('web_socket');
-            } catch (err) {
-              closeAndFinish(done, realtime, err);
-              return;
-            }
-            closeAndFinish(done, realtime);
-          }
-        });
-        monitorConnection(done, realtime);
-      });
-
-      it('use_persisted_transport1', function (done) {
-        window.localStorage.setItem(transportPreferenceName, JSON.stringify({ value: 'xhr_polling' }));
-
-        var realtime = helper.AblyRealtime();
-
-        realtime.connection.connectionManager.on(function (transport) {
-          if (this.event === 'transport.active') {
-            try {
-              expect(transport.shortName).to.equal('xhr_polling');
-            } catch (err) {
-              closeAndFinish(done, realtime, err);
-              return;
-            }
-            closeAndFinish(done, realtime);
-          }
-        });
-        monitorConnection(done, realtime);
-      });
-
       it('browser_transports', function (done) {
         var realtime = helper.AblyRealtime();
         try {
           expect(realtime.connection.connectionManager.baseTransport).to.equal('xhr_polling');
-          expect(realtime.connection.connectionManager.upgradeTransports).to.deep.equal(['web_socket']);
+          expect(realtime.connection.connectionManager.webSocketTransportAvailable).to.be.ok;
         } catch (err) {
           closeAndFinish(done, realtime, err);
           return;

test/common/modules/shared_helper.js

@@ -154,7 +154,7 @@ define([
   }
 
   /* testFn is assumed to be a function of realtimeOptions that returns a mocha test */
-  function testOnAllTransports(name, testFn, excludeUpgrade, skip) {
+  function testOnAllTransports(name, testFn, skip) {
     var itFn = skip ? it.skip : it;
     let transports = availableTransports;
     transports.forEach(function (transport) {
@@ -167,18 +167,16 @@ define([
         testFn({ transports: [transport], useBinaryProtocol: false }),
       );
     });
-    /* Plus one for no transport specified (ie use upgrade mechanism if
+    /* Plus one for no transport specified (ie use websocket/base mechanism if
      * present).  (we explicitly specify all transports since node only does
-     * nodecomet+upgrade if comet is explicitly requested
+     * websocket+nodecomet if comet is explicitly requested)
      * */
-    if (!excludeUpgrade) {
-      itFn(name + '_with_binary_transport', testFn({ transports, useBinaryProtocol: true }));
-      itFn(name + '_with_text_transport', testFn({ transports, useBinaryProtocol: false }));
-    }
+    itFn(name + '_with_binary_transport', testFn({ transports, useBinaryProtocol: true }));
+    itFn(name + '_with_text_transport', testFn({ transports, useBinaryProtocol: false }));
   }
 
-  testOnAllTransports.skip = function (name, testFn, excludeUpgrade) {
-    testOnAllTransports(name, testFn, excludeUpgrade, true);
+  testOnAllTransports.skip = function (name, testFn) {
+    testOnAllTransports(name, testFn, true);
   };
 
   function restTestOnJsonMsgpack(name, testFn, skip) {

test/realtime/channel.test.js

@@ -282,9 +282,6 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
       },
       true,
     );
-    /* NB upgrade is excluded because realtime now sends an ATTACHED
-     * post-upgrade, which can race with the DETACHED if the DETACH is only sent
-     * just after upgrade. Re-include it with 1.1 spec which has IDs in ATTACHs */
 
     /*
      * Attach with an empty channel and expect a channel error
@@ -685,10 +682,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
                 },
                 function (cb) {
                   var channelUpdated = false;
-                  // attached or update: in case this is happening in parallel with
-                  // a transport upgrade, we might flip to attaching, meaning it'll come
-                  // through as attached not update
-                  channel._allChannelChanges.on(['attached', 'update'], function () {
+                  channel._allChannelChanges.on(['update'], function () {
                     channelUpdated = true;
                   });
 

test/realtime/connection.test.js

@@ -165,7 +165,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
     /*
      * Check that a message published on one transport that has not yet been
      * acked will be republished with the same msgSerial on a new transport (eg
-     * after a resume or an upgrade), before any new messages are send (and
+     * after a resume), before any new messages are send (and
      * without being merged with new messages)
      */
     it('connectionQueuing', function (done) {

test/realtime/event_emitter.test.js

@@ -25,10 +25,7 @@ define(['shared_helper', 'chai'], function (helper, chai) {
      */
     it('attachdetach0', function (done) {
       try {
-        /* Note: realtime now sends an ATTACHED post-upgrade, which can race with
-         * the DETACHED if the DETACH is only sent just after upgrade. Remove
-         * bestTransport with 1.1 spec which has IDs in ATTACHs */
-        var realtime = helper.AblyRealtime({ transports: [helper.bestTransport] }),
+        var realtime = helper.AblyRealtime(),
           index,
           expectedConnectionEvents = ['connecting', 'connected', 'closing', 'closed'],
           expectedChannelEvents = ['attaching', 'attached', 'detaching', 'detached'];

test/realtime/failure.test.js

@@ -60,7 +60,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
             .map(function (transport) {
               return failure_test([transport]);
             })
-            .concat(failure_test(null)), // to test not specifying a transport (so will use upgrade mechanism)
+            .concat(failure_test(null)), // to test not specifying a transport (so will use websocket/base mechanism)
           function (err, realtimes) {
             closeAndFinish(done, realtimes, err);
           },
@@ -98,7 +98,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
             .map(function (transport) {
               return break_test([transport]);
             })
-            .concat(break_test(null)), // to test not specifying a transport (so will use upgrade mechanism)
+            .concat(break_test(null)), // to test not specifying a transport (so will use websocket/base mechanism)
           function (err, realtimes) {
             closeAndFinish(done, realtimes, err);
           },
@@ -127,7 +127,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
                * the suspended timeout trips before three connection cycles */
               disconnectedRetryTimeout: 1000,
               realtimeRequestTimeout: 50,
-              preferenceConnectTimeout: 50,
+              webSocketConnectTimeout: 50,
               suspendedRetryTimeout: 1000,
               connectionStateTtl: 2900,
             });
@@ -173,7 +173,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
             .map(function (transport) {
               return lifecycleTest([transport]);
             })
-            .concat(lifecycleTest(null)), // to test not specifying a transport (so will use upgrade mechanism)
+            .concat(lifecycleTest(null)), // to test not specifying a transport (so will use websocket/base mechanism)
           function (err) {
             if (err) {
               done(err);

test/realtime/init.test.js

@@ -31,7 +31,7 @@ define(['ably', 'shared_helper', 'chai'], function (Ably, helper, chai) {
           realtime.connection.on('connected', function () {
             /* check api version */
             var transport = realtime.connection.connectionManager.activeProtocol.transport;
-            var connectUri = helper.isWebsocket(transport) ? transport.uri : transport.recvRequest.uri;
+            var connectUri = helper.isWebsocket(transport) ? transport.uri : transport.recvRequest.recvUri;
             try {
               expect(connectUri.indexOf('v=3') > -1, 'Check uri includes v=3').to.be.ok;
             } catch (err) {
@@ -309,14 +309,14 @@ define(['ably', 'shared_helper', 'chai'], function (Ably, helper, chai) {
       }
     });
 
-    /* Check base and upgrade transports (nodejs only; browser tests in their own section) */
+    /* Check base and websocket transports (nodejs only; browser tests in their own section) */
     if (!isBrowser) {
       it('node_transports', function (done) {
         var realtime;
         try {
           realtime = helper.AblyRealtime({ transports: helper.availableTransports });
           expect(realtime.connection.connectionManager.baseTransport).to.equal('comet');
-          expect(realtime.connection.connectionManager.upgradeTransports).to.deep.equal(['web_socket']);
+          expect(realtime.connection.connectionManager.webSocketTransportAvailable).to.be.ok;
           closeAndFinish(done, realtime);
         } catch (err) {
           closeAndFinish(done, realtime, err);
@@ -331,9 +331,9 @@ define(['ably', 'shared_helper', 'chai'], function (Ably, helper, chai) {
         var keyStr = helper.getTestApp().keys[0].keyStr;
         var realtime = helper.AblyRealtime({ key: keyStr, useTokenAuth: true });
         realtime.connection.connectionManager.once('transport.pending', function (state) {
-          var transport = realtime.connection.connectionManager.pendingTransports[0],
+          var transport = realtime.connection.connectionManager.pendingTransport,
             originalOnProtocolMessage = transport.onProtocolMessage;
-          realtime.connection.connectionManager.pendingTransports[0].onProtocolMessage = function (message) {
+          realtime.connection.connectionManager.pendingTransport.onProtocolMessage = function (message) {
             try {
               if (message.action === 4) {
                 expect(message.connectionDetails.connectionKey).to.be.ok;
@@ -371,13 +371,14 @@ define(['ably', 'shared_helper', 'chai'], function (Ably, helper, chai) {
       var realtime = helper.AblyRealtime({
         httpMaxRetryCount: 3,
         fallbackHosts: ['a', 'b', 'c'],
+        transport: ['web_socket'],
       });
       realtime.connection.once('connected', function () {
         try {
           var hosts = new Ably.Rest._Http()._getHosts(realtime);
           /* restHost rather than realtimeHost as that's what connectionManager
            * knows about; converted to realtimeHost by the websocketTransport */
-          expect(hosts[0]).to.equal(realtime.options.restHost, 'Check connected realtime host is the first option');
+          expect(hosts[0]).to.equal(realtime.options.realtimeHost, 'Check connected realtime host is the first option');
           expect(hosts.length).to.equal(4, 'Check also have three fallbacks');
         } catch (err) {
           closeAndFinish(done, realtime, err);

test/realtime/presence.test.js

@@ -1708,8 +1708,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
 
     /* Enter ten clients while attaching, finish the attach, check they were all entered correctly */
     it('multiple_pending', function (done) {
-      /* single transport to avoid upgrade stalling due to the stubbed attachImpl */
-      var realtime = helper.AblyRealtime({ transports: [helper.bestTransport] }),
+      var realtime = helper.AblyRealtime(),
         channel = realtime.channels.get('multiple_pending'),
         originalAttachImpl = channel.attachImpl;