diff --git a/lib/loading/ComponentRegistry.ts b/lib/loading/ComponentRegistry.ts index 4975c73..34c3338 100644 --- a/lib/loading/ComponentRegistry.ts +++ b/lib/loading/ComponentRegistry.ts @@ -20,6 +20,7 @@ export class ComponentRegistry { private readonly componentResources: Record; private readonly skipContextValidation: boolean; private readonly remoteContextLookups: boolean; + private registeredModules = 0; public constructor(options: IComponentLoaderRegistryOptions) { this.moduleState = options.moduleState; @@ -36,8 +37,10 @@ export class ComponentRegistry { * Effectively, this will load in all components that are available in the main package and its dependencies. */ public async registerAvailableModules(): Promise { + // Sorted, as blank node labels depend on the registration order, and modules are discovered in any order. await Promise.all(Object.values(this.moduleState.componentModules) .flatMap(x => Object.values(x)) + .sort((left, right) => left < right ? -1 : Number(left > right)) .map((moduleResourceUrl: string) => this.registerModule(moduleResourceUrl))); } @@ -46,9 +49,12 @@ export class ComponentRegistry { * @param urlOrPath An RDF document URL or local file path. */ public async registerModule(urlOrPath: string): Promise { + // Determined before loading, so that blank node labels only depend on the order of the calls. + const blankNodePrefix = `module_${this.registeredModules++}_`; const data = await RdfParser.fetchFileOrUrl(urlOrPath); await this.registerModuleStream(new RdfParser().parse(data, { path: urlOrPath, + blankNodePrefix, contexts: this.moduleState.contexts, importPaths: this.moduleState.importPaths, logger: this.logger, diff --git a/lib/loading/ConfigRegistry.ts b/lib/loading/ConfigRegistry.ts index 60a8690..6b7000c 100644 --- a/lib/loading/ConfigRegistry.ts +++ b/lib/loading/ConfigRegistry.ts @@ -15,6 +15,7 @@ export class ConfigRegistry { private readonly logger: Logger; private readonly skipContextValidation: boolean; private readonly remoteContextLookups: boolean; + private registeredConfigs = 0; public constructor(options: IConfigLoaderRegistryOptions) { this.moduleState = options.moduleState; @@ -29,9 +30,12 @@ export class ConfigRegistry { * @param urlOrPath An RDF document URL or local file path. */ public async register(urlOrPath: string): Promise { + // Determined before loading, so that blank node labels only depend on the order of the calls. + const blankNodePrefix = `config_${this.registeredConfigs++}_`; const data = await RdfParser.fetchFileOrUrl(urlOrPath); return await this.registerStream(new RdfParser().parse(data, { path: urlOrPath, + blankNodePrefix, contexts: this.moduleState.contexts, importPaths: this.moduleState.importPaths, ignoreImports: false, diff --git a/lib/rdf/RdfParser.ts b/lib/rdf/RdfParser.ts index 0ff9f92..7c196c8 100644 --- a/lib/rdf/RdfParser.ts +++ b/lib/rdf/RdfParser.ts @@ -1,6 +1,7 @@ import { createReadStream, promises as fs } from 'node:fs'; import type { Readable } from 'node:stream'; import type * as RDF from '@rdfjs/types'; +import { DataFactory } from 'rdf-data-factory'; import type { ParseOptions } from 'rdf-parse'; import { rdfParser } from 'rdf-parse'; import type { Logger } from 'winston'; @@ -57,7 +58,12 @@ export class RdfParser { }; // Execute parsing - const quadStream = rdfParser.parse(textStream, options); + const quadStream = rdfParser.parse(textStream, { + ...options, + dataFactory: options.dataFactory ?? (options.blankNodePrefix === undefined ? + undefined : + new DataFactory({ blankNodePrefix: options.blankNodePrefix })), + }); const includedQuadStream = quadStream.pipe(new RdfStreamIncluder(options)); quadStream.on('error', (error: Error) => includedQuadStream .emit('error', RdfParser.addPathToError(error, options.path))); @@ -115,6 +121,10 @@ export type RdfParserOptions = ParseOptions & { * Undefined if this file is the root file. */ importedFromPath?: string; + /** + * The prefix of the labels of blank nodes that are created while parsing. + */ + blankNodePrefix?: string; /** * An optional logger. */ diff --git a/lib/rdf/RdfStreamIncluder.ts b/lib/rdf/RdfStreamIncluder.ts index 85894d1..1279635 100644 --- a/lib/rdf/RdfStreamIncluder.ts +++ b/lib/rdf/RdfStreamIncluder.ts @@ -11,9 +11,8 @@ import type { RdfParserOptions } from './RdfParser'; * A RdfStreamIncluder takes a triple stream and detects owl:includes to automatically include other files. */ export class RdfStreamIncluder extends Transform { - private runningImporters = 1; private readonly parserOptions: RdfParserOptions; - private flushCallback: TransformCallback | undefined; + private readonly imports: Promise[] = []; public constructor(parserOptions: RdfParserOptions) { super({ objectMode: true }); @@ -31,10 +30,19 @@ export class RdfStreamIncluder extends Transform { // eslint-disable-next-line ts/naming-convention public _flush(callback: TransformCallback): void { - if (--this.runningImporters === 0) { - callback(); - } else { - this.flushCallback = callback; + this.flushImports().then(() => callback(), () => { + // The error has already been emitted + }); + } + + /** + * Emit the quads of all imports in a deterministic order + */ + protected async flushImports(): Promise { + for (const importStream of this.imports) { + for await (const quad of await importStream) { + this.push(quad); + } } } @@ -44,7 +52,6 @@ export class RdfStreamIncluder extends Transform { */ public handleImports(quad: RDF.Quad): void { if (!this.parserOptions.ignoreImports && quad.predicate.value === IRIS_RDFS.seeAlso) { - this.runningImporters++; let relativeFilePath = decodeURI(quad.object.value); // Try overriding path using defined import paths @@ -60,26 +67,36 @@ export class RdfStreamIncluder extends Transform { } } - // Recursively call the parser - RdfParser.fetchFileOrUrl(relativeFilePath) - .then((rawStream: Readable) => { - const data: Readable = new RdfParser().parse(rawStream, { - ...this.parserOptions, - baseIRI: undefined, - path: relativeFilePath, - importedFromPath: this.parserOptions.path, - }); - data - .on('data', (subData: RDF.Quad) => this.push(subData)) - .on('error', (error: Error): boolean => this.emit('error', error)) - .on('end', () => { - if (this.flushCallback && --this.runningImporters === 0) { - this.flushCallback(); - } - }); - }) - .catch((error: Error) => this.emit('error', RdfParser.addPathToError(error, this.parserOptions.path))); + // Recursively call the parser. + // Errors are emitted immediately, so they don't have to wait for preceding imports. + const importStream = this.parseImport(relativeFilePath, this.imports.length); + importStream.catch((error: Error) => this.emit('error', error)); + this.imports.push(importStream); + } + } + + /** + * Start parsing the given imported file. + * @param relativeFilePath The path or URL of the imported file. + * @param index The position of the import among the imports of the current stream. + */ + protected async parseImport(relativeFilePath: string, index: number): Promise { + let rawStream: Readable; + try { + rawStream = await RdfParser.fetchFileOrUrl(relativeFilePath); + } catch (error: unknown) { + throw RdfParser.addPathToError( error, this.parserOptions.path); } + return new RdfParser().parse(rawStream, { + ...this.parserOptions, + baseIRI: undefined, + path: relativeFilePath, + importedFromPath: this.parserOptions.path, + // Labels of the current stream are its prefix followed by digits, so these never clash. + blankNodePrefix: this.parserOptions.blankNodePrefix === undefined ? + undefined : + `${this.parserOptions.blankNodePrefix}${index}_`, + }).on('error', (error: Error) => this.emit('error', error)); } /** diff --git a/test/unit/loading/ComponentRegistry-test.ts b/test/unit/loading/ComponentRegistry-test.ts index 1a45ae8..cfd90d7 100644 --- a/test/unit/loading/ComponentRegistry-test.ts +++ b/test/unit/loading/ComponentRegistry-test.ts @@ -5,6 +5,7 @@ import { RdfObjectLoader } from 'rdf-object'; import type { Logger } from 'winston'; import { ComponentRegistry } from '../../../lib/loading/ComponentRegistry'; import type { IModuleState } from '../../../lib/loading/ModuleStateBuilder'; +import { RdfParser } from '../../../lib/rdf/RdfParser'; describe('ComponentRegistry', () => { let moduleState: IModuleState; @@ -54,9 +55,33 @@ describe('ComponentRegistry', () => { expect(Object.keys(objectLoader.resources) .includes('http://example.org/HelloWorldModule#SayHelloComponentNested')).toBeTruthy(); }); + + it('should register discovered modules in sorted order', async() => { + const registerModule = jest.spyOn(componentRegistry, 'registerModule').mockResolvedValue(); + moduleState.componentModules = { + B: { 1: 'b1', 2: 'b2' }, + A: { 1: 'a1' }, + C: { 1: 'c1' }, + }; + await componentRegistry.registerAvailableModules(); + expect(registerModule.mock.calls.map(args => args[0])).toEqual([ 'a1', 'b1', 'b2', 'c1' ]); + }); }); describe('registerModule', () => { + it('should assign blank node prefixes in the order of the calls', async() => { + const parse = jest.spyOn(RdfParser.prototype, 'parse'); + await Promise.all([ + componentRegistry.registerModule(Path.join(__dirname, '../../assets/module.jsonld')), + componentRegistry.registerModule(Path.join(__dirname, '../../assets/module-dynamicentries.jsonld')), + ]); + expect(parse.mock.calls.map(args => [ args[1].path, args[1].blankNodePrefix ])).toEqual(expect.arrayContaining([ + [ Path.join(__dirname, '../../assets/module.jsonld').replaceAll('\\', '/'), 'module_0_' ], + [ Path.join(__dirname, '../../assets/module-dynamicentries.jsonld').replaceAll('\\', '/'), 'module_1_' ], + ])); + parse.mockRestore(); + }); + it('should handle a valid module file', async() => { await componentRegistry.registerModule(Path.join(__dirname, '../../assets/module.jsonld')); expect(Object.keys(objectLoader.resources) diff --git a/test/unit/loading/ConfigRegistry-test.ts b/test/unit/loading/ConfigRegistry-test.ts index b5f79ab..5b0696f 100644 --- a/test/unit/loading/ConfigRegistry-test.ts +++ b/test/unit/loading/ConfigRegistry-test.ts @@ -5,6 +5,7 @@ import { stringToTerm } from 'rdf-string'; import type { Logger } from 'winston'; import { ConfigRegistry } from '../../../lib/loading/ConfigRegistry'; import type { IModuleState } from '../../../lib/loading/ModuleStateBuilder'; +import { RdfParser } from '../../../lib/rdf/RdfParser'; describe('ConfigRegistry', () => { let moduleState: IModuleState; @@ -42,6 +43,19 @@ describe('ConfigRegistry', () => { it('should throw on an invalid module file', async() => { await expect(configRegistry.register(`not-exists.jsonld`)).rejects.toThrow(/ENOENT/u); }); + + it('should assign blank node prefixes in the order of the calls', async() => { + const parse = jest.spyOn(RdfParser.prototype, 'parse'); + await Promise.all([ + configRegistry.register(Path.join(__dirname, '../../assets/config.jsonld')), + configRegistry.register(Path.join(__dirname, '../../assets/config-lazy.jsonld')), + ]); + expect(parse.mock.calls.map(args => [ args[1].path, args[1].blankNodePrefix ])).toEqual(expect.arrayContaining([ + [ Path.join(__dirname, '../../assets/config.jsonld').replaceAll('\\', '/'), 'config_0_' ], + [ Path.join(__dirname, '../../assets/config-lazy.jsonld').replaceAll('\\', '/'), 'config_1_' ], + ])); + parse.mockRestore(); + }); }); describe('registerCustom', () => { diff --git a/test/unit/rdf/RdfParser-test.ts b/test/unit/rdf/RdfParser-test.ts index d7e19af..4cba72f 100644 --- a/test/unit/rdf/RdfParser-test.ts +++ b/test/unit/rdf/RdfParser-test.ts @@ -1,4 +1,7 @@ import * as Path from 'node:path'; +import { PassThrough } from 'node:stream'; +import type * as RDF from '@rdfjs/types'; +import { DataFactory } from 'rdf-data-factory'; import type { RdfParserOptions } from '../../../lib/rdf/RdfParser'; import { RdfParser } from '../../../lib/rdf/RdfParser'; import 'jest-rdf'; @@ -9,7 +12,21 @@ const quad = require('rdf-quad'); const stringifyStream = require('stream-to-string'); const streamifyString = require('streamify-string'); +// The body of http://example.org/myfilestreaming.ttl, which tests write to themselves +let resolveStreamingBody: (body: PassThrough) => void; +let streamingBody: Promise; + globalThis.fetch = jest.fn(async(url: string) => { + if (url === 'http://example.org/myfilestreaming.ttl') { + const body = new PassThrough(); + resolveStreamingBody(body); + return { + body, + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } if (url === 'http://example.org/myfile1.ttl') { return { body: streamifyString(` .`), @@ -34,6 +51,41 @@ globalThis.fetch = jest.fn(async(url: string) => { statusText: 'OK', }; } + if (url === 'http://example.org/myfileslow.ttl') { + await new Promise(resolve => setTimeout(resolve, 50)); + return { + body: streamifyString(` .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } + if (url === 'http://example.org/myfileslownest.ttl') { + await new Promise(resolve => setTimeout(resolve, 50)); + return { + body: streamifyString(` .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } + if (url === 'http://example.org/myfileblank.ttl') { + return { + body: streamifyString(`[] . + .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } + if (url === 'http://example.org/myfileblanknest.ttl') { + return { + body: streamifyString(`[] .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } if (url === 'http://example.org/myfileerror.ttl') { return { body: streamifyString(`<...`), @@ -296,6 +348,130 @@ describe('RdfParser', () => { ]); }); + it('for a Turtle stream with imports should emit imports in their order, regardless of loading time', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + . + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('ex:s', 'ex:p', 'ex:o'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileslow.ttl'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile2.ttl'), + quad('ex:s3', 'ex:p3', 'ex:o3'), + quad('ex:s2', 'ex:p2', 'ex:o2'), + ]); + }); + + it('for a Turtle stream with nested imports should emit each import before the next one', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileslownest.ttl'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile2.ttl'), + quad('ex:s3', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile1.ttl'), + quad('ex:s1', 'ex:p1', 'ex:o1'), + quad('ex:s2', 'ex:p2', 'ex:o2'), + ]); + }); + + it('for a Turtle stream with an import should emit quads of the import while it is being parsed', async() => { + streamingBody = new Promise(resolve => resolveStreamingBody = resolve); + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + . +`; + const quads: RDF.Quad[] = []; + let resolveImportedQuad: () => void; + const importedQuad = new Promise(resolve => resolveImportedQuad = resolve); + const stream = parser.parse(streamifyString(doc), options) + .on('data', (data: RDF.Quad) => { + quads.push(data); + if (data.subject.value === 'ex:s4') { + resolveImportedQuad(); + } + }); + const ended = new Promise(resolve => stream.on('end', resolve)); + + // The first quad of the import is emitted before the import has ended + const body = await streamingBody; + body.write(` .\n`); + await importedQuad; + await new Promise(resolve => setImmediate(resolve)); + expect(quads).toEqualRdfQuadArray([ + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfilestreaming.ttl'), + quad('ex:s4', 'ex:p4', 'ex:o4'), + ]); + + body.end(` .\n`); + await ended; + expect(quads).toEqualRdfQuadArray([ + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfilestreaming.ttl'), + quad('ex:s4', 'ex:p4', 'ex:o4'), + quad('ex:s5', 'ex:p5', 'ex:o5'), + ]); + }); + + it('for a Turtle stream with a slow import followed by an import to unknown file', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).rejects + .toThrow(new Error(`Error while parsing file "path/to/file.ttl": URL not found: http://example.org/myfileunknown.ttl`)); + }); + + it('for a Turtle stream with imports and a blank node prefix should derive prefixes for imports', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + blankNodePrefix: 'p_', + }; + const doc = ` +[] . + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('_:p_0', 'ex:p', 'ex:o'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile1.ttl'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileblank.ttl'), + quad('ex:s1', 'ex:p1', 'ex:o1'), + quad('_:p_1_0', 'ex:p1', 'ex:o1'), + quad('ex:s1', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileblanknest.ttl'), + quad('_:p_1_0_0', 'ex:p2', 'ex:o2'), + ]); + }); + + it('for a Turtle stream with a blank node prefix and a data factory should use the data factory', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + blankNodePrefix: 'p_', + dataFactory: new DataFactory({ blankNodePrefix: 'f_' }), + }; + const doc = ` +[] . + . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('_:f_0', 'ex:p', 'ex:o'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileblanknest.ttl'), + quad('_:f_1', 'ex:p2', 'ex:o2'), + ]); + }); + it('for a Turtle stream with imports, when ignoring imports', async() => { const options: RdfParserOptions = { path: 'path/to/file.ttl',