caroso1222

import { log } from '/path/to/utils/operators';

const source = of(1, 2, 3, 4);

source.pipe(
  log(),
  map(v => v ** 2),
  log(),
  filter(v => v > 5),
  log()

caroso1222

export function log() {
  return function logFn<T>(source: Observable<T>) {
    const output = new Observable<T>(observer => {
      const subscription = source.subscribe(
        val => {
          console.log(val);
          observer.next(val);
        },
        err => {
          console.error(err);

caroso1222

export function log() {
  return function logFn<T>(source: Observable<T>) {
    const output = new Observable<T>(observer => {
      source.subscribe(
        val => {
          console.log(val);
          observer.next(val);
        },
        err => {
          console.error(err);

caroso1222

export function log() {
  return function logFn<T>(source: Observable<T>) {
    const output = new Observable<T>(observer => {
      source.subscribe(
        val => {
          console.log(val);
          observer.next(val);
        },
        err => console.error(err),
        () => console.log('%ccomplete', 'color: green')

caroso1222

export function log() {
  return function logFn<T>(source: Observable<T>) {
    const output = new Observable<T>(observer => {
      source.subscribe(
        val => console.log(val),
        err => console.error(err),
        () => console.log('%ccomplete', 'color: green')
      );
    });
    return output;

caroso1222

export function log() {
  return function logFn<T>(source: Observable<T>) {
    const output = new Observable<T>(observer => {});
    return output;
  };
}

caroso1222

export function log() {
  return function logFn() {};
}

caroso1222

const source = of('World');

// with tap
source.pipe(
  tap(x => console.log(x))
).subscribe(x => {
  this.doSomething(x);
});

// with map

caroso1222

import { pausable, PausableObservable } from 'rxjs-pausable';

const source = interval(1000).pipe(
  map(x => `Hello ${x}!`),
  pauseable()
) as PausableObservable<string>;
 
source.subscribe(console.log);

setTimeout(() => source.pause(), 5000)

caroso1222

class PausableObservable<T> extends Observable<T> {
  private pauser: BehaviorSubject<boolean>;

  pause(){
    this.pauser.next(true);
  }

  resume(){
    this.pauser.next(false);
  }