feat: reuse postgres connection

README.md

@@ -5,6 +5,9 @@ A Pino Transport for PostgreSQL databases.
 ![GitHub](https://img.shields.io/github/license/Xstoudi/pino-pg?style=for-the-badge)
 ![npm](https://img.shields.io/npm/v/pino-pg?style=for-the-badge)
 
+## Requirement
+This package requires at least Node.js version `13.10.0`.
+
 ## Installation
 ```
 npm install pino-pg

package.json

@@ -12,6 +12,9 @@
   "bin": {
     "pino-pg": "./index.js"
   },
+  "engines": {
+    "node": ">=13.10.0"
+  },
   "keywords": [
     "pino",
     "pinojs",
@@ -43,6 +46,7 @@
     "@types/split2": "^2.1.6",
     "@types/through2": "^2.0.34",
     "@types/yargs": "^15.0.4",
+    "@types/node": "^13.9.1",
     "ava": "^3.5.0",
     "concat-stream": "^2.0.0",
     "rimraf": "^3.0.2",

src/index.ts

@@ -1,45 +1,28 @@
-#!/usr/bin/env node
 import { Client } from 'pg'
-import pump from 'pump'
 import split from 'split2'
-import through from 'through2'
+import { pipeline } from 'stream'
 
 import args from './args'
 
-interface PinoPgConfig {
-  connectionUrl: string
-  table: string
-  column: string
+function transporter(table: string, column: string, client: Client) {
+  return async function* (source: AsyncIterable<string>) {
+    for await (const line of source) {
+      await client.query(`INSERT INTO ${table}(${column}) VALUES($1)`, [JSON.parse(line)])
+      yield line
+    }
+  }
 }
 
-function transporter(config: PinoPgConfig) {
-  return through.obj((chunk, encoding, callback) => {
-    const client = new Client({ connectionString: config.connectionUrl})
-    client.connect((connectErr) => {
-      if(connectErr !== null) {
-        console.error('Failed to connect to PostgreSQL server.', connectErr)
-        return callback('Failed to connect to PostgreSQL server.')
-      }
-
-      client.query(`INSERT INTO ${config.table}(${config.column}) VALUES($1)`, [JSON.parse(chunk)], (queryErr, result) => {
-        if(queryErr !== null) {
-          console.error('Query failed.', queryErr)
-          return callback('Query failed.')
-        }
-        client.end(endErr => {
-          if(endErr !== undefined) {
-            console.error('Fail to close PG connection.', endErr)
-            return callback('Fail to close PG connection.')
-          }
-          callback(null, chunk)
-        })
-      })
+function main() {
+  const client = new Client({ connectionString: args.connectionUrl})
+  client.connect((connectErr) => {
+    if(connectErr !== null) {
+      return console.error('Failed to connect to PostgreSQL server.', connectErr)
+    }
+    pipeline(process.stdin, split(), transporter(args.table, args.column, client) as any, process.stdout, () => {
+      client.end()
     })
   })
 }
 
-function main() {
-  pump(process.stdin, split(), transporter(args as PinoPgConfig), process.stdout) 
-}
-
 export { main, transporter };