Reverse (input) Pipelines

[dariomannu]

.pipe(), despite not being a pleasure to repeat so many times, still offers an extremely powerful functionaliy, which is transforming the output of an observable with any operator function.

However, I keep missing an "inbound" equivalent, something like a pipeIn() function, that can be used create transformations used to feed a target subject/behavior/observer.

// shared connection point
const connector = new Subject<string>();

// exposed observer interface:
const inputPipe: Observer<string> = pipeIn(
  map((e: PointerEvent) => e.target.value),
)(connector);

// consumer side:
somePointerEventSource.subscribe(inputPipe);

So, anyone feeding the inputPipe above, will indirectly feed strings to the connector, the actual target, despite emitting PointerEvents.

A concrete use case? UI libraries and templates.

import { rml } from 'rimmel';
const emailAddress = new Subject<email>();
const eventToEmail: Observer<Event> = pipeIn(
  map(e => e.target.value
);

// similar to the pipe function (not the method!), which you can use
// to attach as an output transformation pipeline to an observable,
// you could use pipeIn to attach an inbound, push pipeline to an observer
const emit = eventToEmail(emailAddress);

const random = pipeIn(
  generateRandomAddress()
)(emailAddress);

const template = rml`
  <input type="email" onchange="${emit}">
  <button onclick="${random}" type="button"> default </button
`;

Reverse pipeline operator?

In another dimension, if JS had a reverse pipeline operator, too, ( >| ) we could even create reverse pipelines looking something like this:

import { rml } from 'rimmel';
const emailAddress = new Subject<email>();
const evtToEmail = map(e => e.target.value);

const template = rml`
  <input type="email" onchange="${ evtToEmail >| emailAddress }">
`;

In this example, the onchange event, as it emits, would first go through evtToEmail, before being "sourced" into emailAddress, in (IMHO) a simple, clear and elegant way.

This is as opposed to definining emailAddress with a normal pipeline:

import { rml } from 'rimmel';
const evtToEmail = () => map((e: Event) => e.target.value);
const emailAddress = new Subject().pipe(
  evtToEmail(),
);

const template = rml`
  <input type="email" onchange="${ emailAddress }">
`;

Refactoring and organisation

Anyway, back to the real world. Why should we need all this? Can we not do everything already with the standard pipe method and the pipe function? Yes, we can, except we end up having to refactor our main streams when their sources keep changing over time.

Suppose our main stream is a const email = Subject<EmailAddress>, the source can be a text field, a file reader, local storage, web sockets.

With reverse pipelines, we can keep email untouched, including everything downstream and only add some new reverse pipes to convert from different data sources into what email expects (EmailAddress).

Without reverse pipelines, we first need to create a transformation pipe for each source, then redefine the main stream by merging them all with mergeWith().

// -------------------------
// The main stream
const email = new Subject<EmailAddress>();

email.pipe(
  validation(),
  etc(),
).subscribe(doSomething);

// ---------------------------
// The event sources

const emailFromTextInput = pipeIn(
  map((e: Event) => e.target.value)
)(email);

const emailFromOtherDataSource = pipeIn(
  map((xxx: SomeOtherDataFormat) => process(xxx))
)(email);

// ...
const yetAnotherInputSource = pipeIn( ... )(email);

Without input pipelines:

// ---------------------------
// The event sources

const emailFromTextInput = fromEvent(...).pipe(
  map((e: Event) => e.target.value)
);

const emailFromOtherDataSource = otherEventSource.pipe(
  map((xxx: SomeOtherDataFormat) => process(xxx))
);

// ...
const yetAnotherInputSource = new Subject().pipe( ... );

// -------------------------
// The main stream
const email = emailFromTextInput.pipe(
  mergeWith(emailFromOtherDataSource, yetAnotherInputSource),
);

email.pipe(
  validation(),
  etc(),
).subscribe(doSomething);

I feel llike with input pipes it would also be easier to keep the emphasis on a main stream as such. Without, it easily looks like a secondary derived state.

So, in summary, I wonder if the distinction between the two could not only help organise code more effectively, but possibly even help understand the overall flow more easily to readers?

[AronHoxha]

I may have misunderstood the idea, but in general I believe you can achieve feeding values into subjects by using the share operator, so the first example you showed would look something like this.

// shared connection point
const connector = new Subject<string>();
const shared$ = somePointerEventSource.pipe(
 share({
   connector: () => connector,
   // other options depending on use case
 })
)

// consumer side:
shared$.subscribe();

[dariomannu]

Oh, wow... didn't even realise share has options!

I think you did understand the idea, indeed. In other words, it's "event mapping" — the use of the adapter/mediator patterns applied to source streams, to allow low-level UI events (eg MouseEvent) to be translated to the actual format required by the business logic (strings, dates, etc), in a way the two can stay decoupled.

I had another implementation of pipeIn() which looks like this:

const pipeIn = (target, ...pipeline) => {
    const source = new Subject();
    source.pipe(...pipeline).subscribe(target);
    return source;
};

Currently I'm using it as a tool to make it trivial to create input mappers in rimmel.js.
Say we want one that takes a PointerEvent and maps it to an [x, y] tuple:

const ClientXY = target =>
    pipeIn(target,
        map(e => [e.clientX, e.clientY])
    );

share()

So, from my updated understanding, using share() as you suggest could work just as well (possibly even better, as I see there are a few interesting error handling and stream restarting options there, too).

I would still wrap the whole thing inside a utiliy function, like this pipeIn, to make it explicit (conceptually, I'm not quite a fan of the idea of devs using a specific option of the share operator which could just happen to work great for this purpose :))

The point remains, though. Observables have .pipe to create output pipelines, they may as well have a .pipeIn method to create input pipelines to facilitate the creation of input adapters.

Rx8?

I'm a bit less certain about Rx 8+, where streams would look more like the following, what could input pipes look like?

const stream = rx(
 of(something),
 map(whatever),
 filter(anything),
)

Possibly the same pipeIn(stream, ...pipeline) signature could work just as well? Or the curried pipeIn(...pipeline)(stream)?