iterators.js 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687
  1. import { Packr } from './pack.js'
  2. import { Unpackr } from './unpack.js'
  3. /**
  4. * Given an Iterable first argument, returns an Iterable where each value is packed as a Buffer
  5. * If the argument is only Async Iterable, the return value will be an Async Iterable.
  6. * @param {Iterable|Iterator|AsyncIterable|AsyncIterator} objectIterator - iterable source, like a Readable object stream, an array, Set, or custom object
  7. * @param {options} [options] - msgpackr pack options
  8. * @returns {IterableIterator|Promise.<AsyncIterableIterator>}
  9. */
  10. export function packIter (objectIterator, options = {}) {
  11. if (!objectIterator || typeof objectIterator !== 'object') {
  12. throw new Error('first argument must be an Iterable, Async Iterable, or a Promise for an Async Iterable')
  13. } else if (typeof objectIterator[Symbol.iterator] === 'function') {
  14. return packIterSync(objectIterator, options)
  15. } else if (typeof objectIterator.then === 'function' || typeof objectIterator[Symbol.asyncIterator] === 'function') {
  16. return packIterAsync(objectIterator, options)
  17. } else {
  18. throw new Error('first argument must be an Iterable, Async Iterable, Iterator, Async Iterator, or a Promise')
  19. }
  20. }
  21. function * packIterSync (objectIterator, options) {
  22. const packr = new Packr(options)
  23. for (const value of objectIterator) {
  24. yield packr.pack(value)
  25. }
  26. }
  27. async function * packIterAsync (objectIterator, options) {
  28. const packr = new Packr(options)
  29. for await (const value of objectIterator) {
  30. yield packr.pack(value)
  31. }
  32. }
  33. /**
  34. * Given an Iterable/Iterator input which yields buffers, returns an IterableIterator which yields sync decoded objects
  35. * Or, given an Async Iterable/Iterator which yields promises resolving in buffers, returns an AsyncIterableIterator.
  36. * @param {Iterable|Iterator|AsyncIterable|AsyncIterableIterator} bufferIterator
  37. * @param {object} [options] - unpackr options
  38. * @returns {IterableIterator|Promise.<AsyncIterableIterator}
  39. */
  40. export function unpackIter (bufferIterator, options = {}) {
  41. if (!bufferIterator || typeof bufferIterator !== 'object') {
  42. throw new Error('first argument must be an Iterable, Async Iterable, Iterator, Async Iterator, or a promise')
  43. }
  44. const unpackr = new Unpackr(options)
  45. let incomplete
  46. const parser = (chunk) => {
  47. let yields
  48. // if there's incomplete data from previous chunk, concatinate and try again
  49. if (incomplete) {
  50. chunk = Buffer.concat([incomplete, chunk])
  51. incomplete = undefined
  52. }
  53. try {
  54. yields = unpackr.unpackMultiple(chunk)
  55. } catch (err) {
  56. if (err.incomplete) {
  57. incomplete = chunk.slice(err.lastPosition)
  58. yields = err.values
  59. } else {
  60. throw err
  61. }
  62. }
  63. return yields
  64. }
  65. if (typeof bufferIterator[Symbol.iterator] === 'function') {
  66. return (function * iter () {
  67. for (const value of bufferIterator) {
  68. yield * parser(value)
  69. }
  70. })()
  71. } else if (typeof bufferIterator[Symbol.asyncIterator] === 'function') {
  72. return (async function * iter () {
  73. for await (const value of bufferIterator) {
  74. yield * parser(value)
  75. }
  76. })()
  77. }
  78. }
  79. export const decodeIter = unpackIter
  80. export const encodeIter = packIter