3.0.8

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartrx",
-  "version": "3.0.7",
+  "version": "3.0.8",
   "private": false,
   "description": "smart wrapper for rxjs",
   "main": "dist_ts/index.js",

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartrx',
-  version: '3.0.7',
+  version: '3.0.8',
   description: 'smart wrapper for rxjs'
 }

ts/index.ts

@@ -1,5 +1,6 @@
 import * as plugins from './smartrx.plugins.js';
 export * from './smartrx.classes.observablemap.js';
 export * from './smartrx.classes.observableintake.js';
+export * from './smartrx.functions.js';
 import * as rxjs from './smartrx.plugins.rxjs.js';
 export { rxjs };

ts/smartrx.functions.ts

@@ -0,0 +1,32 @@
+import { Observable } from 'rxjs';
+import { Readable } from 'stream';
+
+export function fromStreamWithBackpressure<T>(stream: Readable): Observable<T> {
+  return new Observable<T>((subscriber) => {
+    const pauseStream = () => stream.pause();
+    const resumeStream = () => process.nextTick(() => stream.resume());
+
+    // Handler for each piece of data
+    const onData = (data: T) => {
+      // Pause the stream to apply backpressure
+      pauseStream();
+      // Emit data and resume the stream if the subscriber is ready
+      subscriber.next(data);
+      resumeStream();
+    };
+
+    // Subscribe to stream events
+    stream.on('data', onData);
+    stream.on('error', (error) => subscriber.error(error));
+    stream.on('end', () => subscriber.complete());
+    stream.on('close', () => subscriber.complete());
+
+    // If the subscriber unsubscribes, clean up the stream listeners
+    return () => {
+      stream.removeListener('data', onData);
+      stream.removeListener('error', subscriber.error);
+      stream.removeListener('end', subscriber.complete);
+      stream.removeListener('close', subscriber.complete);
+    };
+  });
+}