ObservableDeployment.js

import { Subject, fromEventPattern, merge } from "rxjs";
import { takeUntil } from "rxjs/operators";

// Implementation
// see: https://web3js.readthedocs.io/en/1.0/web3-eth-contract.html#deploy
// -----------------------------------------------------------------------------

const CONFIRMATION = "confirmation";
const CONTRACT = "contract";
const ERROR = "error";
const RECEIPT = "receipt";
const TRANSACTION_HASH = "transactionHash";

function ObservableDeployment(deploymentTx, sendOptions) {
  const end$ = new Subject();
  const err$ = new Subject();
  let removed = false;

  let _removeTxSentHandler;
  function removeTxSentHandler(_handler) {
    if (_removeTxSentHandler) return _removeTxSentHandler();
    throw new Error("removeTxSentHandler called before addTxSentHandler");
  }

  function addTxSentHandler(handler) {
    function confirmationHandler(number, receipt) {
      handler({ type: CONFIRMATION, data: [number, receipt] });
    }

    function contractHandler(contractInstance) {
      if (removed) return;
      handler({ type: CONTRACT, data: contractInstance });
      end$.next(true);
    }

    function errorHandler(...args) {
      if (removed) throw args[0];
      err$.error(args);
    }

    function receiptHandler(receipt) {
      handler({ type: RECEIPT, data: receipt });
    }

    function transactionHashHandler(hash) {
      handler({ type: TRANSACTION_HASH, data: hash });
    }

    let txSent;
    try {
      txSent = deploymentTx.send(sendOptions);
      txSent
        .on(CONFIRMATION, confirmationHandler)
        .on(ERROR, errorHandler)
        .on(RECEIPT, receiptHandler)
        .on(TRANSACTION_HASH, transactionHashHandler)
        .then(contractHandler)
        .catch(errorHandler);
    } catch (err) {
      errorHandler(err);
    }

    _removeTxSentHandler = function() {
      removed = true;
      if (txSent) {
        txSent.removeListener(CONFIRMATION, confirmationHandler);
        txSent.removeListener(ERROR, errorHandler);
        txSent.removeListener(RECEIPT, receiptHandler);
        txSent.removeListener(TRANSACTION_HASH, transactionHashHandler);
      }
    };
  }

  return merge(
    fromEventPattern(addTxSentHandler, removeTxSentHandler),
    err$
  ).pipe(takeUntil(end$));
}

export function DeploymentObserver({
  complete,
  confirmation,
  contract,
  error,
  receipt,
  transactionHash
}) {
  return {
    next({ type, data }) {
      switch (type) {
        case CONFIRMATION:
          if (confirmation) confirmation(data);
          break;
        case CONTRACT:
          if (contract) contract(data);
          break;
        case RECEIPT:
          if (receipt) receipt(data);
          break;
        case TRANSACTION_HASH:
          if (transactionHash) transactionHash(data);
          break;
        default:
          throw new TypeError("unknown data type");
      }
    },
    error,
    complete
  };
}

// Example usage
// -----------------------------------------------------------------------------

// some web3 contract
import MyContract from "../MyContract";

const myContract$ = ObservableDeployment(
  MyContract.deploy({/*...*/}),
  {/* options for .send() */}
);

// multiple observable deployments could be operated on by combining observables
// with RxJS statics: concat, merge, etc.

myContract$
  /*.pipe(RxJS operators: filter, map, tap, etc.)*/
  .subscribe(
    DeploymentObserver({
      // all of the following are optional, but if an error handler isn't
      // provided and there is an error then it will result in an uncaught
      // exception
      complete() { console.log("deployment finished!"); },
      confirmation([confirmationNumber, receipt]) {/*...*/},
      contract(myContractInstance) {/*...*/},
      error([err, receiptIfOutOfGas]) {/*...*/},
      receipt({ events, ...receipt }) {/*...*/},
      transactionHash(hash) {/*...*/}
    })
  );

// if .unsubscribe() is called before the observable deployment is complete and
// if a deployment error occurs after .unsubscribe() is called, then it will
// result in an UnhandledPromiseRejection