rxjs the communication flow between contentScript and background

action.ts

export enum ActionType {
  Add = 'add',
  Remove = 'remove',
}

export type MessageStream = {
  message: any;
  sender: chrome.runtime.MessageSender;
  sendResponse: (response?: any) => void;
};

background.ts

import { Subject } from 'rxjs';
import { ActionType, MessageStream } from './util/actions';
import { ofType } from './util/rxjs-utils';

// setup
const onMessages$ = new Subject<MessageStream>();
chrome.runtime.onMessage.addListener((message, sender, sendResponse) => {
  onMessages$.next({ message, sender, sendResponse });
});

// consumer 1
onMessages$.pipe(ofType(ActionType.Add)).subscribe(({ message, sender, sendResponse }) => {
  console.log(message.type, message.payload);
  sendResponse(message);
});

// consumer 2
onMessages$.pipe(ofType(ActionType.Remove)).subscribe(({ message, sender, sendResponse }) => {
  console.log(message.type, message.payload);
  sendResponse(message);
});

contentScript.ts

import { fromEvent } from 'rxjs';
import { tap, map } from 'rxjs/operators';
import { ActionType } from './util/actions';
import { dispatch } from './util/rxjs-utils';

// dispatcher #1
fromEvent(document, 'click')
  .pipe(
    map((e) => e.target!['nodeName']),
    dispatch(ActionType.Add)
  )
  .subscribe((response) => {
    console.log(response);
  });

// dispatcher #2
fromEvent(document, 'keydown')
  .pipe(
    map((e) => e.target!['nodeName']),
    dispatch(ActionType.Remove)
  )
  .subscribe((response) => {
    console.log(response);
  });

rxjs-utils.ts

import { ActionType, MessageStream } from './actions';
import { switchMap, filter } from 'rxjs/operators';

export function dispatch(action: ActionType) {
  return switchMap((payload: any) => {
    return chrome.runtime.sendMessage({ type: action, payload });
  });
}

export function ofType<T extends MessageStream>(action: ActionType) {
  return filter<T>((entry: T) => entry.message.type === action);
}