Use the Callback and Closure Pattern to Build Advanced Async Behaviors

03.compose-callback

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 createOperator = curry((operator, broadcaster, listener) => {
  // new a new broadcaster and invoke original broadcaster inside new broadcaster
  return operator((behaviorListener) => {
    // override the default broadcaster
      return broadcaster(value => {
        // apply common logic
        if(value === done) {
          // stop outer listen to continue emitting values
          listener(done)
          return
        }
        // otherwise, we want to pass forward the value to listener
        behaviorListener(value)
      })
  }, listener)
})

const concat = createOperator((broadcaster, listener) => {
  let string = '';
  return broadcaster((value) => {
    listener((string += value));
  });
});

const map = transform => createOperator((broadcaster, listener) => {
  return broadcaster((value) => {
    listener(transform(value));
  });
});

const filter = predicator => createOperator((broadcaster, listener) => {
  return broadcaster((value) => {
    if (predicator(value)) {
      listener(value);
    }
  });
});

const split = splitter => curry((broadcaster, listener) => {
  let buffer = []
  return broadcaster((value) => {
    if (value === done) {
      listener(buffer)
      listener(done)
      buffer = []
    }
    if (value === splitter) {
      listener(buffer)
      buffer = []
    } else {
      buffer.push(value)
    }
  })
})
// #endregion
const transform =  pipe(
    map((x) => x[1]),
    filter((x) => x !== ','),
    map(toUpper),
    split(" ")
  );
let typeGreeting = transform(
  createZipOf(createInterval(100), createForOf('My Zipo'))
);
const cancelGreating = typeGreeting((value => {
  if(value === done) {
    _log("Shut down")
    return
  }
  _log(value)
}))
// 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()