Enhancing Subjects with Bidirectional Data Flow and Data Transformation

[javieranguloluna]

Hello RxJS community,

I would like to propose an enhancement for Subjects that would enable bidirectional data flow by introducing a middleware function. Currently, when you create a Subject and apply transformations using pipe and operators like switchMap, the Subject is converted into an Observable, making it unidirectional.

Problem:

In certain scenarios, especially when dealing with components or segments of code that need both data consumption and emission capabilities, this unidirectional flow becomes limiting. The current approach restricts the ability to emit new values from the segment of code where the subscription is made.

Proposal:

I suggest extending the functionality of Subjects by introducing a middleware function. This function would allow for internal data transformations, providing the ability to share the Subject across different parts of the codebase while maintaining bidirectional data flow. The middleware function could be applied during the creation of the Subject, allowing users to define custom transformations.

Example:

// FIRST PART

const mySubject = new Subject<number>()

mySubject.next(1) // Subject creator can emit and recieve raw emitted values

// Case A
// First part usually shares a piped observable to apply transformation to all the subscribers (second parts)

// Case B
// First part shares the full subject so each second part must transform the data, it induces to duplicated code
const myObservable = mySubject.pipe(
    switchMap(n => `Number: ${n}`)
)


// SECOND PARTS

myObservable.subscribe({
    next: (text: string) => console.log(text),
    error: error => console.error(error),
    complete: () => console.info('My function subject is completed!')
})

// Second part cannot emit once pipe was used, is needed to share the subject if you want the second part to emit
myObservable.next(2) // Next method lost!!!




// With FunctionSubject


const myFunctionSubject = new FunctionSubject((n: number) => `Number: ${n}`)


myFunctionSubject.subscribe({
    next: (text: string) => console.log(text),
    error: error => console.error(error),
    complete: () => console.info('My function subject is completed!')
})

myFunctionSubject.next(1) // Output: 'Number: 1'

Benefits:

Enables bidirectional data flow with a single Subject.
Simplifies code by avoiding the need for additional Observables in scenarios where bidirectional flow is required.
I would love to hear the community's thoughts on this proposal. Please share your feedback and suggestions.

Thank you!

[voliva]

Ackchyually, what you are suggesting already works since RxJS v4 to v7: https://codesandbox.io/s/infallible-merkle-3n9kw8?file=/src/index.mjs

This is because the official operators from RxJS actually "lift" the source, so the resulting observable inherits all the methods from the original source. However, this can't be done in Typescript since it can't know if an operator will or will not lift 🏋️, so the typescript definition says that a .pipe chain just returns a new Observable.

Unfortunately (for you), this behaviour is deprecated, and it's removed on v8. Here's the issue that discusses removing it #7201 and here's the PR that removed it on v8 #7202

I personally think that it's better to separate the producers from the consumers, so I agree on the removal of this. Emitting values for the original sources from somewhere else can make it really hard to track why is a value being changed around.

[javieranguloluna]

Thanks for your response @voliva

I think is a good deprecation, in mostly of the cases you don't want the consumers to emit values. The point is that there are some cases in wich you need this behaviour but in a controlled way, heres where the middleware functions do its job. You can share the FunctionSubject only with the parts that need to emit, and share it asObservable with the rest of them.

Furthermore, the middleware function is centraly defined by the creator of the FunctionSubject, so consumers with emision capabilities are, firstly, able to ejecute this function as needed and, secondly restricted by the function internal behaviour

See the case of passing a middleware function that execute an api call, for example for pushing a message to a conversation and updating the chat history. Every subscriber will recieve the whole conversation and, in every emision (a new message) of the allowed consumers the conversation will be updated for all consumers and for itself.

So that way you can achieve this behaviour with only one object and only one function!

You can try it

npm i function-subject