Added support for writing received DICOM content to Writable streams

README.md

@@ -21,7 +21,7 @@ Part of the networking code was taken from [dicom-dimse][dicom-dimse-url].
 - Implements C-ECHO, C-FIND, C-STORE, C-MOVE, C-GET, C-CANCEL, N-CREATE, N-ACTION, N-DELETE, N-EVENT-REPORT, N-GET and N-SET services as SCU and SCP.
 - Supports secure DICOM TLS connections and user identity negotiation.
 - Allows custom DICOM implementations (Implementation Class UID and Implementation Version).
-- Provides asynchronous event handlers for incoming SCP requests.
+- Provides asynchronous event handlers and streaming support for incoming SCP requests.
 
 ### Examples
 

index.d.ts

@@ -2,6 +2,7 @@ import { Mixin } from 'ts-mixer';
 import { Logger } from 'winston';
 import { Socket } from 'net';
 import { SecureContext, TLSSocket } from 'tls';
+import { Writable } from 'stream';
 import AsyncEventEmitter from 'async-eventemitter';
 
 declare namespace PresentationContextResult {
@@ -1421,6 +1422,27 @@ declare class Scp extends Network {
     }
   );
 
+  /**
+   * Allows the caller to create a Writable stream to accumulate the C-STORE dataset.
+   * The default implementation creates a memory Writable stream that for, big instances,
+   * could cause out of memory situations.
+   */
+  createStoreWritableStream(
+    acceptedPresentationContext: PresentationContext,
+    request: CStoreRequest
+  ): Writable;
+
+  /**
+   * Allows the caller to create a Dataset from the Writable stream used to
+   * accumulate the C-STORE dataset. The created Dataset is passed to the
+   * Scp.cStoreRequest method for processing.
+   */
+  createDatasetFromStoreWritableStream(
+    writable: Writable,
+    acceptedPresentationContext: PresentationContext,
+    callback: (dataset: Dataset) => void
+  ): void;
+
   /**
    * Association request received.
    */

package.json

@@ -1,6 +1,6 @@
 {
   "name": "dcmjs-dimse",
-  "version": "0.1.27",
+  "version": "0.2.0",
   "description": "DICOM DIMSE implementation for Node.js using dcmjs",
   "main": "build/dcmjs-dimse.min.js",
   "module": "build/dcmjs-dimse.min.js",
@@ -45,6 +45,7 @@
   "dependencies": {
     "async-eventemitter": "^0.2.4",
     "dcmjs": "^0.30.3",
+    "memorystream": "^0.3.1",
     "smart-buffer": "^4.2.0",
     "ts-mixer": "^6.0.4",
     "winston": "^3.12.0"

src/Network.js

@@ -46,6 +46,7 @@ const log = require('./log');
 const { SmartBuffer } = require('smart-buffer');
 const { EOL } = require('os');
 const AsyncEventEmitter = require('async-eventemitter');
+const MemoryStream = require('memorystream');
 
 //#region Network
 class Network extends AsyncEventEmitter {
@@ -70,7 +71,8 @@ class Network extends AsyncEventEmitter {
     this.requests = [];
     this.pending = [];
 
-    this.dimseBuffer = undefined;
+    this.dimseStream = undefined;
+    this.dimseStoreStream = undefined;
     this.dimse = undefined;
 
     opts = opts || {};
@@ -229,6 +231,36 @@ class Network extends AsyncEventEmitter {
     return this.statistics;
   }
 
+  /**
+   * Allows the caller to create a Writable stream to accumulate the C-STORE dataset.
+   * The default implementation creates a memory Writable stream that for, big instances,
+   * could cause out of memory situations.
+   * @method
+   * @param {PresentationContext} acceptedPresentationContext - The accepted presentation context.
+   * @returns {Writable} The created store writable stream.
+   */
+  // eslint-disable-next-line no-unused-vars
+  createStoreWritableStream(acceptedPresentationContext) {
+    return MemoryStream.createWriteStream();
+  }
+
+  /**
+   * Allows the caller to create a Dataset from the Writable stream used to
+   * accumulate the C-STORE dataset. The created Dataset is passed to the
+   * Scp.cStoreRequest method for processing.
+   * @method
+   * @param {Writable} writable - The store writable stream.
+   * @param {PresentationContext} acceptedPresentationContext - The accepted presentation context.
+   * @param {function(Dataset)} callback - Created dataset callback function.
+   */
+  createDatasetFromStoreWritableStream(writable, acceptedPresentationContext, callback) {
+    const dataset = new Dataset(
+      writable.toBuffer(),
+      acceptedPresentationContext.getAcceptedTransferSyntaxUid()
+    );
+    callback(dataset);
+  }
+
   //#region Private Methods
   /**
    * Advances the message ID.
@@ -487,18 +519,42 @@ class Network extends AsyncEventEmitter {
     try {
       const pdvs = pdu.getPdvs();
       pdvs.forEach((pdv) => {
-        if (!this.dimseBuffer) {
-          this.dimseBuffer = SmartBuffer.fromOptions({
-            encoding: 'ascii',
-          });
-        }
-        this.dimseBuffer.writeBuffer(pdv.getValue());
         const presentationContext = this.association.getPresentationContext(
           pdv.getPresentationContextId()
         );
+
+        if (!this.dimse) {
+          // Create stream to receive command
+          if (!this.dimseStream) {
+            this.dimseStream = MemoryStream.createWriteStream();
+            this.dimseStoreStream = undefined;
+          }
+        } else {
+          // Create stream to receive dataset
+          if (this.dimse.getCommandFieldType() === CommandFieldType.CStoreRequest) {
+            if (!this.dimseStoreStream) {
+              this.dimseStream = undefined;
+              this.dimseStoreStream = this.createStoreWritableStream(
+                presentationContext,
+                this.dimse
+              );
+            }
+          } else {
+            if (!this.dimseStream) {
+              this.dimseStream = MemoryStream.createWriteStream();
+              this.dimseStoreStream = undefined;
+            }
+          }
+        }
+
+        const stream = this.dimseStream || this.dimseStoreStream;
+
+        // TODO: Add stream backpressure event handling
+        stream.write(pdv.getValue());
+
         if (pdv.isLastFragment()) {
           if (pdv.isCommand()) {
-            const command = new Command(new Dataset(this.dimseBuffer.toBuffer()));
+            const command = new Command(new Dataset(this.dimseStream.toBuffer()));
             const type = command.getCommandFieldType();
             switch (type) {
               case CommandFieldType.CEchoRequest:
@@ -587,17 +643,35 @@ class Network extends AsyncEventEmitter {
               this.dimse = undefined;
               return;
             }
-            this.dimseBuffer = undefined;
+            this.dimseStream = undefined;
+            this.dimseStoreStream = undefined;
           } else {
-            const dataset = new Dataset(
-              this.dimseBuffer.toBuffer(),
-              presentationContext.getAcceptedTransferSyntaxUid(),
-              this.datasetReadOptions
-            );
-            this.dimse.setDataset(dataset);
-            this._performDimse(presentationContext, this.dimse);
-            this.dimseBuffer = undefined;
-            this.dimse = undefined;
+            if (this.dimse.getCommandFieldType() === CommandFieldType.CStoreRequest) {
+              this.dimseStoreStream.end();
+              this.createDatasetFromStoreWritableStream(
+                this.dimseStoreStream,
+                presentationContext,
+                (dataset) => {
+                  this.dimse.setDataset(dataset);
+                  this._performDimse(presentationContext, this.dimse);
+                  this.dimseStream = undefined;
+                  this.dimseStoreStream = undefined;
+                  this.dimse = undefined;
+                }
+              );
+            } else {
+              this.dimseStream.end();
+              const dataset = new Dataset(
+                this.dimseStream.toBuffer(),
+                presentationContext.getAcceptedTransferSyntaxUid(),
+                this.datasetReadOptions
+              );
+              this.dimse.setDataset(dataset);
+              this._performDimse(presentationContext, this.dimse);
+              this.dimseStream = undefined;
+              this.dimseStoreStream = undefined;
+              this.dimse = undefined;
+            }
           }
         }
       });

src/Server.js

@@ -100,6 +100,37 @@ class Scp extends Network {
     });
   }
 
+  /**
+   * Allows the caller to create a Writable stream to accumulate the C-STORE dataset.
+   * The default implementation creates a memory Writable stream that for, big instances,
+   * could cause out of memory situations.
+   * @method
+   * @param {PresentationContext} acceptedPresentationContext - The accepted presentation context.
+   * @param {CStoreRequest} request - C-STORE request.
+   * @returns {Writable} The created store writable stream.
+   */
+  // eslint-disable-next-line no-unused-vars
+  createStoreWritableStream(acceptedPresentationContext, request) {
+    return super.createStoreWritableStream();
+  }
+
+  /**
+   * Allows the caller to create a Dataset from the Writable stream used to
+   * accumulate the C-STORE dataset. The created Dataset is passed to the
+   * Scp.cStoreRequest method for processing.
+   * @method
+   * @param {Writable} writable - The store writable stream.
+   * @param {PresentationContext} acceptedPresentationContext - The accepted presentation context.
+   * @param {function(Dataset)} callback - Created dataset callback function.
+   */
+  createDatasetFromStoreWritableStream(writable, acceptedPresentationContext, callback) {
+    return super.createDatasetFromStoreWritableStream(
+      writable,
+      acceptedPresentationContext,
+      callback
+    );
+  }
+
   /**
    * Association request received.
    * @method

src/version.js

@@ -1 +1 @@
-module.exports = '0.1.27';
+module.exports = '0.2.0';