Fix unhandled ResultSet.json errors, prepare 1.6.0 (#311)

CHANGELOG.md

@@ -1,9 +1,13 @@
-# Unreleased
+# 1.6.0 (Common, Node.js, Web)
 
 ## New features
 
 - Added optional `real_time_microseconds` field to the `ClickHouseSummary` interface (see https://github.com/ClickHouse/ClickHouse/pull/69032)
 
+## Bug fixes
+
+- Fixed unhandled exceptions produced when calling `ResultSet.json` if the response data was not in fact a valid JSON. ([#311](https://github.com/ClickHouse/clickhouse-js/pull/311))
+
 # 1.5.0 (Node.js)
 
 ## New features

packages/client-common/src/version.ts

@@ -1 +1 @@
-export default '1.5.0'
+export default '1.6.0'

packages/client-node/__tests__/unit/node_result_set.test.ts

@@ -1,4 +1,4 @@
-import type { Row } from '@clickhouse/client-common'
+import type { DataFormat, Row } from '@clickhouse/client-common'
 import { guid } from '@test/utils'
 import Stream, { Readable } from 'stream'
 import { ResultSet } from '../../src'
@@ -63,10 +63,81 @@ describe('[Node.js] ResultSet', () => {
     expect(row.json()).toEqual({ foo: 'bar' })
   })
 
-  function makeResultSet(stream: Stream.Readable) {
+  describe('unhandled exceptions with streamable JSON formats', () => {
+    const logAndQuit = (err: Error | unknown, prefix: string) => {
+      console.error(prefix, err)
+      process.exit(1)
+    }
+    const uncaughtExceptionListener = (err: Error) =>
+      logAndQuit(err, 'uncaughtException:')
+    const unhandledRejectionListener = (err: unknown) =>
+      logAndQuit(err, 'unhandledRejection:')
+
+    const invalidJSON = 'invalid":"foo"}\n'
+
+    beforeAll(() => {
+      process.on('uncaughtException', uncaughtExceptionListener)
+      process.on('unhandledRejection', unhandledRejectionListener)
+    })
+    afterAll(() => {
+      process.off('uncaughtException', uncaughtExceptionListener)
+      process.off('unhandledRejection', unhandledRejectionListener)
+    })
+
+    describe('Streamable JSON formats - JSONEachRow', () => {
+      it('should not be produced (ResultSet.text)', async () => {
+        const rs = makeResultSet(
+          Stream.Readable.from([Buffer.from(invalidJSON)]),
+        )
+        const text = await rs.text()
+        expect(text).toEqual(invalidJSON)
+      })
+
+      it('should not be produced (ResultSet.json)', async () => {
+        const rs = makeResultSet(
+          Stream.Readable.from([Buffer.from(invalidJSON)]),
+        )
+        const jsonPromise = rs.json()
+        await expectAsync(jsonPromise).toBeRejectedWith(
+          jasmine.objectContaining({
+            name: 'SyntaxError',
+          }),
+        )
+      })
+    })
+
+    describe('Non-streamable JSON formats - JSON', () => {
+      it('should not be produced (ResultSet.text)', async () => {
+        const rs = makeResultSet(
+          Stream.Readable.from([Buffer.from(invalidJSON)]),
+          'JSON',
+        )
+        const text = await rs.text()
+        expect(text).toEqual(invalidJSON)
+      })
+
+      it('should not be produced (ResultSet.json)', async () => {
+        const rs = makeResultSet(
+          Stream.Readable.from([Buffer.from(invalidJSON)]),
+          'JSON',
+        )
+        const jsonPromise = rs.json()
+        await expectAsync(jsonPromise).toBeRejectedWith(
+          jasmine.objectContaining({
+            name: 'SyntaxError',
+          }),
+        )
+      })
+    })
+  })
+
+  function makeResultSet(
+    stream: Stream.Readable,
+    format: DataFormat = 'JSONEachRow',
+  ) {
     return ResultSet.instance({
       stream,
-      format: 'JSONEachRow',
+      format,
       query_id: guid(),
       log_error: (err) => {
         console.error(err)

packages/client-node/src/result_set.ts

@@ -82,16 +82,12 @@ export class ResultSet<Format extends DataFormat | unknown>
     // JSONEachRow, etc.
     if (isStreamableJSONFamily(this.format as DataFormat)) {
       const result: T[] = []
-      await new Promise((resolve, reject) => {
-        const stream = this.stream<T>()
-        stream.on('data', (rows: Row[]) => {
-          for (const row of rows) {
-            result.push(row.json())
-          }
-        })
-        stream.on('end', resolve)
-        stream.on('error', reject)
-      })
+      const stream = this.stream<T>()
+      for await (const rows of stream) {
+        for (const row of rows) {
+          result.push(row.json())
+        }
+      }
       return result as any
     }
     // JSON, JSONObjectEachRow, etc.

packages/client-node/src/utils/stream.ts

@@ -8,13 +8,9 @@ export async function getAsText(stream: Stream.Readable): Promise<string> {
   let text = ''
 
   const textDecoder = new TextDecoder()
-  await new Promise((resolve, reject) => {
-    stream.on('data', (chunk) => {
-      text += textDecoder.decode(chunk, { stream: true })
-    })
-    stream.on('end', resolve)
-    stream.on('error', reject)
-  })
+  for await (const chunk of stream) {
+    text += textDecoder.decode(chunk, { stream: true })
+  }
 
   // flush
   const last = textDecoder.decode()

packages/client-node/src/version.ts

@@ -1 +1 @@
-export default '1.5.0'
+export default '1.6.0'

packages/client-web/src/version.ts

@@ -1 +1 @@
-export default '1.5.0'
+export default '1.6.0'