Session Proposal: NodeJS Asynchronous Scheduling about summit HOT 10 CLOSED

import { EventEmitter } from 'events'

// Unhandled exception
setImmediate(async () => {
  const e = await new Promise(resolve => {
    const e = new EventEmitter()
    resolve(e)
    process.nextTick(() => {
      e.emit('error', new Error('nextTick'))
    })
  })
  e.on('error', () => {})
})

// Unhandled exception
setImmediate(async () => {
  const e = await new Promise(resolve => {
    const e = new EventEmitter()
    resolve(e)
    queueMicrotask(() => {
      e.emit('error', new Error('queueMicrotask'))
    })
  })
  e.on('error', () => {})
})

// OK, but slow
setImmediate(async () => {
  const e = await new Promise(resolve => {
    const e = new EventEmitter()
    resolve(e)
    setImmediate(() => {
      e.emit('error', new Error('setImmediate'))
    })
  })
  e.on('error', () => {})
})

const events = require('events')

class EventEmitter extends events.EventEmitter {
  constructor(options) {
    super(options)
    this.buffer = options?.bufferStrategy || new NoBufferStrategy()
  }

  emit(name, ...args) {
    // No listeners yet. Try to buffer the event.
    if (!this.listeners(name).length && this.buffer.buffer(name, args)) {
      return
    }

    return super.emit(name, ...args)
  }

  on(name, ...args) {
    try {
      return super.on(name, ...args)
    } finally {
      // After attaching a handler, flush any buffered events.
      this.buffer.flush(name, (...args) => {
        this.emit(...args)
      })
    }
  }
}

// Default strategy does no buffering.
class NoBufferStrategy {
  buffer(name, task) {
    return false
  }

  flush() {}
}

// A possible alternative strategy could buffer the events until a handler is given.
class BufferStrategy {
  #buffered = new Map()

  buffer(name, args) {
    const map = this.#buffered

    if (!map.has(name)) {
      map.set(name, [])
    }
    map.get(name).push(args)

    return true
  }

  flush(name, handle) {
    const map = this.#buffered
    const queue = map.get(name)
    map.delete(name)

    for (const args of queue) {
      handle(name, ...args)
    }
  }
}

const p = new Promise((resolve) => {
  const e = new EventEmitter({
    bufferStrategy: new BufferStrategy()
  })

  process.nextTick(() => {
    e.emit('error', 'nextTick')
  })

  resolve(e)
})

p.then(e => {
  return new Promise(resolve => {
    setImmediate(resolve, e)
  })
}).then(e => {
  e.on('error', (error) => console.log('error', error))
})