Wrap operator in RxJS so it can be applied on materialized stream

user_3887009 0
Viewed 465

I'm looking for a way to trace value 'route' among stream operators. I have a materialized stream with additional metadata on Notification object (ex. valueId property). It's definition will look something like this:

const x = stream.pipe(
  materialize(),
  map(x => Object.assign(x, {valueId: randomInt()}))
);

Now I need to wrap operators that are applied to x. Let's say I need to use map(x => x * 2), but I cannot do it like this:

x.pipe(dematerialize(), map(x => x * 2))

Because I will lose my metadata. How do I make a wrap function, that will apply to any operator and will still preserve my additional metadata?

x.pipe(wrap(map(x => x * 2)))

I thought about something like this:

function wrap<T, R>(
  operator: OperatorFunction<T, R>
): (source: Observable<TaggedValue<T>>) => Observable<TaggedValue<R>> {
  return source =>
    source.pipe(
      switchMap(x =>
        of(x).pipe(
          dematerialize(),
          operator,
          materialize(),
          map(z => Object.assign(z, { valueId: x.valueId }))
        )
      )
    );
}

But it generates fake complete messages from of(). Sample: https://stackblitz.com/edit/rxjs-fhv54p

reactivex rxjs
3 Answers

The issue in your approach using of is that the complete notifications of of are materialized and passed as value to the observer of source.

Try this:

function wrap<T, R>(op: OperatorFunction<T, R>):
  (source: Observable<any>) => Observable<any> {
  return source => {
    return source.pipe(
      switchMap(x => of(x)
        .pipe(
          dematerialize(),
          op,
          map(y => ({value: y, uuid: x.uuid}))
        )
      ),
      materialize(),
      map((x: any) => Object.assign(x, {
        value: x.value ? x.value.value : undefined,
        uuid: x.value ? x.value.uuid: undefined
      })),
    )
  }
}

Demo: https://stackblitz.com/edit/rxjs-gxj7zl

user_46110680

You can do something like this

const stream$ = stream$.pipe(map((data=> ({...data, x: data.x + 1}))))

Or move it to a wrap function

const mapProp = (propName, fn) => stream$.pipe(map((data=> ({...data, [propName]: fn(data[propName])}))))

//then
const stream$ = stream$.pipe(mapProp('x', x => x+1 ))

if you want to use it for something else but map

const mapProp = (propName, fn) => stream$.pipe(
    mergeMap(data =>
      of(data[propName])
        .pipe(
           fn,
           map(newPropValue => ({ ...data, propName: newPropValue })
        )
    )
)

//Usage 
const stream$ = stream$.pipe(mapProp('x', map(x => x+1)))
user_26399250

For now I've came up with an idea that "preserves" data using wrap function state. I don't really like it's 'side effect' nature, although it's safest implementation I've found so far.

export function wrap<T, R>(
  operator: OperatorFunction<T, R>,
) => (source: Observable<TaggedValue<T>>) => Observable<TaggedValue<R>> {
  return source => {
    let metadata: { stepId: number; streamId: number; valueId: number };

    return source.pipe(
      tap(
        ({ valueId, stepId, streamId }) =>
          (metadata = { valueId, streamId, stepId: stepId + 1 }),
      ),
      dematerialize(),
      operator,
      materialize(),
      map(x => Object.assign(x, metadata, { timestamp: Date.now() })),
    );
  };
}

But it's not working, because metadata is getting overridden randomly.

user_38870090
Related