feat(react): add connectInstance function

The new `connectInstance` function works the same as `connect`, but
takes a function that takes an instance name and produces
`ObservableInput`. This is well suited for namespaced streams.

src/react/connect.spec.ts

@@ -1,8 +1,13 @@
 import testUntyped, { TestFn } from 'ava';
 import { ReactTestRenderer, act, create } from 'react-test-renderer';
 import React from 'react';
-import { BehaviorSubject, EMPTY, Subject, of } from 'rxjs';
-import { NOT_YET_EMITTED, connect, useStream } from './connect';
+import { BehaviorSubject, EMPTY, Observable, Subject, of } from 'rxjs';
+import {
+  NOT_YET_EMITTED,
+  connect,
+  connectInstance,
+  useStream,
+} from './connect';
 import { SinonSpy, spy } from 'sinon';
 import { renderHook } from '../internal/testing/renderHook';
 
@@ -58,13 +63,16 @@ test('useStream unsubscribes on unmount', (t) => {
 });
 
 test('useStream does not resubscribe on rerender', (t) => {
-  const source$ = new Subject<string>();
-  const { rerender } = renderHook(() => useStream(source$));
+  let subscriptions = 0;
+  const source$ = new Observable<string>((obs) => {
+    subscriptions++;
+    obs.next('hello');
+  });
 
-  const subscription = source$.observers[0];
+  const { rerender } = renderHook(() => useStream(source$));
   rerender();
 
-  t.assert(source$.observers[0] === subscription);
+  t.deepEqual(subscriptions, 1);
 });
 
 test('useStream unsubscribes, keeps latest value and subscribes new stream', (t) => {
@@ -195,3 +203,28 @@ test('connect should propagate changes to forwarded props', async (t) => {
     ...initialProps,
   });
 });
+
+test('connectInstance should provide unique instance Id', (t) => {
+  const { WrappedComponentSpy } = t.context;
+  let instance1: string | null = null;
+  let instance2: string | null = null;
+  const HOComponent1 = connectInstance(WrappedComponentSpy, (msg) => {
+    instance1 = msg;
+    return of({ msg });
+  });
+  const HOComponent2 = connectInstance(WrappedComponentSpy, (msg) => {
+    instance2 = msg;
+    return of({ msg });
+  });
+
+  act(() => {
+    create(React.createElement(HOComponent1));
+    create(React.createElement(HOComponent2));
+  });
+
+  t.truthy(instance1);
+  t.truthy(instance2);
+  t.not(instance1, instance2);
+  t.deepEqual(WrappedComponentSpy.firstCall.args[0], { msg: instance1 });
+  t.deepEqual(WrappedComponentSpy.secondCall.args[0], { msg: instance2 });
+});

src/react/connect.ts

@@ -45,12 +45,14 @@ export const useStream = <T>(
  * provided by the stream. In these cases the values from the stream will
  * override any values passed as props.
  *
- * @param WrappedComponent Component that will recieve props from the stream
+ * @param Component Component that will recieve props from the stream
  * @param stream$ Stream that will provide props to the component
- * @see useStream
+ * @returns A component that renders the original component with props retrieved
+ *  from the stream
+ * @see {@link useStream}
  */
 export const connect =
-  <Props extends {}, Observed extends Partial<Props>>( // eslint-disable-line @typescript-eslint/ban-types
+  <Props extends Record<string, unknown>, Observed extends Partial<Props>>(
     Component: ComponentType<Props>,
     stream$: ObservableInput<Observed>
   ) =>
@@ -65,3 +67,38 @@ export const connect =
 
     return React.createElement(Component, newProps);
   };
+
+let instanceCount = 0;
+const nextInstanceName = () => `view-${instanceCount++}`;
+
+/**
+ * Higher order component for connecting a component to a stream
+ *
+ * This is the same as {@link connect }, but takes a factory for the stream. A
+ * new stream will be retrieved from the factory for every new mount of the
+ * component.
+ *
+ * @param Component Component that will receive props from the stream
+ * @param createInstanceStream Function to create the stream for each instance
+ *  of the component
+ * @returns A component that renders the original component with props retrieved
+ *  from the stream
+ * @see {@link connect}
+ */
+export const connectInstance =
+  <Props extends Record<string, unknown>, Observed extends Partial<Props>>(
+    Component: ComponentType<Props>,
+    createInstanceStream: (instance: string) => ObservableInput<Observed>
+  ) =>
+  (props: Omit<Props, keyof Observed>) => {
+    const [stream$] = useState(() => createInstanceStream(nextInstanceName()));
+    const value = useStream(stream$);
+
+    if (value === NOT_YET_EMITTED) return null;
+
+    // Typescript doesn't recognize this as Observed & Props for some reason
+    // Question on StackOverflow: https://stackoverflow.com/q/60758084/1104307
+    const newProps = { ...props, ...value } as Props & Observed;
+
+    return React.createElement(Component, newProps);
+  };

src/react/index.ts

@@ -1 +1 @@
-export { connect, useStream } from './connect';
+export { connect, connectInstance, useStream } from './connect';