Custom Transform stream using ReadableStream and WritableStream instance.

transformStream.js

/**
 * Creates a push controller and exposes a public interface to the readable's controller
 * by exposing the controller to it's parent scope
 * 
 * Hacky but neat trick in order to transform data
 * without the usage of Event Emitters
 */
function createPushReadable() {
    let rController;

    const readable = new ReadableStream({
        start(controller) { rController = controller; }
    });

    function push(data) {
        rController.enqueue(data);
    }

    function close() {
        rController.close();
    }

    return { push, close, readable };
}

/**
  * Custom Transform stream built with the help of concrete
  * ReadableStream and WritableStream instances
  */
function createWordReplaceStream(matchString, replacer) {
    const { readable, push, close: closeReadable } = createPushReadable();
    /**
     * Buffer to be stored without checking
     * in order to reduce the number of operations
     */

    const bufferSize = matchString.length - 1;
    const matchRegexp = new RegExp(matchString, 'ig');
    let wordsBuffer = '';

    const writable = new WritableStream({
        async write(chunk) {
            try {
                processChunk(chunk);
            } catch (error) {
                console.error(error);
            }
        },

        close() {
            // Remaining buffer doesn't contain anything that needs to be replaced
            pushRemainingBuffer();
            closeReadable();
        }
    });

    function processChunk(chunk) {
        if (typeof chunk !== 'string') {
            wordsBuffer += decodeText(chunk);
        } else {
            wordsBuffer += chunk;
        }


        const { string, lastReplacedEnd } = replaceText();
        wordsBuffer = string;

        const newBufferStart = Math.max(wordsBuffer.length - bufferSize, lastReplacedEnd);
        const pushedBuffer = sliceAndPushBuffer(newBufferStart);
    }

    function sliceAndPushBuffer(bufferOffset) {
        const bufferToPush = wordsBuffer.slice(0, bufferOffset);
        wordsBuffer = wordsBuffer.slice(bufferOffset);
        push(bufferToPush);

        return bufferToPush;
    }
    
    /**
     * Push String in the remaining buffer
     */
    function pushRemainingBuffer() {
        push(wordsBuffer);
    }

    /**
     * Replaces the text based on regular expression and 
     * provides the items such as lastReplacedEnd and totalLengthDifferent
     * 
     * This information is important to maintain the buffer
     */
    function replaceText() {
        let totalLengthDifference = 0;
        let lastReplacedEnd = 0;

        const replacedString = wordsBuffer.replace(matchRegexp, function (match, offset) {
            const replacedString = replacer(match, offset);
            
            /**
             * Replaced string length should ideally be extracted out using 
             * a simple formula
             */
            totalLengthDifference += replacedString.length - match.length;
            lastReplacedEnd = offset + match.length + totalLengthDifference;

            return replacedString;
        }); 

        return { string: replacedString, lastReplacedEnd, totalLengthDifference };
    }

    return { readable, writable };
}