Add pipeline workflow

[rumkin]

Pipeline workflow is an alternative to event listening. Which makes request handling more predictable. Code example:

const Plant = require('@plant/plant')
const {createPipeline} = require('@plant/plant')

const plant = new Plant()

plant.use(({res}) => {
  res.body = 'Hello, World!'
})

let isAppClosing = false

const pipeline = createPipeline(plant)

function onError(error) {
  console.error(error)

  if (pipeline.isOpened) {
    pipeline.close()
  }
}

pipeline.listen(8080)
.then(async () => {
  for await (const {handle, reply, reject} of pipeline) {
    if (isAppClosing) {
      reject()
      .catch(onError)
    }
    else {
      handle()
      .then(() => reply())
      .catch(onError)
    }
  }
})

process.on('SIGINT', () => {
  isAppClosing = true
})

Alternative option is to autoreject new requests:

// ...
const pipeline = createPipeline(plant)

pipeline.listen(8080)
.then(async () => {
  for await (const {handle, reply} of pipeline) {
    handle()
    .then(() => reply())
    .catch((error) => {
      if (error.code === 'socket_closed') {
        return
      }

      console.error(error)

      if (pipeline.isOpened) {
        pipeline.close()
      }
    })
  }
})

process.on('SIGINT', () => {
  // Reject all incoming requests and wait existing before close.
  pipeline.shutdown()
})

And the shortest version:

// ...
const pipeline = createPipeline(plant)

pipeline.listenAndHandle(8080, {
  onError(error) { // socket_closed error will be ignored here
    console.error(error)

    return true
  },
})
.then(() => {
  console.log('Pipeline is closed. All requests are settled!')
})

process.on('SIGINT', () => {
  // Reject all incoming requests and wait existing before close.
  pipeline.shutdown()
})