feat: Передача и парсинг json в потоках

Closes #85

Author: niki

src/StreamOptions/BufferToJsonTransform.ts

@@ -0,0 +1,39 @@
+import { Transform, TransformCallback, TransformOptions } from 'stream';
+
+class BufferToJsonTransform<T> extends Transform {
+  private head: Buffer | undefined;
+  private errors = {
+    CONVERSION_ERROR: 'Не удалось преобразовать данные',
+  };
+  constructor(options: TransformOptions) {
+    super({ objectMode: true, highWaterMark: 10, ...options });
+  }
+  async _transform(tail: Buffer, _: BufferEncoding, cb: TransformCallback) {
+    let partnerParams: T | undefined;
+    try {
+      try {
+        if (this.head) {
+          tail = Buffer.concat([this.head, Buffer.from(tail)]);
+          partnerParams = JSON.parse(tail.toString());
+          this.head = undefined;
+        } else {
+          partnerParams = JSON.parse(tail.toString());
+        }
+      } catch (err) {
+        this.head = Buffer.from(tail);
+        cb();
+      }
+
+      if (partnerParams) {
+        cb(null, partnerParams);
+      }
+    } catch (error) {
+      console.error(this.errors.CONVERSION_ERROR, tail.toString());
+      if (error instanceof Error) {
+        cb(error);
+      } else {
+        cb(new Error(this.errors.CONVERSION_ERROR));
+      }
+    }
+  }
+}

src/StreamOptions/JsonToBufferTransform.ts

@@ -0,0 +1,7 @@
+import { Transform, TransformCallback } from 'stream';
+
+export class JsonToBufferTransform<T> extends Transform {
+  transform(chunk: T, encoding: BufferEncoding, cb: TransformCallback) {
+    cb(null, Buffer.from(JSON.stringify(chunk)));
+  }
+}

src/StreamOptions/index.ts

@@ -0,0 +1,2 @@
+export * from './JsonToBufferTransform';
+export * from './BufferToJsonTransform';

src/index.ts

@@ -4,3 +4,4 @@ export * from './Method';
 export * from './injector';
 export * from './Container';
 export * from './interfaces';
+export * from './StreamOptions';