The current pipable variant of this operator is called finalize() (since RxJS 6). The older and now deprecated patch operator was called finally() (until RxJS 5.5).

I think finalize() operator is actually correct. You say:

do that logic only when I subscribe, and after the stream has ended

which is not a problem I think. You can have a single source and use finalize() before subscribing to it if you want. This way you're not required to always use finalize():

let source = new Observable(observer => {

  observer.next(1);

  observer.error('error message');

  observer.next(3);

  observer.complete();

}).pipe(

  publish(),

);



source.pipe(

  finalize(() => console.log('Finally callback')),

).subscribe(

  value => console.log('#1 Next:', value),

  error => console.log('#1 Error:', error),

  () => console.log('#1 Complete')

);



source.subscribe(

  value => console.log('#2 Next:', value),

  error => console.log('#2 Error:', error),

  () => console.log('#2 Complete')

);



source.connect();

This prints to console:

#1 Next: 1

#2 Next: 1

#1 Error: error message

Finally callback

#2 Error: error message

Jan 2019: Updated for RxJS 6