refactor(StreamProvider): remove unnecessary stream multiplexing

CHANGELOG.md

@@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- **BREAKING**: `StreamProvider` no longer accepts a `jsonRpcStreamName` parameter ([#400](https://github.com/MetaMask/providers/pull/400))
+  - Previously, this parameter was used internally to create an ObjectMultiplex stream and substream for JSON-RPC communication
+  - Now, the consumer is responsible for creating and managing the stream multiplexing if needed
+  - The provider will use the provided stream connection directly without any multiplexing
+- **BREAKING**: `MetaMaskInpageProvider` no longer accepts a `jsonRpcStreamName` parameter ([#400](https://github.com/MetaMask/providers/pull/400))
+  - This change is inherited from StreamProvider, as MetaMaskInpageProvider extends StreamProvider
+  - Stream multiplexing should be handled before provider instantiation
+- `initializeInpageProvider` now handles stream multiplexing internally ([#400](https://github.com/MetaMask/providers/pull/400))
+  - Creates an ObjectMultiplex instance and substream using the provided `jsonRpcStreamName`
+  - This maintains backwards compatibility for consumers using `initializeInpageProvider`
+- `createExternalExtensionProvider` now handles stream multiplexing internally ([#400](https://github.com/MetaMask/providers/pull/400))
+  - Creates an ObjectMultiplex instance and substream for JSON-RPC communication
+  - This maintains backwards compatibility for consumers using `createExternalExtensionProvider`
+
 ## [18.3.1]
 
 ### Changed

jest.config.js

@@ -45,10 +45,10 @@ const baseConfig = {
   // An object that configures minimum threshold enforcement for coverage results
   coverageThreshold: {
     global: {
-      branches: 67.6,
-      functions: 69.91,
-      lines: 69.51,
-      statements: 69.52,
+      branches: 66.79,
+      functions: 68.69,
+      lines: 68.35,
+      statements: 68.38,
     },
   },
 

src/MetaMaskInpageProvider.test.ts

@@ -1,4 +1,6 @@
+import ObjectMultiplex from '@metamask/object-multiplex';
 import type { JsonRpcRequest } from '@metamask/utils';
+import { pipeline } from 'readable-stream';
 
 import messages from './messages';
 import {
@@ -64,14 +66,14 @@ async function getInitializedProvider({
   const onWrite = jest.fn();
   const connectionStream = new MockConnectionStream((name, data) => {
     if (
-      name === 'metamask-provider' &&
+      name === MetaMaskInpageProviderStreamName &&
       data.method === 'metamask_getProviderState'
     ) {
       // Wrap in `setTimeout` to ensure a reply is received by the provider
       // after the provider has processed the request, to ensure that the
       // provider recognizes the id.
       setTimeout(() =>
-        connectionStream.reply('metamask-provider', {
+        connectionStream.reply(MetaMaskInpageProviderStreamName, {
           id: onWrite.mock.calls[0][1].id,
           jsonrpc: '2.0',
           result: {
@@ -93,8 +95,13 @@ async function getInitializedProvider({
     }
     onWrite(name, data);
   });
-
-  const provider = new MetaMaskInpageProvider(connectionStream);
+  const mux = new ObjectMultiplex();
+  pipeline(connectionStream, mux, connectionStream, (error: Error | null) => {
+    console.error(error);
+  });
+  const provider = new MetaMaskInpageProvider(
+    mux.createStream(MetaMaskInpageProviderStreamName),
+  );
   await new Promise<void>((resolve: () => void) => {
     provider.on('_initialized', resolve);
   });

src/MetaMaskInpageProvider.ts

@@ -29,9 +29,7 @@ export type MetaMaskInpageProviderOptions = {
    * Whether the provider should send page metadata.
    */
   shouldSendMetadata?: boolean;
-
-  jsonRpcStreamName?: string | undefined;
-} & Partial<Omit<StreamProviderOptions, 'rpcMiddleware' | 'jsonRpcStreamName'>>;
+} & Partial<Omit<StreamProviderOptions, 'rpcMiddleware'>>;
 
 type SentWarningsState = {
   // methods
@@ -86,8 +84,6 @@ export class MetaMaskInpageProvider extends AbstractStreamProvider {
    *
    * @param connectionStream - A Node.js duplex stream.
    * @param options - An options bag.
-   * @param options.jsonRpcStreamName - The name of the internal JSON-RPC stream.
-   * Default: `metamask-provider`.
    * @param options.logger - The logging API to use. Default: `console`.
    * @param options.maxEventListeners - The maximum number of event
    * listeners. Default: 100.
@@ -97,14 +93,12 @@ export class MetaMaskInpageProvider extends AbstractStreamProvider {
   constructor(
     connectionStream: Duplex,
     {
-      jsonRpcStreamName = MetaMaskInpageProviderStreamName,
       logger = console,
       maxEventListeners = 100,
       shouldSendMetadata,
     }: MetaMaskInpageProviderOptions = {},
   ) {
     super(connectionStream, {
-      jsonRpcStreamName,
       logger,
       maxEventListeners,
       rpcMiddleware: getDefaultExternalMiddleware(logger),

src/StreamProvider.test.ts

@@ -1,5 +1,7 @@
 import type { JsonRpcMiddleware } from '@metamask/json-rpc-engine';
+import ObjectMultiplex from '@metamask/object-multiplex';
 import type { Json, JsonRpcParams } from '@metamask/utils';
+import { pipeline } from 'readable-stream';
 
 import messages from './messages';
 import { StreamProvider } from './StreamProvider';
@@ -20,7 +22,6 @@ function getStreamProvider(
 ) {
   const mockStream = new MockConnectionStream();
   const streamProvider = new StreamProvider(mockStream, {
-    jsonRpcStreamName: mockStreamName,
     rpcMiddleware,
   });
 
@@ -38,9 +39,7 @@ describe('StreamProvider', () => {
       const networkVersion = '1';
       const isUnlocked = true;
 
-      const streamProvider = new StreamProvider(new MockConnectionStream(), {
-        jsonRpcStreamName: mockStreamName,
-      });
+      const streamProvider = new StreamProvider(new MockConnectionStream());
 
       const requestMock = jest
         .spyOn(streamProvider, 'request')
@@ -370,10 +369,13 @@ describe('StreamProvider', () => {
     describe('events', () => {
       it('calls chainChanged when the chainId changes', async () => {
         const mockStream = new MockConnectionStream();
-        const streamProvider = new StreamProvider(mockStream, {
-          jsonRpcStreamName: mockStreamName,
+        const mux = new ObjectMultiplex();
+        pipeline(mockStream, mux, mockStream, (error: Error | null) => {
+          console.error(error);
         });
-
+        const streamProvider = new StreamProvider(
+          mux.createStream(mockStreamName),
+        );
         const requestMock = jest
           .spyOn(streamProvider, 'request')
           .mockImplementationOnce(async () => {
@@ -404,9 +406,13 @@ describe('StreamProvider', () => {
 
       it('handles chain changes with intermittent disconnection', async () => {
         const mockStream = new MockConnectionStream();
-        const streamProvider = new StreamProvider(mockStream, {
-          jsonRpcStreamName: mockStreamName,
+        const mux = new ObjectMultiplex();
+        pipeline(mockStream, mux, mockStream, (error: Error | null) => {
+          console.error(error);
         });
+        const streamProvider = new StreamProvider(
+          mux.createStream(mockStreamName),
+        );
 
         const requestMock = jest
           .spyOn(streamProvider, 'request')

src/StreamProvider.ts

@@ -1,6 +1,5 @@
 import type { JsonRpcMiddleware } from '@metamask/json-rpc-engine';
 import { createStreamMiddleware } from '@metamask/json-rpc-middleware-stream';
-import ObjectMultiplex from '@metamask/object-multiplex';
 import type SafeEventEmitter from '@metamask/safe-event-emitter';
 import type { Json, JsonRpcParams } from '@metamask/utils';
 import { duplex as isDuplex } from 'is-stream';
@@ -16,12 +15,7 @@ import {
   isValidNetworkVersion,
 } from './utils';
 
-export type StreamProviderOptions = {
-  /**
-   * The name of the stream used to connect to the wallet.
-   */
-  jsonRpcStreamName: string;
-} & BaseProviderOptions;
+export type StreamProviderOptions = BaseProviderOptions;
 
 export type JsonRpcConnection = {
   events: SafeEventEmitter;
@@ -43,7 +37,6 @@ export abstract class AbstractStreamProvider extends BaseProvider {
    *
    * @param connectionStream - A Node.js duplex stream.
    * @param options - An options bag.
-   * @param options.jsonRpcStreamName - The name of the internal JSON-RPC stream.
    * @param options.logger - The logging API to use. Default: `console`.
    * @param options.maxEventListeners - The maximum number of event
    * listeners. Default: 100.
@@ -52,11 +45,10 @@ export abstract class AbstractStreamProvider extends BaseProvider {
   constructor(
     connectionStream: Duplex,
     {
-      jsonRpcStreamName,
       logger = console,
       maxEventListeners = 100,
       rpcMiddleware = [],
-    }: StreamProviderOptions,
+    }: StreamProviderOptions = {},
   ) {
     super({ logger, maxEventListeners, rpcMiddleware });
 
@@ -67,15 +59,6 @@ export abstract class AbstractStreamProvider extends BaseProvider {
     // Bind functions to prevent consumers from making unbound calls
     this._handleStreamDisconnect = this._handleStreamDisconnect.bind(this);
 
-    // Set up connectionStream multiplexing
-    const mux = new ObjectMultiplex();
-    pipeline(
-      connectionStream,
-      mux as unknown as Duplex,
-      connectionStream,
-      this._handleStreamDisconnect.bind(this, 'MetaMask'),
-    );
-
     // Set up RPC connection
     // Typecast: The type of `Duplex` is incompatible with the type of
     // `JsonRpcConnection`.
@@ -84,9 +67,9 @@ export abstract class AbstractStreamProvider extends BaseProvider {
     }) as unknown as JsonRpcConnection;
 
     pipeline(
+      connectionStream,
       this._jsonRpcConnection.stream,
-      mux.createStream(jsonRpcStreamName) as unknown as Duplex,
-      this._jsonRpcConnection.stream,
+      connectionStream,
       this._handleStreamDisconnect.bind(this, 'MetaMask RpcProvider'),
     );
 

src/extension-provider/createExternalExtensionProvider.ts

@@ -1,6 +1,7 @@
+import ObjectMultiplex from '@metamask/object-multiplex';
 import { detect } from 'detect-browser';
 import { PortDuplexStream as PortStream } from 'extension-port-stream';
-import type { Duplex } from 'readable-stream';
+import { pipeline } from 'readable-stream';
 import type { Runtime } from 'webextension-polyfill';
 
 import config from './external-extension-config.json';
@@ -28,8 +29,16 @@ export function createExternalExtensionProvider(
     const metamaskPort = chrome.runtime.connect(extensionId) as Runtime.Port;
 
     const pluginStream = new PortStream(metamaskPort);
-    provider = new StreamProvider(pluginStream as unknown as Duplex, {
-      jsonRpcStreamName: MetaMaskInpageProviderStreamName,
+    const streamName = MetaMaskInpageProviderStreamName;
+    const mux = new ObjectMultiplex();
+    pipeline(pluginStream, mux, pluginStream, (error: Error | null) => {
+      let warningMsg = `Lost connection to "${streamName}".`;
+      if (error?.stack) {
+        warningMsg += `\n${error.stack}`;
+      }
+      console.warn(warningMsg);
+    });
+    provider = new StreamProvider(mux.createStream(streamName), {
       logger: console,
       rpcMiddleware: getDefaultExternalMiddleware(console),
     });

src/initializeInpageProvider.ts

@@ -1,11 +1,15 @@
-import type { Duplex } from 'readable-stream';
+import ObjectMultiplex from '@metamask/object-multiplex';
+import { type Duplex, pipeline } from 'readable-stream';
 
 import type { CAIP294WalletData } from './CAIP294';
 import { announceWallet } from './CAIP294';
 import { announceProvider as announceEip6963Provider } from './EIP6963';
 import { getBuildType } from './extension-provider/createExternalExtensionProvider';
 import type { MetaMaskInpageProviderOptions } from './MetaMaskInpageProvider';
-import { MetaMaskInpageProvider } from './MetaMaskInpageProvider';
+import {
+  MetaMaskInpageProvider,
+  MetaMaskInpageProviderStreamName,
+} from './MetaMaskInpageProvider';
 import { shimWeb3 } from './shimWeb3';
 import type { BaseProviderInfo } from './types';
 
@@ -29,6 +33,10 @@ type InitializeProviderOptions = {
    * Whether the window.web3 shim should be set.
    */
   shouldShimWeb3?: boolean;
+  /**
+   * The name of the stream used to connect to the wallet.
+   */
+  jsonRpcStreamName?: string;
 } & MetaMaskInpageProviderOptions;
 
 /**
@@ -47,20 +55,30 @@ type InitializeProviderOptions = {
  */
 export function initializeProvider({
   connectionStream,
-  jsonRpcStreamName,
+  jsonRpcStreamName = MetaMaskInpageProviderStreamName,
   logger = console,
   maxEventListeners = 100,
   providerInfo,
   shouldSendMetadata = true,
   shouldSetOnWindow = true,
   shouldShimWeb3 = false,
 }: InitializeProviderOptions): MetaMaskInpageProvider {
-  const provider = new MetaMaskInpageProvider(connectionStream, {
-    jsonRpcStreamName,
-    logger,
-    maxEventListeners,
-    shouldSendMetadata,
+  const mux = new ObjectMultiplex();
+  pipeline(connectionStream, mux, connectionStream, (error: Error | null) => {
+    let warningMsg = `Lost connection to "${jsonRpcStreamName}".`;
+    if (error?.stack) {
+      warningMsg += `\n${error.stack}`;
+    }
+    console.warn(warningMsg);
   });
+  const provider = new MetaMaskInpageProvider(
+    mux.createStream(jsonRpcStreamName),
+    {
+      logger,
+      maxEventListeners,
+      shouldSendMetadata,
+    },
+  );
 
   const proxiedProvider = new Proxy(provider, {
     // some common libraries, e.g. web3@1.x, mess with our API