gen-browser Pinger/Ponger refactor part 2 - enter RxJS 6

AddressInput.js

// file: src/AddressInput.js
import {getPropPathValue} from './Misc';

// --- DOM hook
const _addressInput = document.querySelector('main input[type="text"]');

function selectAddress(event) {
  const name = getPropPathValue(event, ['target','constructor','name'], null);
  if(name == 'HTMLInputElement') {
    event.target.select();
  }
  event.preventDefault();
}

export function addSelectListener() {
  _addressInput.addEventListener('click', selectAddress, true);
}

export function setAddress(text) {
  _addressInput.value = text;
}

export function getAddress(defaultValue) {
  const value = _addressInput.value;
  return !value ? defaultValue : value;
}

Connect.js

// file: src/Connect.js
import {start} from './GenBrowser'; // formerly client.js, now GenBrowser.js
import {from, Observable} from "rxjs";
import {publish, switchMap} from "rxjs/operators";

function messageNext(observer, message) {
  if(observer) {
    observer.next(message);
  } else {
    console.info('messageNext: no observer. message %O', message);
  }
}

function messageComplete(observer) {
  if(observer) {
    observer.complete(message);
  } else {
    console.info('messageComplete: no observer');
  }
}

function makeMessageObservable(client) {
  const {mailbox} = client;

  const connectMessageObserver = observer => {
    const ref = observer;
    const unsubscribe = () => {
      ref = null;                      // disconnect observer
      mailbox.close();                 // close client mailbox
    };                                 // NOTE: where to close client interface?

    mailbox.setHandler(                // connect message observer to client mailbox
      msg => messageNext(ref, msg),
      () => messageComplete(ref)
    );

    return unsubscribe;
  }; // end connectMessageObserver

  return Observable.create(connectMessageObserver);
}

function makeSetupClient(clientCb) {
  const setupClient = client => {
    const observable = makeMessageObservable(client);
    if (clientCb) {
      clientCb(client);                 // opportunity to copy client interface
    }                                   // TODO: expose client interface value
    return observable;                  // via a second Observable/Subject
  };

  return setupClient;                   // function that converts "client" to a message Observable
}

export function makeMessageConnectable(url, onClient = null) {
  let source = from(start(url)).pipe(
    switchMap(makeSetupClient(onClient)) // - From the client interface value generate a messageObservable
  );                                     // and "switch" to that Observable to stream messages
                                         // - NOTE: error is propagated to the subscriber
  return (publish())(source);            // - Turn Observable into multi-cast capable Subject
}                                        // inside a ConnectableObservable

Misc.js

// file: src/Misc.js

export const PING = 'ping';
export const PONG = 'pong';

export function now() {
  return new Date().toLocaleTimeString();
}

export function toTextFrom(from) {
  const [_, signature] = from.split('--');
  return signature.substring(0, 20) + '...';
};

export function getPropPathValue(obj, path, defaultValue) {
  let val = obj;
  for(let name of path) {
    if (!val) {
      val = defaultValue;
      break;
    }
    val = val[name];
  }
  return val;
}

export function clientSend({send, address}, to, type) {
  if (send) {
    send(to, {type, from: address});

  } else {
    console.info('clientSend - no client. to: %s type: %s', to, type);
  }
}

export function makeCleanup(subscription) {
  const cleanup = event => {
    subscription.unsubscribe();
    // event.returnValue = "\o/"; // for leave prompt
  };

  return cleanup;
}

package.json

{
  "name": "gen-browser",
  "scripts": {
    "build": "node build.js",
    "test": "jest",
    "pretest": "npm run build"
  },
  "devDependencies": {
    "eventsourcemock": "^2.0.0",
    "jest": "^23.6.0",
    "rollup": "^0.66.6",
    "rollup-plugin-node-resolve": "^3.4.0",
    "rxjs": "^6.3.3"
  }
}

Pinger.js

// file: src/Pinger.js
import {setStatus} from './StatusDisplay';
import {addSelectListener, getAddress, setAddress} from './AddressInput';
import {clientSend, makeCleanup, now, PING, PONG, toTextFrom} from './Misc';
import {prependMsgRow} from './MsgTable';

import {makeMessageConnectable} from './Connect';
import {fromEvent} from "rxjs";
import {filter, tap} from "rxjs/operators";

addSelectListener();
setStatus('Connecting ...');

let connectable = makeMessageConnectable('http://localhost:8080', onClient);

let subscription = connectable.pipe(
  tap(msg => console.info('received message: %O', msg)),
  filter(({type}) => type === PONG)
).subscribe(onPong, onError, onComplete);

fromEvent(window, 'beforeunload')
  .subscribe(makeCleanup(subscription));

connectable.connect();

// --- === ---
let clickSubscription = null;

function onClient(client) {
  console.info(client.address);

  if (clickSubscription) {
    clickSubscription.unsubscribe();
    clickSubscription = null;
  }
  const button = document.querySelector('form > button');
  clickSubscription = fromEvent(button, 'click')
    .subscribe(makeSendPing(client));

  setStatus('Ready');
}

function onPong({type, from}) {
  prependMsgRow(type, now(), toTextFrom(from));
}
function onError(error) {
  setStatus(error.toString());
}
function onComplete() {
  console.info('onComplete ');
}

function makeSendPing(client) {
  const {config} = client;

  return event => {
    clientSend(
      client,
      getAddress(config.logger),
      PING
    );

    event.preventDefault();
  };
}

Ponger.js

// file: src/Ponger.js
import {setStatus} from './StatusDisplay';
import {addSelectListener, setAddress} from './AddressInput';
import {clientSend, makeCleanup, now, PING, PONG, toTextFrom} from './Misc';
import {prependMsgRow} from './MsgTable';

import {makeMessageConnectable} from './Connect';
import {fromEvent} from "rxjs";
import {filter, tap} from "rxjs/operators";

addSelectListener();
setStatus('Connecting ...');

let connectable = makeMessageConnectable('http://localhost:8080', onClient);

let subscription = connectable.pipe(
  tap(msg => console.info('received message: %O', msg)),
  filter(({type}) => type === PING)
).subscribe(onPing, onError, onComplete);

fromEvent(window, 'beforeunload')
  .subscribe(makeCleanup(subscription));

connectable.connect();

// --- === ---

let currentClient = null;

function onClient(client) {
  currentClient = client;

  console.info(client.address);
  setAddress(client.address);
  setStatus('Ready');
}

function onPing({type, from}) {
  prependMsgRow(type, now(), toTextFrom(from));
  clientSend(currentClient, from, PONG);
}
function onError(error) {
  setStatus(error.toString());
}
function onComplete() {
  console.info('onComplete ');
}