diff --git a/docs/AsyncDicomReader-skill.md b/docs/AsyncDicomReader-skill.md index 13c0c0f7..72077eda 100644 --- a/docs/AsyncDicomReader-skill.md +++ b/docs/AsyncDicomReader-skill.md @@ -98,6 +98,71 @@ async function readWithCustomListener(arrayBuffer) { } ``` +### Reading Metadata from a Cancellable Stream + +`readFileFromAsyncStream()` accepts Node readable streams and browser +`ReadableStream` sources. It owns the source lifecycle and stops the source +when parsing completes or reaches an early-stop condition. + +```javascript +import fs from "node:fs"; +import dcmjs from "dcmjs"; + +const { AsyncDicomReader } = dcmjs.async; +const { TagHex } = dcmjs.constants; + +const reader = new AsyncDicomReader(); +const source = fs.createReadStream("image.dcm"); + +const result = await reader.readFileFromAsyncStream(source, { + untilTag: TagHex.PixelData, + includeUntilTagValue: false, + streamOptions: { + readAheadHighWaterMark: 64 * 1024 + } +}); + +console.log(result.dict); +console.log(result.stopInfo); +``` + +`readAheadHighWaterMark` applies between source chunks. A single incoming +chunk is added atomically, so retained bytes may exceed the mark by up to one +source chunk. + +The same API accepts a browser response body: + +```javascript +const response = await fetch("/image.dcm"); +const reader = await AsyncDicomReader.readFileFromAsyncStream(response.body, { + untilTag: "7FE00010" +}); +``` + +The exposed pump distinguishes successful parser completion from external +processing failure. The reader stops its source successfully after an +`untilTag`, `stopOnGreaterTag`, or `shouldStop` match. An external workflow, +such as a STOW handler, can reject the read with its own error: + +```javascript +const readPromise = reader.readFileFromAsyncStream(source, options); +const processingPromise = processIncomingRequest().catch(error => { + reader.pump.abort(error); + throw error; +}); + +await Promise.all([readPromise, processingPromise]); +``` + +`pump.abort(error)` rejects a read that is still pending. It cannot change the +state of a read promise that has already fulfilled, so the processing promise +must also be awaited as shown above. + +Generic async iterables remain supported by the lower-level +`reader.stream.fromAsyncStream()` API. They are not accepted by +`readFileFromAsyncStream()` because the async iterator protocol does not +guarantee that `return()` interrupts a pending `next()` call. + ## Architecture ### Core Components @@ -402,6 +467,20 @@ Reads the entire DICOM file including meta information and dataset. **Returns:** `Promise` - The reader instance +#### `async readFileFromAsyncStream(stream, options)` +Reads a DICOM file from a cancellable Node stream or browser +`ReadableStream`. The source is stopped after parsing completes or reaches an +early-stop condition. + +**Options:** +- All `readFile()` options +- `streamOptions.readAheadHighWaterMark` (number): Pause source reads between + chunks when unread bytes reach this threshold +- `readerOptions` (Object, static factory only): Options for the reader + constructor + +**Returns:** `Promise` - The reader instance + #### `async readMeta(options)` Reads only the file meta information (Group 0x0002). @@ -447,6 +526,10 @@ Reads a single tag header (tag, VR, length). - `dict` (Object): Dataset dictionary - `stream` (ReadBufferStream): Underlying buffer stream - `listener` (DicomMetadataListener): Current listener instance +- `pump` (Object): Active stream handle exposing `abort(error)`, `failure`, + and `finished` +- `stopInfo` (Object): Early-stop reason, tag/value offsets, available bytes, + and loaded end offset ## Common Patterns @@ -720,7 +803,6 @@ The AsyncDicomReader is marked as preliminary. Future versions may include: - Files without DICM preamble - Improved streaming for network sources - Progress callbacks -- Cancelable operations - More robust error recovery ## References diff --git a/src/AsyncDicomReader.js b/src/AsyncDicomReader.js index 6a6be593..41c80468 100644 --- a/src/AsyncDicomReader.js +++ b/src/AsyncDicomReader.js @@ -19,6 +19,7 @@ import { DicomMetadataListener } from "./utilities/DicomMetadataListener.js"; import { log } from "./log.js"; const readLog = log.getLogger("AsyncDicomReader"); +const NORMALIZED_READ_OPTIONS = Symbol("normalizedReadOptions"); /** * This is an asynchronous binary DICOM reader. @@ -30,6 +31,8 @@ const readLog = log.getLogger("AsyncDicomReader"); */ export class AsyncDicomReader { syntax = EXPLICIT_LITTLE_ENDIAN; + stopInfo = null; + pump = null; constructor(options = {}) { this.isLittleEndian = options?.isLittleEndian; @@ -44,6 +47,52 @@ export class AsyncDicomReader { /** Sentinel returned when stream is Part 10 but has no preamble (starts with meta). */ static PART10_NO_PREAMBLE = Symbol("PART10_NO_PREAMBLE"); + static async readFileFromAsyncStream(stream, options = {}) { + const { readerOptions, ...readOptions } = options; + const reader = new this(readerOptions); + return reader.readFileFromAsyncStream(stream, readOptions); + } + + async readFileFromAsyncStream(stream, options = {}) { + const { streamOptions: inputStreamOptions, ...readOptions } = options; + const streamOptions = { + readAheadHighWaterMark: 1024, + ...inputStreamOptions, + requireCancellation: true + }; + + const pump = this.stream.pumpAsyncStream(stream, streamOptions); + this.pump = { + abort: pump.abort, + cancellable: pump.cancellable, + failure: pump.failure, + finished: pump.finished, + get aborted() { + return pump.aborted; + }, + get reason() { + return pump.reason; + } + }; + try { + const result = await Promise.race([ + this.readFile(readOptions), + pump.failure + ]); + pump.stop(); + await pump.finished; + return result; + } catch (error) { + pump.abort(error); + try { + await pump.finished; + } catch { + // Preserve the parsing, source, or externally supplied error. + } + throw error; + } + } + /** * Reads the preamble and checks for the DICM marker. * Returns true if found/read, leaving the stream past the @@ -56,7 +105,7 @@ export class AsyncDicomReader { */ async readPreamble() { const { stream } = this; - await stream.ensureAvailable(); + await stream.ensureAvailable(132); stream.reset(); stream.increment(128); if (stream.readAsciiString(4) !== "DICM") { @@ -150,6 +199,8 @@ export class AsyncDicomReader { } async readFile(options = undefined) { + options = this.normalizeReadOptions(options); + this.stopInfo = null; const hasPreamble = await this.readPreamble(); if (hasPreamble === AsyncDicomReader.PART10_NO_PREAMBLE) { // Part 10 without preamble: stream starts at (0002,0000) or rest of meta @@ -215,9 +266,9 @@ export class AsyncDicomReader { */ async readMeta(options = undefined) { const { stream } = this; - await stream.ensureAvailable(); + await stream.ensureAvailable(12); const { offset: metaStartPos } = stream; - const el = this.readTagHeader(); + const el = this.readTagHeader(undefined, true); if (el.tag !== TagHex.FileMetaInformationGroupLength) { // meta length tag is missing if (!options?.ignoreErrors) { @@ -257,16 +308,29 @@ export class AsyncDicomReader { } async read(listener, options) { + options = this.normalizeReadOptions(options); const untilOffset = options?.untilOffset || Number.MAX_SAFE_INTEGER; + const checksAfterTag = + options.includeUntilTagValue || + typeof options.shouldStop === "function"; this.listener = listener; const { stream } = this; - await stream.ensureAvailable(); - while (stream.offset < untilOffset && stream.isAvailable(1, false)) { + await stream.ensureAvailable(12); + while ( + !this.stopInfo && + stream.offset < untilOffset && + stream.isAvailable(1, false) + ) { readLog.debug("read loop", stream.offset, untilOffset); // Consume before reading the tag so that data before the // current tag can be cleared. stream.consume(); - const tagInfo = this.readTagHeader(options); + const tagInfo = this.readTagHeader(options, true); + + if (tagInfo.isPastUntilTag) { + this.setStopInfo("stopOnGreaterTag", tagInfo); + break; + } // Stop when the requested tag boundary is reached. readTagHeader() // has already consumed the 4-byte tag but nothing beyond it, so @@ -274,6 +338,7 @@ export class AsyncDicomReader { // the VR field for explicit-LE). Callers that need the start // offset of the tag itself should subtract 4 from stream.offset. if (tagInfo.isUntilTag) { + this.setStopInfo("untilTag", tagInfo); break; } @@ -306,21 +371,74 @@ export class AsyncDicomReader { await this.readSingle(tagInfo, listener, options); } listener.pop(); - await this.stream.ensureAvailable(); + if (this.stopInfo) { + break; + } + if ( + checksAfterTag && + (await this.shouldStopAfterTag(tagInfo, listener, options)) + ) { + break; + } + await this.stream.ensureAvailable(12); } return listener.pop(); } + async shouldStopAfterTag(tagInfo, listener, options) { + if (options?.includeUntilTagValue && options.untilTag === tagInfo.tag) { + this.setStopInfo("untilTag", tagInfo); + return true; + } + + const { shouldStop } = options || {}; + if (typeof shouldStop !== "function") { + return false; + } + + const result = await shouldStop({ + tagInfo, + listener, + reader: this, + stream: this.stream + }); + if (!result) { + return false; + } + + this.setStopInfo("shouldStop", tagInfo); + return true; + } + + setStopInfo(reason, tagInfo) { + this.stopInfo = { + reason, + tag: tagInfo.tag, + offset: tagInfo.tagStartOffset, + tagStartOffset: tagInfo.tagStartOffset, + valueOffset: tagInfo.valueOffset, + valueLength: tagInfo.length, + stopOffset: this.stream.offset, + availableBytes: this.stream.available, + loadedEndOffset: this.stream.endOffset + }; + return this.stopInfo; + } + async readSequence(listener, sqTagInfo, options) { const { length } = sqTagInfo; - const { stream, syntax } = this; + const { stream } = this; const endOffset = length === UNDEFINED_LENGTH_FIX ? Number.MAX_SAFE_INTEGER : stream.offset + length; - while (stream.offset < endOffset && (await stream.ensureAvailable())) { + while ( + !this.stopInfo && + stream.offset < endOffset && + (await stream.ensureAvailable(12)) + ) { readLog.debug("readSequence loop", stream.offset, endOffset); - const tagInfo = this.readTagHeader(syntax, options); + const tagInfo = this.readTagHeader(undefined, true); const { tag } = tagInfo; if (tag === TagHex.Item) { listener.startObject(); @@ -335,10 +453,18 @@ export class AsyncDicomReader { itemLength === UNDEFINED_LENGTH_FIX ? endOffset : Math.min(stream.offset + itemLength, endOffset); - await this.read(listener, { + const itemOptions = { ...options, + untilTag: null, + includeUntilTagValue: false, + shouldStop: undefined, + stopOnGreaterTag: false, untilOffset: itemUntilOffset - }); + }; + await this.read(listener, itemOptions); + if (this.stopInfo) { + return; + } } else if (tag === TagHex.SequenceDelimitationEnd) { // Sequence of undefined lengths end in sequence delimitation item return; @@ -419,7 +545,7 @@ export class AsyncDicomReader { readLog.debug("readCompressed frame loop", frameNumber); stream.consume(); await stream.ensureAvailable(); - const frameTag = this.readTagHeader(); + const frameTag = this.readTagHeader(undefined, true); if (frameTag.tag === TagHex.SequenceDelimitationEnd) { if (lastFrame) { // Always deliver frames as arrays, using streaming splitFrame @@ -449,7 +575,7 @@ export class AsyncDicomReader { } async readOffsets() { - const tagInfo = this.readTagHeader(); + const tagInfo = this.readTagHeader(undefined, true); if (tagInfo.tag !== TagHex.Item) { throw new Error(`Offsets tag is missing: ${tagInfo.tag}`); } @@ -618,6 +744,24 @@ export class AsyncDicomReader { return vr === "SQ" || (vr === "UN" && length === UNDEFINED_LENGTH_FIX); } + normalizeReadOptions(options = {}) { + options ||= {}; + if (options[NORMALIZED_READ_OPTIONS]) { + return options; + } + const normalizedOptions = { + ...options, + untilTag: DicomMetaDictionary.normalizeTagOption( + options.untilTag, + "untilTag" + ) + }; + Object.defineProperty(normalizedOptions, NORMALIZED_READ_OPTIONS, { + value: true + }); + return normalizedOptions; + } + /** * Reads a tag header. */ @@ -625,22 +769,48 @@ export class AsyncDicomReader { options = { untilTag: null, includeUntilTagValue: false - } + }, + optionsAreNormalized = false ) { + if (!optionsAreNormalized) { + options = this.normalizeReadOptions(options); + } const { stream, syntax } = this; const { untilTag, includeUntilTagValue } = options; const implicit = syntax == IMPLICIT_LITTLE_ENDIAN; const isLittleEndian = syntax !== EXPLICIT_BIG_ENDIAN; stream.setEndian(isLittleEndian); + const tagStartOffset = stream.offset; const tagObj = Tag.readTag(stream); const tag = tagObj.cleanString; if (untilTag && untilTag === tag) { if (!includeUntilTagValue) { - return { tag, tagObj, vr: 0, values: 0, isUntilTag: true }; + return { + tag, + tagObj, + vr: 0, + values: 0, + isUntilTag: true, + tagStartOffset, + stopOffset: stream.offset + }; } } + if (untilTag && options.stopOnGreaterTag && tag > untilTag) { + stream.increment(tagStartOffset - stream.offset); + return { + tag, + tagObj, + vr: 0, + values: 0, + isPastUntilTag: true, + tagStartOffset, + stopOffset: stream.offset + }; + } + let length = null; let vr = null; let vrType; @@ -703,7 +873,9 @@ export class AsyncDicomReader { tagObj, vm: entry?.vm, name: entry?.name, - length: length === UNDEFINED_LENGTH ? -1 : length + length: length === UNDEFINED_LENGTH ? -1 : length, + tagStartOffset, + valueOffset: stream.offset }; return header; } diff --git a/src/BufferStream.js b/src/BufferStream.js index e9acf7ec..c34e5a16 100644 --- a/src/BufferStream.js +++ b/src/BufferStream.js @@ -11,9 +11,12 @@ export class BufferStream { view = new SplitDataView(); /** The available listeners are those waiting for a query response */ availableListeners = []; + readAheadListeners = []; + requestedAvailable = 0; /** Indicates if this buffer stream is complete/has finished being created */ isComplete = false; + failure = null; /** A flag to set to indicate to clear buffers as they get consumed */ clearBuffers = false; @@ -32,6 +35,12 @@ export class BufferStream { setComplete(value = true) { this.isComplete = value; this.notifyAvailableListeners(); + this.notifyReadAheadListeners(); + } + + setFailed(error) { + this.failure = error; + this.setComplete(); } /** @@ -53,10 +62,23 @@ export class BufferStream { * EOF. By default waits for at least 1k to be available. */ ensureAvailable(bytes = 1024) { + if (this.failure) { + return Promise.reject(this.failure); + } if (!this.isAvailable(bytes)) { - return new Promise(resolve => { + this.requestedAvailable = Math.max(this.requestedAvailable, bytes); + this.notifyReadAheadListeners(); + return new Promise((resolve, reject) => { const recheckAvailable = () => { + if (this.failure) { + reject(this.failure); + return; + } if (this.isAvailable(bytes)) { + if (this.requestedAvailable <= bytes) { + this.requestedAvailable = 0; + } + this.notifyReadAheadListeners(); resolve(true); return; } @@ -244,11 +266,10 @@ export class BufferStream { readUint16Array(length) { var sixlen = length / 2, - arr = new Uint16Array(sixlen), - i = 0; - while (i++ < sixlen) { + arr = new Uint16Array(sixlen); + for (let i = 0; i < sixlen; i++) { arr[i] = this.view.getUint16(this.offset, this.isLittleEndian); - this.offset += 2; + this.increment(2); } return arr; } @@ -339,7 +360,7 @@ export class BufferStream { new Uint8Array(stream.slice(stream.startOffset, stream.size)), this.offset ); - this.offset += stream.size; + this.increment(stream.size); this.size = this.offset; this.endOffset = this.size; return this.view.availableSize; @@ -350,21 +371,357 @@ export class BufferStream { if (this.offset > this.size) { this.size = this.offset; } + if (this.readAheadListeners.length > 0) { + this.notifyReadAheadListeners(); + } + if (step < 0 && this.availableListeners.length > 0) { + this.notifyAvailableListeners(); + } return step; } /** * Reads from an async stream delivering to addBuffer. */ - async fromAsyncStream(stream) { - for await (const chunk of stream) { - const ab = chunk.buffer.slice( + async fromAsyncStream(stream, options = {}) { + return this.pumpAsyncStream(stream, options).finished; + } + + /** + * Starts reading from an async stream and returns a handle that can abort + * the source without requiring callers to read the rest of it first. + */ + pumpAsyncStream(stream, options = {}) { + if ( + options.requireCancellation && + !BufferStream.isCancellableAsyncStream(stream) + ) { + throw new TypeError( + "Cancellable stream must be a Node stream or ReadableStream" + ); + } + const source = BufferStream.asyncSourceFromStream(stream); + let rejectFailure; + let resolveStop; + const failure = new Promise((_resolve, reject) => { + rejectFailure = reject; + }); + failure.catch(() => {}); + const stopSignal = new Promise(resolve => { + resolveStop = resolve; + }); + const state = { + cancelRequested: false, + cancelPromise: Promise.resolve(), + finalizing: false, + failure: undefined, + hasFailure: false, + settled: false, + stopSignal, + stopped: false + }; + + const reportFailure = error => { + if (state.hasFailure) { + return; + } + const failure = BufferStream.toError(error, "Async stream failed"); + state.failure = failure; + state.hasFailure = true; + rejectFailure(failure); + this.setFailed(failure); + }; + + const cancel = error => { + if (error !== undefined) { + reportFailure(error); + } + if (state.cancelRequested || state.finalizing || state.settled) { + return; + } + state.cancelRequested = true; + state.stopped = true; + resolveStop(); + const cancelSource = + error !== undefined ? source.abort : source.stop; + const reason = + error !== undefined + ? state.failure + : BufferStream.createStopReason(); + state.cancelPromise = BufferStream.cancelSource( + cancelSource, + reason + ); + if (error === undefined) { + this.setComplete(); + } + }; + state.cancel = cancel; + state.reportFailure = reportFailure; + state.removeErrorListener = source.onError?.(error => + cancel(BufferStream.toError(error, "Async source failed")) + ); + + const finished = this._pumpAsyncStream(source, state, options); + finished.catch(() => {}); + + return { + abort: error => cancel(BufferStream.toError(error)), + cancellable: source.cancellable, + failure, + finished, + stop: () => cancel(), + get aborted() { + return state.hasFailure; + }, + get reason() { + return state.failure; + }, + get stopped() { + return state.stopped; + } + }; + } + + async _pumpAsyncStream(source, state, options) { + let reachedEnd = false; + try { + while (!state.stopped) { + const next = Promise.resolve() + .then(() => source.next()) + .then(result => ({ result })); + const nextResult = await Promise.race([next, state.stopSignal]); + if (!nextResult || state.stopped) { + break; + } + const { done, value } = nextResult.result; + if (done) { + reachedEnd = true; + break; + } + this.addBuffer(BufferStream.arrayBufferFromChunk(value)); + await this.waitForReadAhead( + options.readAheadHighWaterMark, + state + ); + } + } catch (error) { + const expectedStopError = + state.stopped && + !state.hasFailure && + source.isExpectedStopError?.(error); + const isReportedFailure = + state.hasFailure && error === state.failure; + if (!expectedStopError && !isReportedFailure) { + state.cancel( + BufferStream.toError(error, "Async source failed") + ); + } + } finally { + if ( + reachedEnd && + options.requireCancellation && + !state.cancelRequested + ) { + state.cancel(); + } + state.finalizing = true; + try { + await state.cancelPromise; + } catch (error) { + state.reportFailure(error); + } finally { + try { + source.release?.(); + } catch (error) { + state.reportFailure(error); + } + state.removeErrorListener?.(); + state.settled = true; + if (!state.hasFailure) { + this.setComplete(); + } + } + } + + if (state.hasFailure) { + throw state.failure; + } + } + + static createStopReason() { + return new Error("BufferStream stopped"); + } + + static toError(error, fallbackMessage = "BufferStream aborted") { + if (error instanceof Error) { + return error; + } + return new Error(error ? String(error) : fallbackMessage); + } + + static cancelSource(cancelSource, reason) { + try { + return Promise.resolve(cancelSource?.(reason)); + } catch (error) { + return Promise.reject(error); + } + } + + static isCancellableAsyncStream(stream) { + if (typeof stream?.getReader === "function") { + return true; + } + return ( + typeof stream?.[Symbol.asyncIterator] === "function" && + typeof stream.destroy === "function" && + typeof stream.once === "function" && + typeof stream.on === "function" && + typeof stream.off === "function" + ); + } + + static destroyNodeStream(stream, reason) { + if ( + typeof stream.once !== "function" || + typeof stream.on !== "function" || + typeof stream.off !== "function" + ) { + stream.destroy(reason); + return Promise.resolve(); + } + if (stream.closed) { + return Promise.resolve(); + } + + return new Promise((resolve, reject) => { + let destroyError; + let isSettled = false; + let waitTimer; + const cleanup = () => { + clearTimeout(waitTimer); + stream.off("close", onClose); + stream.off("error", onError); + }; + const settle = () => { + if (isSettled) { + return; + } + isSettled = true; + cleanup(); + destroyError ? reject(destroyError) : resolve(); + }; + const onError = error => { + destroyError ||= error; + }; + const onClose = () => settle(); + const waitForClosed = () => { + if (stream.closed) { + waitTimer = setTimeout(settle, 0); + } else { + waitTimer = setTimeout(waitForClosed, 0); + } + }; + + stream.once("close", onClose); + stream.on("error", onError); + try { + stream.destroy(reason); + if (stream._readableState?.emitClose === false) { + waitForClosed(); + } + } catch (error) { + destroyError ||= error; + settle(); + } + }); + } + + static asyncSourceFromStream(stream) { + if (typeof stream?.getReader === "function") { + const reader = stream.getReader(); + return { + cancellable: true, + next: () => reader.read(), + abort: reason => reader.cancel(reason), + stop: reason => reader.cancel(reason), + release: () => reader.releaseLock() + }; + } + + const iterator = stream?.[Symbol.asyncIterator]?.(); + if (iterator) { + const hasDestroy = typeof stream.destroy === "function"; + const canObserveClose = + typeof stream.once === "function" && + typeof stream.on === "function" && + typeof stream.off === "function"; + return { + cancellable: hasDestroy && canObserveClose, + next: () => iterator.next(), + abort: hasDestroy + ? reason => BufferStream.destroyNodeStream(stream, reason) + : reason => iterator.return?.(reason), + stop: hasDestroy + ? () => BufferStream.destroyNodeStream(stream) + : () => iterator.return?.(), + onError: + typeof stream.on === "function" && + typeof stream.off === "function" + ? handler => { + stream.on("error", handler); + return () => stream.off("error", handler); + } + : undefined, + isExpectedStopError: hasDestroy + ? error => + error?.name === "AbortError" || + error?.code === "ABORT_ERR" || + error?.code === "ERR_STREAM_PREMATURE_CLOSE" || + error?.message === "Premature close" + : undefined + }; + } + + throw new TypeError( + "Async stream must be an async iterable or ReadableStream" + ); + } + + static arrayBufferFromChunk(chunk) { + if (chunk instanceof ArrayBuffer) { + return chunk; + } + if (ArrayBuffer.isView(chunk)) { + return chunk.buffer.slice( chunk.byteOffset, chunk.byteOffset + chunk.byteLength ); - this.addBuffer(ab); } - this.setComplete(); + throw new TypeError("Async stream chunks must be ArrayBuffer views"); + } + async waitForReadAhead(readAheadHighWaterMark, state) { + if (typeof readAheadHighWaterMark !== "number") { + return; + } + + while ( + !state.stopped && + this.isPastReadAheadLimit(readAheadHighWaterMark) + ) { + await new Promise(resolve => { + this.readAheadListeners.push(resolve); + }); + } + } + + isPastReadAheadLimit(readAheadHighWaterMark) { + const limit = this.readAheadLimit(readAheadHighWaterMark); + return limit === 0 ? this.available > 0 : this.available >= limit; + } + + readAheadLimit(readAheadHighWaterMark) { + return Math.max(0, readAheadHighWaterMark, this.requestedAvailable); } /** @@ -396,6 +753,15 @@ export class BufferStream { existingListeners.forEach(listener => listener()); } + notifyReadAheadListeners() { + if (this.readAheadListeners.length === 0) { + return; + } + const existingListeners = [...this.readAheadListeners]; + this.readAheadListeners.splice(0, this.readAheadListeners.length); + existingListeners.forEach(listener => listener()); + } + /** * Consumes the data up to the given offset. * This will clear the references to the data buffers, and will @@ -437,7 +803,7 @@ export class BufferStream { } reset() { - this.offset = 0; + this.increment(-this.offset); return this; } @@ -446,7 +812,7 @@ export class BufferStream { } toEnd() { - this.offset = this.view.byteLength; + this.increment(this.view.byteLength - this.offset); } /** @@ -495,7 +861,7 @@ export class ReadBufferStream extends BufferStream { } reset() { - this.offset = this.startOffset; + this.increment(this.startOffset - this.offset); return this; } @@ -504,7 +870,7 @@ export class ReadBufferStream extends BufferStream { } toEnd() { - this.offset = this.endOffset; + this.increment(this.endOffset - this.offset); } writeUint8(value) { diff --git a/src/DicomMessage.js b/src/DicomMessage.js index 95318cc1..e381d9f0 100644 --- a/src/DicomMessage.js +++ b/src/DicomMessage.js @@ -18,6 +18,7 @@ import { deepEqual } from "./utilities/deepEqual"; import { ValueRepresentation } from "./ValueRepresentation.js"; export const singleVRs = ["SQ", "OF", "OW", "OB", "UN", "LT"]; +const NORMALIZED_READ_OPTIONS = Symbol("normalizedReadOptions"); export class DicomMessage { static read( @@ -42,10 +43,15 @@ export class DicomMessage { includeUntilTagValue = false ) { log.warn("DicomMessage.readTag to be deprecated after dcmjs 0.24.x"); - return this._readTag(bufferStream, syntax, { - untilTag: untilTag, - includeUntilTagValue: includeUntilTagValue - }); + return this._readTag( + bufferStream, + syntax, + this._normalizeReadOptions({ + untilTag: untilTag, + includeUntilTagValue: includeUntilTagValue + }), + true + ); } static _read( @@ -58,6 +64,7 @@ export class DicomMessage { stopOnGreaterTag: false } ) { + options = DicomMessage._normalizeReadOptions(options); const { ignoreErrors, untilTag, stopOnGreaterTag } = options; var dict = {}; try { @@ -67,7 +74,8 @@ export class DicomMessage { const readInfo = DicomMessage._readTag( bufferStream, syntax, - options + options, + true ); const cleanTagString = readInfo.tag.toCleanString(); if (untilTag && stopOnGreaterTag && cleanTagString > untilTag) { @@ -164,7 +172,7 @@ export class DicomMessage { var metaStartPos = stream.offset; // read the first tag to check if it's the meta length tag - var el = DicomMessage._readTag(stream, useSyntax); + var el = DicomMessage._readTag(stream, useSyntax, undefined, true); var metaHeader = {}; if (el.tag.cleanString !== TagHex.FileMetaInformationGroupLength) { @@ -212,6 +220,24 @@ export class DicomMessage { return dicomDict; } + static _normalizeReadOptions(options = {}) { + options ||= {}; + if (options[NORMALIZED_READ_OPTIONS]) { + return options; + } + const normalizedOptions = { + ...options, + untilTag: DicomMetaDictionary.normalizeTagOption( + options.untilTag, + "untilTag" + ) + }; + Object.defineProperty(normalizedOptions, NORMALIZED_READ_OPTIONS, { + value: true + }); + return normalizedOptions; + } + static writeTagObject(stream, tagString, vr, values, syntax, writeOptions) { var tag = Tag.fromString(tagString); @@ -272,8 +298,12 @@ export class DicomMessage { options = { untilTag: null, includeUntilTagValue: false - } + }, + optionsAreNormalized = false ) { + if (!optionsAreNormalized) { + options = DicomMessage._normalizeReadOptions(options); + } const { untilTag, includeUntilTagValue } = options; var implicit = syntax == IMPLICIT_LITTLE_ENDIAN ? true : false, isLittleEndian = diff --git a/src/DicomMetaDictionary.js b/src/DicomMetaDictionary.js index 6a8132ba..87620b83 100644 --- a/src/DicomMetaDictionary.js +++ b/src/DicomMetaDictionary.js @@ -29,6 +29,26 @@ export class DicomMetaDictionary { return tag.substring(1, 10).replace(",", ""); } + static normalizeTag(tag) { + if (tag === null || tag === undefined) { + return tag; + } + const normalizedTag = String(tag) + .replace(/[(),\s]/g, "") + .toUpperCase(); + if (/^[0-9A-F]{8}$/.test(normalizedTag)) { + return normalizedTag; + } + } + + static normalizeTagOption(tag, optionName = "tag") { + const normalizedTag = DicomMetaDictionary.normalizeTag(tag); + if (tag !== null && tag !== undefined && !normalizedTag) { + throw new Error(`Invalid ${optionName}: ${tag}`); + } + return normalizedTag; + } + static parseIntFromTag(tag) { const integerValue = parseInt( "0x" + DicomMetaDictionary.unpunctuateTag(tag) diff --git a/test/DicomMetaDictionary.test.js b/test/DicomMetaDictionary.test.js index a6888329..b634530e 100644 --- a/test/DicomMetaDictionary.test.js +++ b/test/DicomMetaDictionary.test.js @@ -11,6 +11,73 @@ describe("DicomMetaDictionary", () => { unpunctuatedTag ); }); + + it("returns non-punctuated strings unchanged.", () => { + const originalTag = "PixelData"; + + expect(DicomMetaDictionary.unpunctuateTag(originalTag)).toBe( + originalTag + ); + }); + }); + + describe("punctuateTag", () => { + it("returns private dictionary keys with commas unchanged.", () => { + const privateTag = '(0009,"ACUSON",00)'; + + expect(DicomMetaDictionary.punctuateTag(privateTag)).toBe( + privateTag + ); + }); + }); + + describe("normalizeTag", () => { + it("returns clean uppercase tags for supported tag formats.", () => { + const expectedTag = "7FE00010"; + const tagFormats = [ + "7FE00010", + "7fe00010", + "(7FE0,0010)", + "(7fe0,0010)", + "7FE0,0010" + ]; + + tagFormats.forEach(tag => { + expect(DicomMetaDictionary.normalizeTag(tag)).toBe( + expectedTag + ); + }); + }); + + it("returns undefined for invalid tag strings.", () => { + const invalidTag = "PixelData"; + + expect(DicomMetaDictionary.normalizeTag(invalidTag)).toBe( + undefined + ); + }); + }); + + describe("normalizeTagOption", () => { + it("returns normalized tags for valid option values.", () => { + const originalTag = "(7fe0,0010)"; + const expectedTag = "7FE00010"; + + expect( + DicomMetaDictionary.normalizeTagOption(originalTag) + ).toBe(expectedTag); + }); + + it("throws for invalid option values.", () => { + const invalidTag = "PixelData"; + + expect(() => + DicomMetaDictionary.normalizeTagOption( + invalidTag, + "untilTag" + ) + ).toThrow("Invalid untilTag: PixelData"); + }); }); describe("parseIntFromTag", () => { diff --git a/test/anonymizer.test.js b/test/anonymizer.test.js index 3215170b..b057241b 100644 --- a/test/anonymizer.test.js +++ b/test/anonymizer.test.js @@ -1,6 +1,7 @@ import dcmjs from "../src/index.js"; import fs from "fs"; import { validationLog } from "./../src/log.js"; +import { readFileArrayBuffer } from "./testUtils.js"; // Ignore validation errors validationLog.setLevel(5); @@ -41,7 +42,7 @@ it("test_anonymization", () => { it("test_anonymization_no_change_ref", () => { // given - const arrayBuffer = fs.readFileSync("test/sample-sr.dcm").buffer; + const arrayBuffer = readFileArrayBuffer("test/sample-sr.dcm"); const dicomDict = DicomMessage.readFile(arrayBuffer); // multiple value name diff --git a/test/async-data.test.js b/test/async-data.test.js index cee1fab3..5315b756 100644 --- a/test/async-data.test.js +++ b/test/async-data.test.js @@ -1,4 +1,6 @@ import fs from "fs"; +import { Readable } from "stream"; +import { ReadableStream } from "stream/web"; import dcmjs from "../src/index.js"; import { TagHex, @@ -23,7 +25,675 @@ const { DicomMetadataListener } = dcmjs.utilities; // Ensure DicomMessage is set on DicomDict DicomDict.setDicomMessageClass(DicomMessage); +async function waitFor(predicate) { + for (let attempts = 0; attempts < 100; attempts++) { + if (predicate()) { + return; + } + await new Promise(resolve => setTimeout(resolve, 0)); + } + throw new Error("Timed out waiting for condition"); +} + +function createStalledNodeStream(buffer) { + let hasRead = false; + return new Readable({ + read() { + if (hasRead) { + return; + } + hasRead = true; + this.push(new Uint8Array(buffer)); + } + }); +} + describe("AsyncDicomReader", () => { + it("rejects unsupported stream sources instead of hanging.", async () => { + const reader = new AsyncDicomReader(); + + await expect( + reader.readFileFromAsyncStream(new ArrayBuffer(16)) + ).rejects.toThrow( + "Cancellable stream must be a Node stream or ReadableStream" + ); + }); + + it("rejects non-cancellable async iterables.", async () => { + let iteratorAcquisitions = 0; + const source = { + [Symbol.asyncIterator]() { + iteratorAcquisitions++; + return { + next() { + return Promise.resolve({ + done: false, + value: new Uint8Array([1, 2, 3]) + }); + } + }; + } + }; + const reader = new AsyncDicomReader(); + + await expect(reader.readFileFromAsyncStream(source)).rejects.toThrow( + "Cancellable stream must be a Node stream or ReadableStream" + ); + expect(iteratorAcquisitions).toBe(0); + }); + + it("rejects Node-like sources missing error observation.", async () => { + let iteratorAcquisitions = 0; + const source = { + destroy() {}, + off() {}, + once() {}, + [Symbol.asyncIterator]() { + iteratorAcquisitions++; + return { + next() { + return new Promise(() => {}); + } + }; + } + }; + const reader = new AsyncDicomReader(); + + await expect(reader.readFileFromAsyncStream(source)).rejects.toThrow( + "Cancellable stream must be a Node stream or ReadableStream" + ); + expect(iteratorAcquisitions).toBe(0); + }); + + it("preserves subclasses in the static stream factory.", async () => { + class CustomAsyncDicomReader extends AsyncDicomReader {} + + const buffer = createSampleDicom(); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(buffer)); + controller.close(); + } + }); + + const reader = await CustomAsyncDicomReader.readFileFromAsyncStream( + source, + { + untilTag: TagHex.PixelData, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + } + ); + + expect(reader).toBeInstanceOf(CustomAsyncDicomReader); + }); + + it("normalizes untilTag in direct readTagHeader calls.", () => { + const pixelDataTag = new Uint8Array([0xe0, 0x7f, 0x10, 0x00]); + const reader = new AsyncDicomReader(); + reader.stream.addBuffer(pixelDataTag.buffer); + reader.stream.setComplete(); + + const tagInfo = reader.readTagHeader({ + untilTag: "(7fe0,0010)" + }); + + expect(tagInfo.isUntilTag).toBe(true); + expect(tagInfo.tag).toBe(TagHex.PixelData); + }); + + it("rewinds direct readTagHeader calls that pass untilTag.", () => { + const rowsTag = new Uint8Array([0x28, 0x00, 0x10, 0x00]); + const reader = new AsyncDicomReader(); + reader.stream.addBuffer(rowsTag.buffer); + reader.stream.setComplete(); + + const tagInfo = reader.readTagHeader({ + untilTag: "00280009", + stopOnGreaterTag: true + }); + + expect(tagInfo.isPastUntilTag).toBe(true); + expect(tagInfo.stopOffset).toBe(0); + expect(reader.stream.offset).toBe(0); + }); + + it("stops a node stream before reading pixel data when untilTag matches.", async () => { + const filePath = "test/sample-dicom.dcm"; + const readAheadHighWaterMark = 12; + const highWaterMark = 4; + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + const stream = fs.createReadStream(filePath, { + highWaterMark + }); + + const { dict } = await reader.readFileFromAsyncStream(stream, { + listener, + untilTag: TagHex.PixelData, + includeUntilTagValue: false, + streamOptions: { readAheadHighWaterMark } + }); + + expect(dict[TagHex.Rows].Value[0]).toBe(512); + expect(dict[TagHex.PixelData]).toBeUndefined(); + expect(reader.stopInfo).toEqual( + expect.objectContaining({ + reason: "untilTag", + tag: TagHex.PixelData + }) + ); + expect(reader.stopInfo.valueOffset).toBeUndefined(); + expect(reader.stopInfo.stopOffset).toBe( + reader.stopInfo.tagStartOffset + 4 + ); + expect(reader.stopInfo.availableBytes).toBeGreaterThanOrEqual(0); + expect(reader.stopInfo.loadedEndOffset).toBe(reader.stream.endOffset); + expect(reader.stream.size).toBeLessThanOrEqual( + reader.stopInfo.tagStartOffset + readAheadHighWaterMark + ); + expect(stream.destroyed).toBe(true); + }); + + it("stops a Node stream with a pending read.", async () => { + const buffer = createSampleDicom(); + let hasRead = false; + const source = new Readable({ + read() { + if (!hasRead) { + hasRead = true; + this.push(new Uint8Array(buffer)); + } + } + }); + const reader = new AsyncDicomReader(); + + const { dict } = await reader.readFileFromAsyncStream(source, { + untilTag: TagHex.PixelData, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + }); + + expect(dict[TagHex.PixelData]).toBeUndefined(); + expect(source.destroyed).toBe(true); + expect(source.closed).toBe(true); + }); + + it("stops Node streams that suppress the close event.", async () => { + const buffer = createSampleDicom(); + let hasRead = false; + const source = new Readable({ + emitClose: false, + read() { + if (!hasRead) { + hasRead = true; + this.push(new Uint8Array(buffer)); + } + } + }); + const reader = new AsyncDicomReader(); + + const { dict } = await reader.readFileFromAsyncStream(source, { + untilTag: TagHex.PixelData, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + }); + + expect(dict[TagHex.PixelData]).toBeUndefined(); + expect(source.destroyed).toBe(true); + expect(source.closed).toBe(true); + }); + + it("rejects delayed Node teardown errors.", async () => { + const buffer = createSampleDicom(); + const teardownError = new Error("delayed close failure"); + let hasRead = false; + const source = new Readable({ + read() { + if (!hasRead) { + hasRead = true; + this.push(new Uint8Array(buffer)); + } + }, + destroy(_error, callback) { + setTimeout(() => callback(teardownError), 0); + } + }); + const reader = new AsyncDicomReader(); + + await expect( + reader.readFileFromAsyncStream(source, { + untilTag: TagHex.PixelData, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + }) + ).rejects.toBe(teardownError); + expect(source.closed).toBe(true); + }); + + it("rejects synchronous Node teardown errors.", async () => { + const buffer = createSampleDicom(); + const teardownError = new Error("synchronous close failure"); + let hasRead = false; + const source = new Readable({ + read() { + if (!hasRead) { + hasRead = true; + this.push(new Uint8Array(buffer)); + } + }, + destroy(_error, callback) { + callback(teardownError); + } + }); + const reader = new AsyncDicomReader(); + + await expect( + reader.readFileFromAsyncStream(source, { + untilTag: TagHex.PixelData, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + }) + ).rejects.toBe(teardownError); + expect(source.closed).toBe(true); + }); + + it("stops after reading an included untilTag value.", async () => { + const filePath = "test/sample-dicom.dcm"; + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + const stream = fs.createReadStream(filePath, { + highWaterMark: 4 + }); + + const { dict } = await reader.readFileFromAsyncStream(stream, { + listener, + untilTag: TagHex.Rows, + includeUntilTagValue: true, + streamOptions: { readAheadHighWaterMark: 12 } + }); + + expect(dict[TagHex.Rows].Value[0]).toBe(512); + expect(dict[TagHex.Columns]).toBeUndefined(); + expect(reader.stopInfo).toEqual( + expect.objectContaining({ + reason: "untilTag", + tag: TagHex.Rows + }) + ); + expect(stream.destroyed).toBe(true); + }); + + it("waits for a ReadableStream source to cancel before resolving.", async () => { + const buffer = createSampleDicom(); + const bytes = new Uint8Array(buffer); + const readAheadHighWaterMark = 12; + const highWaterMark = 4; + let offset = 0; + let isReadResolved = false; + let resolveCancel; + const source = new ReadableStream({ + pull(controller) { + if (offset >= bytes.length) { + controller.close(); + return; + } + const nextOffset = Math.min( + offset + highWaterMark, + bytes.length + ); + controller.enqueue(bytes.slice(offset, nextOffset)); + offset = nextOffset; + }, + cancel() { + return new Promise(resolve => { + resolveCancel = resolve; + }); + } + }); + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + + const readPromise = reader + .readFileFromAsyncStream(source, { + listener, + untilTag: TagHex.PixelData, + includeUntilTagValue: false, + streamOptions: { readAheadHighWaterMark } + }) + .then(result => { + isReadResolved = true; + return result; + }); + + await waitFor(() => resolveCancel); + await Promise.resolve(); + + expect(isReadResolved).toBe(false); + + resolveCancel(); + const { dict } = await readPromise; + + expect(dict[TagHex.PixelData]).toBeUndefined(); + expect(source.locked).toBe(false); + }); + + it("rejects when the source fails after enough bytes were buffered.", async () => { + const buffer = createSampleDicom(); + const sourceError = new Error("source failed"); + const source = createStalledNodeStream(buffer); + const reader = new AsyncDicomReader(); + let markShouldStopEntered; + const shouldStopEntered = new Promise(resolve => { + markShouldStopEntered = resolve; + }); + + const readPromise = reader.readFileFromAsyncStream(source, { + shouldStop: ({ tagInfo }) => { + if (tagInfo.tag !== TagHex.Rows) { + return false; + } + markShouldStopEntered(); + return new Promise(() => {}); + }, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + }); + + await shouldStopEntered; + source.emit("error", sourceError); + + await expect(readPromise).rejects.toBe(sourceError); + }); + + it("rejects abort-shaped source errors while parsing is paused.", async () => { + const buffer = createSampleDicom(); + const sourceError = new Error("upstream aborted"); + sourceError.name = "AbortError"; + const source = createStalledNodeStream(buffer); + const reader = new AsyncDicomReader(); + let markShouldStopEntered; + const shouldStopEntered = new Promise(resolve => { + markShouldStopEntered = resolve; + }); + + const readPromise = reader.readFileFromAsyncStream(source, { + shouldStop: ({ tagInfo }) => { + if (tagInfo.tag !== TagHex.Rows) { + return false; + } + markShouldStopEntered(); + return new Promise(() => {}); + }, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + }); + + await shouldStopEntered; + source.emit("error", sourceError); + + await expect(readPromise).rejects.toBe(sourceError); + }); + + it("rejects when the exposed pump is aborted externally.", async () => { + const externalError = new Error("STOW processing failed"); + const source = new ReadableStream({ + pull() { + return new Promise(() => {}); + } + }); + const reader = new AsyncDicomReader(); + const readPromise = reader.readFileFromAsyncStream(source); + + await waitFor(() => reader.pump); + reader.pump.abort(externalError); + + await expect(readPromise).rejects.toBe(externalError); + }); + + it("rejects a late external abort after source settlement.", async () => { + const buffer = createSampleDicom(); + const externalError = new Error("late STOW processing failure"); + let markShouldStopEntered; + const shouldStopEntered = new Promise(resolve => { + markShouldStopEntered = resolve; + }); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(buffer)); + controller.close(); + } + }); + const reader = new AsyncDicomReader(); + const readPromise = reader.readFileFromAsyncStream(source, { + shouldStop: ({ tagInfo }) => { + if (tagInfo.tag !== TagHex.Rows) { + return false; + } + markShouldStopEntered(); + return new Promise(() => {}); + }, + streamOptions: { + readAheadHighWaterMark: buffer.byteLength + 1 + } + }); + + await shouldStopEntered; + await reader.pump.finished; + reader.pump.abort(externalError); + + await expect(readPromise).rejects.toBe(externalError); + }); + + it("stops a node stream when a sorted tag passes the requested untilTag.", async () => { + const filePath = "test/sample-dicom.dcm"; + const missingTagBeforeRows = "(0028,0009)"; + const readAheadHighWaterMark = 12; + const highWaterMark = 4; + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + const stream = fs.createReadStream(filePath, { + highWaterMark + }); + + const { dict } = await reader.readFileFromAsyncStream(stream, { + listener, + untilTag: missingTagBeforeRows, + stopOnGreaterTag: true, + streamOptions: { readAheadHighWaterMark } + }); + + expect(dict[TagHex.Rows]).toBeUndefined(); + expect(dict[TagHex.PixelData]).toBeUndefined(); + expect(reader.stopInfo).toEqual( + expect.objectContaining({ + reason: "stopOnGreaterTag", + tag: TagHex.Rows + }) + ); + expect(reader.stopInfo.stopOffset).toBe(reader.stopInfo.tagStartOffset); + expect(stream.destroyed).toBe(true); + }); + + it("throws when untilTag is not a valid tag.", async () => { + const buffer = fs.readFileSync("test/sample-dicom.dcm"); + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + + reader.stream.addBuffer(buffer); + reader.stream.setComplete(); + + await expect( + reader.readFile({ + listener, + untilTag: "PixelData", + includeUntilTagValue: false + }) + ).rejects.toThrow("Invalid untilTag: PixelData"); + }); + + it("does not apply top-level sorted tag stops inside sequence items.", async () => { + const missingTagBeforeRows = "00280009"; + const referencedSeriesSequence = "00081115"; + const nestedRows = 7; + const buffer = createSampleDicom({ + dict: { + [referencedSeriesSequence]: { + vr: "SQ", + Value: [ + { + [TagHex.Rows]: { + vr: "US", + Value: [nestedRows] + } + } + ] + } + } + }); + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + + reader.stream.addBuffer(buffer); + reader.stream.setComplete(); + const { dict } = await reader.readFile({ + listener, + untilTag: missingTagBeforeRows, + stopOnGreaterTag: true + }); + + expect( + dict[referencedSeriesSequence].Value[0][TagHex.Rows].Value[0] + ).toBe(nestedRows); + expect(dict[TagHex.Rows]).toBeUndefined(); + expect(reader.stopInfo).toEqual( + expect.objectContaining({ + reason: "stopOnGreaterTag", + tag: TagHex.Rows + }) + ); + }); + + it("does not apply top-level untilTag matches inside sequence items.", async () => { + const referencedSeriesSequence = "00081115"; + const nestedRows = 7; + const buffer = createSampleDicom({ + dict: { + [referencedSeriesSequence]: { + vr: "SQ", + Value: [ + { + [TagHex.Rows]: { + vr: "US", + Value: [nestedRows] + } + } + ] + } + } + }); + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + + reader.stream.addBuffer(buffer); + reader.stream.setComplete(); + const { dict } = await reader.readFile({ + listener, + untilTag: TagHex.Rows, + includeUntilTagValue: false + }); + + expect( + dict[referencedSeriesSequence].Value[0][TagHex.Rows].Value[0] + ).toBe(nestedRows); + expect(dict[TagHex.Rows]).toBeUndefined(); + expect(reader.stopInfo).toEqual( + expect.objectContaining({ + reason: "untilTag", + tag: TagHex.Rows + }) + ); + }); + + it("does not apply top-level shouldStop callbacks inside sequence items.", async () => { + const referencedSeriesSequence = "00081115"; + const nestedRows = 7; + const buffer = createSampleDicom({ + dict: { + [referencedSeriesSequence]: { + vr: "SQ", + Value: [ + { + [TagHex.Rows]: { + vr: "US", + Value: [nestedRows] + } + } + ] + } + } + }); + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + + reader.stream.addBuffer(buffer); + reader.stream.setComplete(); + const { dict } = await reader.readFile({ + listener, + shouldStop: ({ tagInfo }) => tagInfo.tag === TagHex.Rows + }); + + expect( + dict[referencedSeriesSequence].Value[0][TagHex.Rows].Value[0] + ).toBe(nestedRows); + expect(dict[TagHex.Rows].Value[0]).toBe(defaultImage.rows); + expect(reader.stopInfo).toEqual( + expect.objectContaining({ + reason: "shouldStop", + tag: TagHex.Rows + }) + ); + }); + + it("stops a node stream after the shouldStop callback accepts a read tag.", async () => { + const filePath = "test/sample-dicom.dcm"; + const fileSize = fs.statSync(filePath).size; + const readAheadHighWaterMark = 12; + const highWaterMark = 4; + const reader = new AsyncDicomReader(); + const listener = new DicomMetadataListener(); + const stream = fs.createReadStream(filePath, { + highWaterMark + }); + + const { dict } = await reader.readFileFromAsyncStream(stream, { + listener, + shouldStop: ({ tagInfo }) => tagInfo.tag === TagHex.Rows, + streamOptions: { readAheadHighWaterMark } + }); + + expect(dict[TagHex.Rows].Value[0]).toBe(512); + expect(dict[TagHex.PixelData]).toBeUndefined(); + expect(reader.stopInfo).toEqual( + expect.objectContaining({ + reason: "shouldStop", + tag: TagHex.Rows + }) + ); + expect(reader.stopInfo.valueOffset).toBeLessThan( + reader.stopInfo.stopOffset + ); + expect(reader.stream.size).toBeLessThan(fileSize); + expect(stream.destroyed).toBe(true); + }); + test("DICOM part 10 complete listener uncompressed", async () => { const buffer = fs.readFileSync("test/sample-dicom.dcm"); const reader = new AsyncDicomReader(); diff --git a/test/data-options.test.js b/test/data-options.test.js index 8ab2e748..96a16266 100644 --- a/test/data-options.test.js +++ b/test/data-options.test.js @@ -44,6 +44,29 @@ it("test_untilTag", () => { expect(dataset.PixelData).toEqual(0); }); +it("normalizes untilTag before reading a file", () => { + const buffer = fs.readFileSync("test/sample-dicom.dcm"); + const dicomData = DicomMessage.readFile(buffer.buffer, { + untilTag: "(7fe0,0010)", + includeUntilTagValue: false + }); + + const dataset = DicomMetaDictionary.naturalizeDataset(dicomData.dict); + + expect(dataset.PixelData).toEqual(0); +}); + +it("throws when untilTag is not a valid tag", () => { + const buffer = fs.readFileSync("test/sample-dicom.dcm"); + + expect(() => + DicomMessage.readFile(buffer.buffer, { + untilTag: "PixelData", + includeUntilTagValue: false + }) + ).toThrow("Invalid untilTag: PixelData"); +}); + it("noCopy multiframe DICOM which has trailing padding", async () => { const url = "https://github.com/dcmjs-org/data/releases/download/binary-parsing-stressors/multiframe-ultrasound.dcm"; diff --git a/test/data.test.js b/test/data.test.js index a0864e5d..1503a06f 100644 --- a/test/data.test.js +++ b/test/data.test.js @@ -5,7 +5,11 @@ import path from "path"; import { WriteBufferStream } from "../src/BufferStream"; import dcmjs from "../src/index.js"; import { log } from "./../src/log.js"; -import { getTestDataset, getZippedTestDataset } from "./testUtils.js"; +import { + getTestDataset, + getZippedTestDataset, + readFileArrayBuffer +} from "./testUtils.js"; import { promisify } from "util"; import arrayItem from "./arrayItem.json"; @@ -985,8 +989,8 @@ describe("The same DICOM file loaded from both DCM and JSON", () => { let jsonData; beforeEach(() => { - const file = fs.readFileSync("test/sample-sr.dcm"); - dicomData = dcmjs.data.DicomMessage.readFile(file.buffer, { + const arrayBuffer = readFileArrayBuffer("test/sample-sr.dcm"); + dicomData = dcmjs.data.DicomMessage.readFile(arrayBuffer, { // ignoreErrors: true, }); jsonData = JSON.parse(JSON.stringify(sampleDicomSR)); diff --git a/test/readBufferStream.test.js b/test/readBufferStream.test.js index 6f44bf87..da27fd98 100644 --- a/test/readBufferStream.test.js +++ b/test/readBufferStream.test.js @@ -1,4 +1,6 @@ -import { ReadBufferStream } from "../src/BufferStream"; +import { ReadBufferStream, WriteBufferStream } from "../src/BufferStream"; +import { Readable } from "stream"; +import { ReadableStream } from "stream/web"; const size = 128; const buffer = new ArrayBuffer(size); @@ -7,6 +9,32 @@ for (let i = 0; i < size; i++) { dataView.setUint8(i, i % 256); } +function createCountingAsyncSource(chunks) { + let index = 0; + let nextCalls = 0; + return { + source: { + [Symbol.asyncIterator]() { + return { + next() { + nextCalls++; + if (index >= chunks.length) { + return Promise.resolve({ done: true }); + } + return Promise.resolve({ + done: false, + value: chunks[index++] + }); + } + }; + } + }, + get nextCalls() { + return nextCalls; + } + }; +} + describe("ReadBufferStream Tests", () => { it("reads single buffer", () => { const stream = new ReadBufferStream(buffer, true); @@ -154,5 +182,276 @@ describe("ReadBufferStream Tests", () => { expect(stream.isAvailable(remaining, false)).toBe(true); expect(stream.isAvailable(remaining + 1, false)).toBe(false); }); + + it("rechecks pending availability after reset.", async () => { + const stream = new ReadBufferStream(null, false); + stream.addBuffer(new ArrayBuffer(4)); + stream.increment(4); + const available = stream.ensureAvailable(4); + + stream.reset(); + + await expect(available).resolves.toBe(true); + }); + }); + + describe("async stream pumping", () => { + it("rejects unsupported stream sources synchronously.", () => { + const stream = new ReadBufferStream(null, false); + + expect(() => stream.pumpAsyncStream(new ArrayBuffer(16))).toThrow( + "Async stream must be an async iterable or ReadableStream" + ); + }); + + it("reads chunks from a ReadableStream source.", async () => { + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2])); + controller.enqueue(new Uint8Array([3])); + controller.close(); + } + }); + const stream = new ReadBufferStream(null, false); + + await stream.fromAsyncStream(source); + stream.reset(); + + expect(stream.readUint8()).toBe(1); + expect(stream.readUint8()).toBe(2); + expect(stream.readUint8()).toBe(3); + expect(source.locked).toBe(false); + }); + + it("retains low-level support for generic async iterables.", async () => { + const source = { + async *[Symbol.asyncIterator]() { + yield new Uint8Array([1, 2]); + yield new Uint8Array([3]); + } + }; + const stream = new ReadBufferStream(null, false); + + const pump = stream.pumpAsyncStream(source); + expect(pump.cancellable).toBe(false); + await pump.finished; + stream.reset(); + + expect(stream.readUint8()).toBe(1); + expect(stream.readUint8()).toBe(2); + expect(stream.readUint8()).toBe(3); + }); + + it("resumes a bounded pump as bytes are read.", async () => { + const countedSource = createCountingAsyncSource([ + new Uint8Array([1, 2]), + new Uint8Array([3, 4]) + ]); + const stream = new ReadBufferStream(null, false); + const pump = stream.pumpAsyncStream(countedSource.source, { + readAheadHighWaterMark: 2 + }); + + await stream.ensureAvailable(2); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(countedSource.nextCalls).toBe(1); + expect(stream.readUint8()).toBe(1); + expect(stream.readUint8()).toBe(2); + await stream.ensureAvailable(2); + expect(countedSource.nextCalls).toBe(2); + expect(stream.readUint8()).toBe(3); + expect(stream.readUint8()).toBe(4); + await pump.finished; + + expect(stream.isComplete).toBe(true); + expect(countedSource.nextCalls).toBe(3); + }); + + it("resumes a bounded pump after bulk cursor movement.", async () => { + const countedSource = createCountingAsyncSource([ + new Uint8Array([1, 2, 3, 4]), + new Uint8Array([5, 6]) + ]); + const stream = new ReadBufferStream(null, true); + const pump = stream.pumpAsyncStream(countedSource.source, { + readAheadHighWaterMark: 4 + }); + + await stream.ensureAvailable(4); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(countedSource.nextCalls).toBe(1); + const values = stream.readUint16Array(4); + expect(Array.from(values)).toEqual([513, 1027]); + await stream.ensureAvailable(2); + expect(countedSource.nextCalls).toBe(2); + stream.toEnd(); + await pump.finished; + + expect(stream.isComplete).toBe(true); + expect(countedSource.nextCalls).toBe(3); + }); + + it("resumes a bounded pump after concat advances the cursor.", async () => { + const countedSource = createCountingAsyncSource([ + new Uint8Array([1, 2, 3, 4]), + new Uint8Array([5, 6]) + ]); + const stream = new WriteBufferStream(4, false); + const consumed = new WriteBufferStream(4, false); + consumed.writeUint8Repeat(0, 4); + consumed.reset(); + const pump = stream.pumpAsyncStream(countedSource.source, { + readAheadHighWaterMark: 4 + }); + + await stream.ensureAvailable(4); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(countedSource.nextCalls).toBe(1); + + stream.concat(consumed); + await stream.ensureAvailable(2); + + expect(countedSource.nextCalls).toBe(2); + pump.stop(); + await pump.finished; + }); + + it("cancels a source error while the pump is waiting on read-ahead.", async () => { + const sourceError = new Error("source failed"); + let hasRead = false; + const source = new Readable({ + read() { + if (!hasRead) { + hasRead = true; + this.push(new Uint8Array([1, 2])); + } + } + }); + const stream = new ReadBufferStream(null, false); + const pump = stream.pumpAsyncStream(source, { + readAheadHighWaterMark: 2 + }); + + await stream.ensureAvailable(2); + source.emit("error", sourceError); + + await expect(pump.failure).rejects.toBe(sourceError); + await expect(pump.finished).rejects.toBe(sourceError); + }); + + it("applies the read-ahead high-water mark between source chunks.", async () => { + const countedSource = createCountingAsyncSource([ + new Uint8Array(8), + new Uint8Array(1) + ]); + const stream = new ReadBufferStream(null, false); + const pump = stream.pumpAsyncStream(countedSource.source, { + readAheadHighWaterMark: 2 + }); + + await stream.ensureAvailable(1); + await new Promise(resolve => setTimeout(resolve, 0)); + + expect(stream.size).toBe(8); + expect(countedSource.nextCalls).toBe(1); + + pump.stop(); + await pump.finished; + }); + + it("cancels a Web source when chunk processing fails.", async () => { + let wasCancelled = false; + const source = new ReadableStream({ + start(controller) { + controller.enqueue("invalid chunk"); + }, + cancel() { + wasCancelled = true; + } + }); + const stream = new ReadBufferStream(null, false); + const pump = stream.pumpAsyncStream(source); + + await expect(pump.finished).rejects.toThrow( + "Async stream chunks must be ArrayBuffer views" + ); + expect(wasCancelled).toBe(true); + expect(source.locked).toBe(false); + }); + + it("normalizes reasonless source rejections.", async () => { + const source = { + [Symbol.asyncIterator]() { + return { + next() { + return Promise.reject(); + } + }; + } + }; + const stream = new ReadBufferStream(null, false); + const pump = stream.pumpAsyncStream(source); + + await expect(pump.failure).rejects.toThrow("Async source failed"); + await expect(pump.finished).rejects.toThrow("Async source failed"); + }); + + it("cancels a ReadableStream source when stopped.", async () => { + let receivedStopReason; + let resolveCancel; + let isFinished = false; + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + }, + cancel(reason) { + receivedStopReason = reason; + return new Promise(resolve => { + resolveCancel = resolve; + }); + } + }); + const stream = new ReadBufferStream(null, false); + const pump = stream.pumpAsyncStream(source, { + readAheadHighWaterMark: 1 + }); + + await stream.ensureAvailable(1); + pump.stop(); + const finished = pump.finished.then(() => { + isFinished = true; + }); + await Promise.resolve(); + + expect(isFinished).toBe(false); + + resolveCancel(); + await finished; + + expect(receivedStopReason).toBeInstanceOf(Error); + expect(pump.aborted).toBe(false); + expect(pump.stopped).toBe(true); + expect(stream.isComplete).toBe(true); + expect(source.locked).toBe(false); + }); + + it("rejects when aborted by an external error.", async () => { + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + } + }); + const stream = new ReadBufferStream(null, false); + const pump = stream.pumpAsyncStream(source, { + readAheadHighWaterMark: 1 + }); + const externalError = new Error("storage failed"); + + await stream.ensureAvailable(1); + pump.abort(externalError); + + await expect(pump.finished).rejects.toBe(externalError); + expect(pump.aborted).toBe(true); + }); }); }); diff --git a/test/testUtils.js b/test/testUtils.js index 2d5c1337..0bfc7cb4 100644 --- a/test/testUtils.js +++ b/test/testUtils.js @@ -82,4 +82,20 @@ async function getTestDataset(url, filename) { return targetPath; } -export { getTestDataset, getZippedTestDataset }; +function bufferToArrayBuffer(buffer) { + return buffer.buffer.slice( + buffer.byteOffset, + buffer.byteOffset + buffer.byteLength + ); +} + +function readFileArrayBuffer(filePath) { + return bufferToArrayBuffer(fs.readFileSync(filePath)); +} + +export { + bufferToArrayBuffer, + getTestDataset, + getZippedTestDataset, + readFileArrayBuffer +};