[Functional Programming] Start With the API You Want Then Implement

If you can think ahead to how you want your code to look into the future, then you can tackle your problems from the inside out. You design how you want the code to look, then go about implementing the internals. This is similar to a "test driven development" mindset where you test the implementation first, then write the actual implementation of the function after.

Original way:

const transform =  pipe(
    map((x) => x[1]),
    filter((x) => x !== ','),
    concat,
    map(toUpper)
  );
let typeGreeting = transform(
  createZipOf(createInterval(100), createForOf('My Zipo'))
);
const cancelGreating = typeGreeting(_log)
// cancelGreating()

New:

const myZip = (boradcaster1, boradcaster2) => (...operators) => {
  return pipe(...operators)(createZipOf(boradcaster1, boradcaster2))
}

const typeGreeting2 = myZip(
  // boradcasters
 createInterval(100), createForOf('My Zipo')
)(
 // operators
    map((x) => x[1]),
    filter((x) => x !== ','),
    concat,
    map(toUpper)
)
const cancelGreating2 = typeGreeting2(_log)
// cancelGreating2()

---

import { curry, compose, toUpper, pipe } from 'ramda';

// #region listeners
const _log = (value) => console.log(value);
// #endregion

// #region broadcasters
const done = Symbol('done');
const addListener = curry((element, eventType, listener) => {
  return element.addEventListener(evenType, listener);
});
const createInterval = curry((time, listener) => {
  let i = 0;
  const id = setInterval(() => {
    listener(i++);
  }, time);
  return () => {
    clearInterval(id);
  };
});
const createForOf = curry((iterator, listener) => {
  const id = setTimeout(() => {
    for (let item of iterator) {
      listener(item);
    }
    listener(done);
  }, 0);
  return () => {
    clearTimeout(id);
  };
});
const createZipOf = curry((broadcaster1, broadcaster2, listener) => {
  let cancelBoth;
  let buffer1 = [];
  const cancel1 = broadcaster1((value) => {
    buffer1.push(value);
    if (buffer2.length) {
      listener([buffer1.shift(), buffer2.shift()]);
      if (buffer1[0] === done || buffer2[0] === done) {
        listener(done);
        cancelBoth();
      }
    }
  });

  let buffer2 = [];
  const cancel2 = broadcaster2((value) => {
    buffer2.push(value);
    if (buffer1.length) {
      listener([buffer1.shift(), buffer2.shift()]);
      if (buffer1[0] === done || buffer2[0] === done) {
        listener(done);
        cancelBoth();
      }
    }
  });
  cancelBoth = () => {
    cancel1();
    cancel2();
  };
  return cancelBoth;
});
// #endregion

// #region operators
const concat = curry((broadcaster, listener) => {
  let string = '';
  return broadcaster((value) => {
    if (value === done) {
      listener(done);
      return;
    }
    listener((string += value));
  });
});

const map = curry((transform, broadcaster, listener) => {
  return broadcaster((value) => {
    if (value === done) {
      listener(done);
      return;
    }
    listener(transform(value));
  });
});

const filter = curry((predicator, broadcaster, listener) => {
  return broadcaster((value) => {
    if (value === done) {
      listener(done);
      return;
    }
    if (predicator(value)) {
      listener(value);
    }
  });
});
// #endregion
const transform =  pipe(
    map((x) => x[1]),
    filter((x) => x !== ','),
    concat,
    map(toUpper)
  );
let typeGreeting = transform(
  createZipOf(createInterval(100), createForOf('My Zipo'))
);
const cancelGreating = typeGreeting(_log)
// cancelGreating()

const myZip = (boradcaster1, boradcaster2) => (...operators) => {
  return pipe(...operators)(createZipOf(boradcaster1, boradcaster2))
}

const typeGreeting2 = myZip(
  // boradcasters
 createInterval(100), createForOf('My Zipo')
)(
 // operators
    map((x) => x[1]),
    filter((x) => x !== ','),
    concat,
    map(toUpper)
)
const cancelGreating2 = typeGreeting2(_log)
// cancelGreating2()