Gradienthealth/file streaming of tar files

packages/core/src/enums/RequestType.ts

@@ -10,6 +10,8 @@ enum RequestType {
   Prefetch = 'prefetch',
   /** Lower priority, often used for background computations in the worker */
   Compute = 'compute',
+  /** Lowest Priority, used to cache adjascent studies using GoogleSheetService */
+  PreCache = 'precache',
 }
 
 export default RequestType;

packages/core/src/requestPool/imageLoadPoolManager.ts

@@ -39,5 +39,6 @@ imageLoadPoolManager.grabDelay = 0;
 imageLoadPoolManager.setMaxSimultaneousRequests(RequestType.Interaction, 1000);
 imageLoadPoolManager.setMaxSimultaneousRequests(RequestType.Thumbnail, 1000);
 imageLoadPoolManager.setMaxSimultaneousRequests(RequestType.Prefetch, 1000);
+imageLoadPoolManager.setMaxSimultaneousRequests(RequestType.PreCache, 1000);
 
 export default imageLoadPoolManager;

packages/core/src/requestPool/imageRetrievalPoolManager.ts

@@ -20,6 +20,7 @@ imageRetrievalPoolManager.setMaxSimultaneousRequests(
   200
 );
 imageRetrievalPoolManager.setMaxSimultaneousRequests(RequestType.Prefetch, 200);
+imageRetrievalPoolManager.setMaxSimultaneousRequests(RequestType.PreCache, 200);
 imageRetrievalPoolManager.grabDelay = 0;
 
 export default imageRetrievalPoolManager;

packages/core/src/requestPool/requestPoolManager.ts

@@ -79,13 +79,15 @@ class RequestPoolManager {
     thumbnail: 0,
     prefetch: 0,
     compute: 0,
+    precache: 0,
   };
   /* maximum number of requests of each type. */
   public maxNumRequests: {
     interaction: number;
     thumbnail: number;
     prefetch: number;
     compute: number;
+    precache: number;
   };
   /* A public property that is used to set the delay between requests. */
   public grabDelay: number;
@@ -104,6 +106,7 @@ class RequestPoolManager {
       thumbnail: { 0: [] },
       prefetch: { 0: [] },
       compute: { 0: [] },
+      precache: { 0: [] },
     };
 
     this.grabDelay = 5;
@@ -114,13 +117,15 @@ class RequestPoolManager {
       thumbnail: 0,
       prefetch: 0,
       compute: 0,
+      precache: 0,
     };
 
     this.maxNumRequests = {
       interaction: 6,
       thumbnail: 6,
       prefetch: 5,
       compute: 15,
+      precache: 5,
     };
   }
 
@@ -271,12 +276,16 @@ class RequestPoolManager {
       RequestType.Prefetch
     );
     const hasRemainingComputeRequests = this.sendRequests(RequestType.Compute);
+    const hasRemainingPreCacheRequests = this.sendRequests(
+      RequestType.PreCache
+    );
 
     if (
       !hasRemainingInteractionRequests &&
       !hasRemainingThumbnailRequests &&
       !hasRemainingPrefetchRequests &&
-      !hasRemainingComputeRequests
+      !hasRemainingComputeRequests &&
+      !hasRemainingPreCacheRequests
     ) {
       this.awake = false;
     }

packages/core/src/webWorkerManager/webWorkerManager.js

@@ -213,6 +213,42 @@ class CentralizedWorkerManager {
       worker.terminate();
     });
   }
+
+  postMessage(workerName, message) {
+    const workerProperties = this.workerRegistry[workerName];
+    if (!workerProperties) {
+      console.error(`Worker type '${workerName}' is not registered.`);
+      return;
+    }
+
+    workerProperties.nativeWorkers.forEach((worker) =>
+      worker.postMessage(message)
+    );
+  }
+
+  addEventListener(workerName, eventType = 'message', listener) {
+    const workerProperties = this.workerRegistry[workerName];
+    if (!workerProperties) {
+      console.error(`Worker type '${workerName}' is not registered.`);
+      return;
+    }
+
+    workerProperties.nativeWorkers.forEach((worker) =>
+      worker.addEventListener(eventType, listener)
+    );
+  }
+
+  removeEventListener(workerName, eventType = 'message', listener) {
+    const workerProperties = this.workerRegistry[workerName];
+    if (!workerProperties) {
+      console.error(`Worker type '${workerName}' is not registered.`);
+      return;
+    }
+
+    workerProperties.nativeWorkers.forEach((worker) =>
+      worker.removeEventListener(eventType, listener)
+    );
+  }
 }
 
 export default CentralizedWorkerManager;

packages/dicomImageLoader/src/imageLoader/internal/loadTarRequest.ts

@@ -0,0 +1,180 @@
+import { metaData, getWebWorkerManager, Enums } from '@cornerstonejs/core';
+import tarFileManager from '../wadors/tarFileManager';
+import { getOptions } from './index';
+import {
+  FILE_STREAMING_WORKER_NAME,
+  MAXIMUM_WORKER_FETCH_SIZE,
+} from '../wadors/registerFileStreaming';
+
+interface tarImageUrl {
+  tarUrl: string;
+  dicomPath: string;
+}
+const THRESHOLD = 10000;
+
+const tarPromises = {};
+
+function loadTarRequest(
+  uri: string,
+  imageId: string,
+  headers: Record<string, string> = {}
+): Promise<ArrayBufferLike> {
+  const instance = metaData.get('instance', imageId);
+  const { FileOffsets } = instance;
+  const { tarUrl } = parseuri(uri);
+  const { Range } = headers;
+  const handledOffsets = getHandledOffsets(FileOffsets, Range);
+  const extractedFile = tarFileManager.get(tarUrl, handledOffsets);
+
+  if (extractedFile) {
+    return new Promise<ArrayBufferLike>((resolveRequest, rejectRequest) => {
+      try {
+        resolveRequest(extractedFile.buffer);
+      } catch (error) {
+        rejectRequest(error);
+      }
+    });
+  }
+
+  const webWorkerManager = getWebWorkerManager();
+  let tarPromise: Promise<void>;
+
+  if (tarPromises[tarUrl]) {
+    tarPromise = tarPromises[tarUrl];
+  } else {
+    tarPromise = new Promise<void>(async (resolveTar, rejectTar) => {
+      if (
+        tarFileManager.getTotalSize() + THRESHOLD >
+        MAXIMUM_WORKER_FETCH_SIZE
+      ) {
+        throw new Error(
+          `fileStreaming.ts: Maximum size(${MAXIMUM_WORKER_FETCH_SIZE}) for fetching files reached`
+        );
+      }
+
+      const options = getOptions();
+      const beforeSendHeaders = options.beforeSend();
+
+      function handleFirstChunk(evt) {
+        const { url, position, fileArraybuffer } = evt.data;
+
+        if (url === tarUrl) {
+          tarFileManager.set(url, { data: fileArraybuffer, position });
+
+          webWorkerManager.removeEventListener(
+            FILE_STREAMING_WORKER_NAME,
+            'message',
+            handleFirstChunk
+          );
+        }
+      }
+
+      webWorkerManager.addEventListener(
+        FILE_STREAMING_WORKER_NAME,
+        'message',
+        handleFirstChunk
+      );
+
+      webWorkerManager
+        .executeTask(
+          FILE_STREAMING_WORKER_NAME,
+          'stream',
+          {
+            url: tarUrl,
+            headers: beforeSendHeaders,
+          },
+          { requestType: Enums.RequestType.Prefetch }
+        )
+        .then(() => {
+          resolveTar();
+        })
+        .catch((error) => {
+          webWorkerManager.removeEventListener(
+            FILE_STREAMING_WORKER_NAME,
+            'message',
+            handleFirstChunk
+          );
+          rejectTar(error);
+        })
+        .finally(() => delete tarPromises[tarUrl]);
+    });
+
+    tarPromises[tarUrl] = tarPromise;
+  }
+
+  return new Promise<ArrayBufferLike>(async (resolveRequest, rejectRequest) => {
+    function handleChunkAppend(evt) {
+      const { url, position, isAppending } = evt.data;
+
+      isAppending && tarFileManager.setPosition(url, position);
+
+      if (position > handledOffsets.endByte && url === tarUrl) {
+        try {
+          const file = tarFileManager.get(url, handledOffsets);
+
+          webWorkerManager.removeEventListener(
+            FILE_STREAMING_WORKER_NAME,
+            'message',
+            handleChunkAppend
+          );
+
+          resolveRequest(file.buffer);
+        } catch (error) {
+          rejectRequest(error);
+        }
+      }
+    }
+
+    webWorkerManager.addEventListener(
+      FILE_STREAMING_WORKER_NAME,
+      'message',
+      handleChunkAppend
+    );
+
+    await tarPromise.catch((error) => {
+      webWorkerManager.removeEventListener(
+        FILE_STREAMING_WORKER_NAME,
+        'message',
+        handleChunkAppend
+      );
+      rejectRequest(error);
+    });
+  });
+}
+
+function parseuri(uri: string): tarImageUrl {
+  const [tarUrl, dicomPath] = uri.split('.tar://');
+  return { tarUrl: tarUrl + '.tar', dicomPath };
+}
+
+function parseRangeHeader(rangeHeader: string): {
+  start?: number;
+  end?: number;
+} {
+  if (!rangeHeader) {
+    return {};
+  }
+
+  const parts = rangeHeader.split('=');
+  const rangePart = parts[1];
+  const rangeParts = rangePart.split('-');
+  const start = parseInt(rangeParts[0], 10);
+  const end = parseInt(rangeParts[1], 10);
+
+  return { start, end };
+}
+
+function getHandledOffsets(
+  fileOffsets: { startByte: number; endByte: number },
+  rangeHeader: string
+): { startByte: number; endByte: number } {
+  const { startByte: fileStartByte, endByte: fileEndByte } = fileOffsets;
+  const { start, end } = parseRangeHeader(rangeHeader);
+
+  const startByte = start ? fileStartByte + start : fileStartByte;
+  const endByte = end ? fileStartByte + end : fileEndByte;
+
+  return { startByte, endByte };
+}
+
+export default loadTarRequest;

packages/dicomImageLoader/src/imageLoader/internal/tarRequest.ts

@@ -0,0 +1,82 @@
+import { Types, Enums, metaData } from '@cornerstonejs/core';
+import { LoaderXhrRequestPromise } from '../../types';
+import { CornerstoneWadoRsLoaderOptions } from '../wadors/loadImage';
+import parseImageId from '../wadouri/parseImageId';
+import loadTarRequest from './loadTarRequest';
+import external from '../../externalModules';
+
+export default function tarRequest(
+  url: string,
+  imageId: string,
+  defaultHeaders: Record<string, string> = {},
+  options: CornerstoneWadoRsLoaderOptions = {}
+): LoaderXhrRequestPromise<{
+  contentType: string;
+  pixelData: Uint8Array;
+  imageQualityStatus: Enums.ImageQualityStatus;
+  percentComplete: number;
+}> | void {
+  const { pixelDataFrame } = parseImageId(imageId);
+  let imagePromise;
+
+  const instance = metaData.get('instance', imageId) || {};
+  const { CustomOffsetTable, CustomOffsetTableLengths } = instance;
+
+  if (CustomOffsetTable && CustomOffsetTableLengths) {
+    const startByte = CustomOffsetTable[pixelDataFrame];
+    const endByte = startByte + CustomOffsetTableLengths[pixelDataFrame];
+    const headerRange = `bytes=0-${CustomOffsetTable[0] - 1}`;
+    const pixelDataRange = `bytes=${startByte}-${endByte}`;
+
+    const headerPromise = loadTarRequest(url, imageId, {
+      ...defaultHeaders,
+      Range: headerRange,
+    });
+    const pixelDataPromise = loadTarRequest(url, imageId, {
+      ...defaultHeaders,
+      Range: pixelDataRange,
+    });
+
+    imagePromise = Promise.all([headerPromise, pixelDataPromise]).then(
+      (results) => ({ headerBuffer: results[0], pixelDataBuffer: results[1] })
+    );
+  } else {
+    imagePromise = loadTarRequest(url, imageId, defaultHeaders).then(
+      (arraybuffer) => ({
+        fileBuffer: arraybuffer,
+      })
+    );
+  }
+
+  return imagePromise.then((result) => {
+    const { headerBuffer, pixelDataBuffer, fileBuffer } = result;
+    const dataSet = external.dicomParser.parseDicom(
+      new Uint8Array(fileBuffer || headerBuffer),
+      { ...(headerBuffer && { untilTag: 'x7fe00010' }) }
+    );
+    const transferSyntax = dataSet.string('x00020010');
+    let pixelData;
+
+    if (fileBuffer) {
+      const pixelDataElement = dataSet.elements.x7fe00010;
+      let { dataOffset, length } = pixelDataElement;
+      if (pixelDataElement.hadUndefinedLength) {
+        ({ position: dataOffset, length } = pixelDataElement.fragments[0]);
+      } else {
+        // Adding 8 bytes for 4 bytes tag + 4 bytes length for uncomppressed pixelData
+        dataOffset += 8;
+      }
+      const slice = fileBuffer.slice(dataOffset, dataOffset + length);
+      pixelData = new Uint8Array(slice);
+    } else {
+      pixelData = new Uint8Array(pixelDataBuffer);
+    }
+
+    return {
+      contentType: `transfer-syntax=${transferSyntax}`,
+      imageQualityStatus: Enums.ImageQualityStatus.FULL_RESOLUTION,
+      pixelData,
+      percentComplete: 100,
+    };
+  });
+}