init

Author: jonathanong

.gitignore

@@ -0,0 +1,3 @@
+.DS_Store
+node_modules
+coverage

LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2015 Jonathan Ong me@jongleberry.com
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,92 @@
+# @koa/json-async-iterator
+
+Koa JSON response body streaming for async iterators
+
+## Usage
+
+```js
+const jsonAsyncIterator = require('@koa/json-async-iterator')
+const app = new Koa()
+
+app.use(jsonAsyncIterator())
+
+app.use(async ctx => {
+  const iterator = async function* () {
+    yield Promise.resolve({ foo: 'bar' })
+    yield Promise.resolve({ baz: 'qux' })
+  }
+  ctx.body = iterator()
+})
+```
+
+Output:
+
+```json
+[
+{"foo":"bar"}
+,
+{"baz":"qux"}
+]
+```
+
+Example with PostgreSQL:
+
+```js
+import QueryStream from 'pg-query-stream'
+import pg from 'pg'
+
+const client = new pg.Client()
+await client.connect()
+
+// later in your middleware
+app.use(async (ctx) => {
+    // stream all the rows from the database to the client
+    ctx.body = client.query(new QueryStream(`
+        SELECT * 
+        FROM posts
+        ORDER BY id ASC
+    `))
+})
+```
+
+## Options
+
+### `before`
+
+The string to write before the first value in the iterator.
+
+Default: `'[\n'`
+
+### `separator`
+
+The string to write between each value in the iterator.
+
+Default: `'\n,\n'`
+
+### `after`
+
+The string to write after the last value in the iterator.
+
+Default: `'\n]'`
+
+### `formatError`
+
+A function that takes an error and returns a value to be stringified.
+
+Default: `(err) => ({ error: { message: err.message } })`
+
+### `pretty`
+
+Whether to pretty-print the JSON.
+
+Default: `false`
+
+### `spaces`
+
+The number of spaces to use for pretty-printing.
+
+Default: `2`
+
+## License
+
+MIT

__tests__/index.js

@@ -0,0 +1,53 @@
+const supertest = require('supertest')
+const Koa = require('koa')
+const jsonAsyncIterator = require('..')
+
+describe('koa-json-async-iterator', () => {
+    it('should stream an async iterator', async () => {
+        const app = new Koa()
+        app.use(jsonAsyncIterator())
+        app.use(async ctx => {
+            const iterator = async function* () {
+                yield new Promise((resolve) => setTimeout(resolve, 100))
+                yield Promise.resolve({ foo: 'bar' })
+                yield new Promise((resolve) => setTimeout(resolve, 100))
+                yield Promise.resolve({ baz: 'qux' })
+            }
+            ctx.body = iterator()
+        })
+
+        const response = await supertest(app.callback()).get('/')
+        expect(response.status).toBe(200)
+        expect(response.headers['content-type']).toBe('application/json; charset=utf-8')
+        expect(response.body).toEqual([
+            { foo: 'bar' },
+            { baz: 'qux' }
+        ])
+    })
+
+    it('should thrown errors', async () => {
+        const app = new Koa()
+        app.use(jsonAsyncIterator())
+        app.use(async ctx => {
+            const iterator = async function* () {
+                yield new Promise((resolve) => setTimeout(resolve, 100))
+                yield Promise.resolve({ foo: 'bar' })
+                yield new Promise((resolve) => setTimeout(resolve, 100))
+                yield Promise.resolve({ baz: 'qux' })
+                yield new Promise((resolve) => setTimeout(resolve, 100))
+                throw new Error('test')
+            }
+            ctx.body = iterator()
+        })
+
+        const response = await supertest(app.callback()).get('/')
+        expect(response.status).toBe(200)
+        expect(response.body).toEqual([
+            { foo: 'bar' },
+            { baz: 'qux' },
+            { error: {
+                message: 'test',
+            } }
+        ])
+    })
+})

index.js

@@ -0,0 +1,53 @@
+const { PassThrough } = require('stream')
+
+module.exports = (options = {}) => {
+    const before = options.before || '[\n'
+    const after = options.after || '\n]'
+    const separator = options.separator || '\n,\n'
+    const pretty = options.pretty || false
+    const spaces = pretty ? options.spaces || 2 : 0
+    const formatError = options.formatError || ((err) => ({
+        error: {
+            message: err.message,
+        }
+    }))
+    return async (ctx, next) => {
+        await next()
+
+        if (ctx.body !== ctx.body?.[Symbol.asyncIterator]?.()) return
+
+        const iterator = ctx.body
+
+        ctx.response.type = 'json'
+        const stream = 
+        ctx.response.body = new PassThrough()
+
+        stream.write(before)
+
+        let ended = false
+        ;(async () => {
+            let first = true
+            try {
+                for await (const value of iterator) {
+                    if (!value) continue
+                    if (first) {
+                        first = false
+                    } else {
+                        stream.write(separator)
+                    }
+                    stream.write(JSON.stringify(value, null, spaces))
+                }
+                ended = true
+            } catch (err) {
+                if (!first) {
+                    stream.write(separator)
+                }
+                stream.write(JSON.stringify(formatError(err), null, spaces))
+                ctx.app.emit('error', err, ctx)
+            } finally {
+                stream.write(after)
+                stream.end()
+            }
+        })()
+    }
+}