Add chains of observables

src/API.ts

@@ -321,11 +321,23 @@ export interface ContributeAPIOptions<TAPI> {
 
 export type StateObserverUnsubscribe = () => void
 export type StateObserver<TSelectorAPI> = (next: TSelectorAPI) => void
+export interface StateObserverSubscribeResult<TSelectorAPI> {
+    currentSelector: TSelectorAPI
+    unsubscribe: StateObserverUnsubscribe
+}
 export interface ObservableState<TSelectorAPI> {
-    subscribe(fromShell: Shell, callback: StateObserver<TSelectorAPI>): StateObserverUnsubscribe
-    current(): TSelectorAPI
+    subscribe(fromShell: Shell, callback: StateObserver<TSelectorAPI>): StateObserverSubscribeResult<TSelectorAPI>
+}
+
+export interface ObservablesMap {
+    [key: string]: ObservableState<any>
+}
+
+export type ObservedSelectorsMap<M extends ObservablesMap> = {
+    [K in keyof M]: M[K] extends ObservableState<infer S> ? S : undefined
 }
 
+
 export type AnyFunction = (...args: any[]) => any
 export type FunctionWithSameArgs<F extends AnyFunction> = (...args: Parameters<F>) => any
 
@@ -405,17 +417,33 @@ export interface Shell extends Pick<AppHost, Exclude<keyof AppHost, 'getStore' |
      * Contribute a Redux reducer that will be added to the host store
      * Use it for rapidly changing state (e.g. changing on every mouse movement event)
      * Changes to this state won't trigger the usual subscribers.
-     * In order to subscribe to changes in this state, use the observer object returned by this function.
+     * In order to subscribe to changes in this state, use the observable object returned by this function.
      *
      * @template TState
      * @param {ReducersMapObjectContributor<TState>} contributor
-     * @return {TAPI} Observer object for subscribing to state changes. The observer can also be passed to {connectWithShell}.
+     * @return {ObservableState<TSelector>} Observer object for subscribing to state changes. The observer can also be passed to {connectWithShell}.
      */
     contributeObservableState<TState, TSelector, TAction extends Redux.AnyAction = Redux.AnyAction>(
         contributor: ReducersMapObjectContributor<TState, TAction>,
         selectorFactory: (state: TState) => TSelector
     ): ObservableState<TSelector>
 
+    /**
+     * Create an observable which in turn uses other observable(s) as the source data of its state
+     *
+     * @template M an object type whose properties contain source observables
+     * @template TSelector the result ("selectors object") of this observable
+     * @param sourceSelectors An object type whose properties contain source observables
+     * @return {ObservableState<TSelector>} The resulting observable object which notifies whenever one of the source observables is changed
+     */
+    createChainedObservable<
+        M extends ObservablesMap, 
+        TSelector
+    >(
+        sources: M, 
+        selectorFactory: (sourceSelectors: ObservedSelectorsMap<M>) => TSelector
+    ): ObservableState<TSelector>
+
     /**
      * Contribute the main view (root) of the application
      * Intended to be used by a single {Shell} in an application

src/appHost.ts

@@ -28,7 +28,9 @@ import {
     APILayer,
     CustomExtensionSlotHandler,
     CustomExtensionSlot,
-    ObservableState
+    ObservableState,
+    ObservablesMap,
+    ObservedSelectorsMap,
 } from './API'
 import _ from 'lodash'
 import { AppHostAPI, AppHostServicesProvider, createAppHostServicesEntryPoint } from './appHostServices'
@@ -41,7 +43,8 @@ import {
     PrivateThrottledStore,
     StateContribution,
     ThrottledStore,
-    updateThrottledStore
+    updateThrottledStore,
+    createChainedObservable
 } from './throttledStore'
 import { ConsoleHostLogger, createShellLogger } from './loggers'
 import { monitorAPI } from './monitorAPI'
@@ -846,6 +849,14 @@ miss: ${memoizedWithMissHit.miss}
                 return observable
             },
 
+            createChainedObservable<M extends ObservablesMap, TSelector>(
+                sources: M, 
+                selectorFactory: (sourceSelectors: ObservedSelectorsMap<M>) => TSelector
+            ) {
+                const observableUniqueName = `${entryPoint.name}/observable_${nextObservableId++}`
+                return createChainedObservable(shell,observableUniqueName, sources, selectorFactory)
+            },
+
             getStore<TState>(): ScopedStore<TState> {
                 return {
                     dispatch: host.getStore().dispatch,

src/connectWithShell.tsx

@@ -2,11 +2,12 @@ import _ from 'lodash'
 import React from 'react'
 import { connect as reduxConnect, Options as ReduxConnectOptions } from 'react-redux'
 import { Action, Dispatch } from 'redux'
-import { Shell, AnyFunction, ObservableState, StateObserverUnsubscribe } from './API'
+import { Shell, AnyFunction, StateObserverUnsubscribe, ObservablesMap, ObservedSelectorsMap } from './API'
 import { ErrorBoundary } from './errorBoundary'
 import { ShellContext } from './shellContext'
 import { StoreContext } from './storeContext'
 import { propsDeepEqual } from './propsDeepEqual'
+import { subscribeToObservables } from './throttledStore'
 
 interface WrapperMembers<S, OP, SP, DP> {
     connectedComponent: any
@@ -148,24 +149,8 @@ export function connectWithShell<S = {}, OP = {}, SP = {}, DP = {}>(
     }
 }
 
-export interface ObservablesMap {
-    [key: string]: ObservableState<any>
-}
-
-export type ObservedSelectorsMap<M> = {
-    [K in keyof M]: M[K] extends ObservableState<infer S> ? S : undefined
-}
-
 export type OmitObservedSelectors<T, M> = Omit<T, keyof M>
 
-export function mapObservablesToSelectors<M extends ObservablesMap>(map: M): ObservedSelectorsMap<M> {
-    const result = _.mapValues(map, observable => {
-        const selector = observable.current()
-        return selector
-    })
-    return result
-}
-
 export function observeWithShell<OM extends ObservablesMap, OP extends ObservedSelectorsMap<OM>>(
     observables: OM,
     boundShell: Shell
@@ -186,17 +171,18 @@ export function observeWithShell<OM extends ObservablesMap, OP extends ObservedS
                     super(props)
                     this.connectedComponent = innerFactory(pureComponent)
                     this.unsubscribes = []
-                    this.state = mapObservablesToSelectors(observables)
                 }
 
                 public componentDidMount() {
-                    for (const key in observables) {
-                        const unsubscribe = observables[key].subscribe(boundShell, () => {
-                            const newState = mapObservablesToSelectors(observables)
-                            this.setState(newState)
-                        })
-                        this.unsubscribes.push(unsubscribe)
-                    }
+                    const { unsubscribes, selectors } = subscribeToObservables(
+                        boundShell, 
+                        observables, 
+                        () => this.state, 
+                        (newState) => this.setState(newState)
+                    )
+
+                    this.unsubscribes = unsubscribes
+                    this.state = selectors
                 }
 
                 public componentWillUnmount() {
@@ -205,17 +191,21 @@ export function observeWithShell<OM extends ObservablesMap, OP extends ObservedS
                 }
 
                 public render() {
+                    if (!this.state) {
+                        return null
+                    }
+
                     const ConnectedComponent = this.connectedComponent
                     const connectedComponentProps: OP = {
-                        ...this.props, // OP excluding observed selectors
+                        ...this.props, // own props excluding observed selectors
                         ...this.state // observed selectors
                     } as OP // TypeScript doesn't get it
                     return <ConnectedComponent {...connectedComponentProps} />
                 }
             }
 
             const hoc = (props: WithChildren<OmitObservedSelectors<OP, OM>>) => {
-                return <ObservableWrapperComponent {...props} {...mapObservablesToSelectors(observables)} />
+                return <ObservableWrapperComponent {...props} />
             }
 
             return hoc

src/throttledStore.tsx

@@ -2,7 +2,7 @@ import { Reducer, createStore, Store, ReducersMapObject, combineReducers, AnyAct
 import { devToolsEnhancer } from 'redux-devtools-extension'
 import { AppHostServicesProvider } from './appHostServices'
 import _ from 'lodash'
-import { AppHost, ExtensionSlot, ReducersMapObjectContributor, ObservableState, StateObserver, Shell, SlotKey } from './API'
+import { AppHost, ExtensionSlot, ReducersMapObjectContributor, ObservableState, StateObserver, Shell, SlotKey, ObservablesMap, ObservedSelectorsMap, StateObserverUnsubscribe } from './API'
 import { contributeInstalledShellsState } from './installedShellsState'
 import { interceptAnyObject } from './interceptAnyObject'
 import { invokeSlotCallbacks } from './invokeSlotCallbacks'
@@ -248,15 +248,93 @@ export const createObservable = <TState, TSelector>(
     return {
         subscribe(fromShell, callback) {
             observersSlot.contribute(fromShell, callback)
-            return () => {
-                observersSlot.discardBy(item => item.contribution === callback)
+            return {
+                currentSelector: getOrCreateCachedSelector(),
+                unsubscribe: () => {
+                    observersSlot.discardBy(item => item.contribution === callback)
+                }
             }
         },
         notify() {
             cachedSelector = undefined
             const newSelector = getOrCreateCachedSelector()
             invokeSlotCallbacks(observersSlot, newSelector)
         },
-        current: getOrCreateCachedSelector
     }
 }
+
+export interface SubscribeToObservablesResult<M extends ObservablesMap> {
+    selectors: ObservedSelectorsMap<M>
+    unsubscribes: StateObserverUnsubscribe[]
+}
+
+export function subscribeToObservables<M extends ObservablesMap>(
+    shell: Shell, 
+    observablesMap: M, 
+    getState: () => ObservedSelectorsMap<M>,
+    setState: (newState: ObservedSelectorsMap<M>) => void,
+): SubscribeToObservablesResult<M> {
+    const unsubscribes: StateObserverUnsubscribe[] = [];
+    const selectors = _.mapValues(observablesMap, (observable, key) => {
+        const { currentSelector, unsubscribe } = observable.subscribe(shell, newSelector => {
+            const nextState = {
+                ...getState(),
+                [key]: newSelector
+            };
+            setState(nextState)
+        });
+        unsubscribes.push(unsubscribe)
+        return currentSelector
+    })
+    return {
+        selectors,
+        unsubscribes
+    }
+}
+
+export function createChainedObservable<
+    M extends ObservablesMap, 
+    TSelector
+>(
+    shell: Shell, 
+    uniqueName: string,
+    sources: M, 
+    selectorFactory: (sourceSelectors: ObservedSelectorsMap<M>) => TSelector
+): ObservableState<TSelector> {
+    const subscribersSlotKey: SlotKey<StateObserver<TSelector>> = {
+        name: uniqueName
+    }
+    const observersSlot = shell.declareSlot(subscribersSlotKey)
+
+    let targetSelector: TSelector | undefined = undefined
+    let lastObservedSourceSelectors: ObservedSelectorsMap<M> | undefined = undefined
+
+    const subscription = subscribeToObservables(
+        shell, 
+        sources, 
+        () => {
+            return lastObservedSourceSelectors!
+        },
+        (newObservedSelectors) => {
+            lastObservedSourceSelectors = newObservedSelectors
+            targetSelector = selectorFactory(newObservedSelectors)
+            invokeSlotCallbacks(observersSlot, targetSelector)
+        }
+    )
+
+    lastObservedSourceSelectors = subscription.selectors
+    targetSelector = selectorFactory(lastObservedSourceSelectors)
+
+    return {
+        subscribe(fromShell, callback) {
+            observersSlot.contribute(fromShell, callback)
+            return {
+                currentSelector: targetSelector!,
+                unsubscribe: () => {
+                    observersSlot.discardBy(item => item.contribution === callback)
+                }
+            }
+        },
+    }
+}
+

testKit/index.tsx

@@ -1,7 +1,7 @@
 import { mount, ReactWrapper } from 'enzyme'
 import _ from 'lodash'
 import React, { ReactElement } from 'react'
-import { EntryPoint, ObservableState, PrivateShell, ShellBoundaryAspect } from '../src/API'
+import { EntryPoint, ObservablesMap, ObservableState, PrivateShell, ShellBoundaryAspect } from '../src/API'
 import { AnySlotKey, AppHost, AppMainView, createAppHost as _createAppHost, EntryPointOrPackage, Shell, SlotKey } from '../src/index'
 import { ShellRenderer } from '../src/renderSlotComponents'
 import { createShellLogger } from '../src/loggers'
@@ -190,6 +190,7 @@ function createShell(host: AppHost): PrivateShell {
         },
         contributeState: _.noop,
         contributeObservableState: <TState, TSelectors, TAction>() => mockObservable<TSelectors>(undefined as any),
+        createChainedObservable: <M extends ObservablesMap, TSelector>() => mockObservable<TSelector>(undefined as any),
         contributeMainView: _.noop,
         flushMemoizedForState: _.noop,
         memoizeForState: _.identity,
@@ -203,10 +204,10 @@ function createShell(host: AppHost): PrivateShell {
 export function mockObservable<T>(value: T): ObservableState<T> {
     return {
         subscribe: () => {
-            return () => {}
+            return { 
+                unsubscribe: () => {},
+                currentSelector: value
+            }
         },
-        current() {
-            return value
-        }
     }
 }