Pass datasource metadata as a property.

RubenVerborgh · RubenVerborgh · commit 315cf87d1407 · 2016-09-26T15:16:47.000+02:00
With the previous event-based implementation,
views might miss the `metadata` event if they attach late.
diff --git a/bin/generate-summary b/bin/generate-summary
@@ -85,7 +85,7 @@ function fromDataSource(uri, datasource, callback, end, chunksize) {
     var count = 0;
 
     var stream = datasource.select({ limit: limit, offset: offset }, console.error)
-        .on('metadata', function (metadata) {
+        .getProperty('metadata', function (metadata) {
           var progress = Math.round((offset / metadata.totalCount) * 100);
           console.log(progress);
 
diff --git a/lib/controllers/SummaryController.js b/lib/controllers/SummaryController.js
@@ -42,7 +42,7 @@ SummaryController.prototype._handleRequest = function (request, response, next)
 
     // Render the summary
     var view = this._negotiateView('Summary', request, response);
-    view.render({ prefixes: this._prefixes, resultStream: streamParser }, request, response);
+    view.render({ prefixes: this._prefixes, results: streamParser }, request, response);
   }
   else
     next();
diff --git a/lib/controllers/TriplePatternFragmentsController.js b/lib/controllers/TriplePatternFragmentsController.js
@@ -36,8 +36,8 @@ TriplePatternFragmentsController.prototype._handleRequest = function (request, r
   // Generate the query result
   var view = this._negotiateView('TriplePatternFragments', request, response),
       settings = this._createFragmentMetadata(request, query, datasourceSettings);
-  settings.resultStream = datasourceSettings.datasource.select(query,
-                          function (error) { error && next(error); });
+  settings.results = datasourceSettings.datasource.select(query,
+                     function (error) { error && next(error); });
 
   // Execute the extensions and render the query result
   var extensions = this._extensions, extensionId = 0;
diff --git a/lib/datasources/CompositeDatasource.js b/lib/datasources/CompositeDatasource.js
@@ -89,60 +89,64 @@ CompositeDatasource.prototype._getDatasourceInfo = function (query, absoluteOffs
       // We checked all datasources, return our accumulated information
       callback(chosenDatasource, chosenOffset, totalCount, hasExactCount);
     else {
-      var ignoreTriples = { _push: noop, close: noop };
       var datasource = self._getDatasourceById(datasourceIndex);
-      datasource._executeQuery(emptyQuery, ignoreTriples, function (metadata) {
-        var count = metadata.totalCount;
-        var exact = metadata.hasExactCount;
-        // If we are still looking for an appropriate datasource, we need exact counts!
-        if (offset > 0 && !exact) {
-          self._getExactCount(datasource, query, function (exactCount) {
-            count = exactCount;
-            exact = true;
-            continueRecursion();
-          });
-        }
-        else
-          continueRecursion();
-
-        function continueRecursion() {
-          if (chosenDatasource < 0 && offset < count) {
-            // We can start querying from this datasource
-            setImmediate(function () {
-              findRecursive(datasourceIndex + 1, offset - count, datasourceIndex, offset,
-                  totalCount + count, hasExactCount && exact);
+      var metadataReader = {
+        _push: noop,
+        close: noop,
+        setProperty: function (name, metadata) {
+          if (name !== 'metadata') return;
+          // If we are still looking for an appropriate datasource, we need exact counts
+          var count = metadata.totalCount, exact = metadata.hasExactCount;
+          if (offset > 0 && !exact) {
+            self._getExactCount(datasource, query, function (exactCount) {
+              count = exactCount;
+              exact = true;
+              continueRecursion();
             });
           }
-          else {
-            // We forward our accumulated information and go check the next datasource
-            setImmediate(function () {
-              findRecursive(datasourceIndex + 1, offset - count, chosenDatasource, chosenOffset,
-                  totalCount + count, hasExactCount && exact);
-            });
+          else
+            continueRecursion();
+
+          function continueRecursion() {
+            if (chosenDatasource < 0 && offset < count) {
+              // We can start querying from this datasource
+              setImmediate(function () {
+                findRecursive(datasourceIndex + 1, offset - count, datasourceIndex, offset,
+                    totalCount + count, hasExactCount && exact);
+              });
+            }
+            else {
+              // We forward our accumulated information and go check the next datasource
+              setImmediate(function () {
+                findRecursive(datasourceIndex + 1, offset - count, chosenDatasource, chosenOffset,
+                    totalCount + count, hasExactCount && exact);
+              });
+            }
           }
-        }
-      });
+        },
+      };
+      datasource._executeQuery(emptyQuery, metadataReader);
     }
   }
 };
 
 // Writes the results of the query to the given triple stream
-CompositeDatasource.prototype._executeQuery = function (query, destination, metadataCallback) {
+CompositeDatasource.prototype._executeQuery = function (query, destination) {
   var offset =  query.offset || 0, limit = query.limit || Infinity;
   var self = this;
   this._getDatasourceInfo(query, offset, function (datasourceIndex, relativeOffset, totalCount, hasExactCount) {
     if (datasourceIndex < 0) {
       // No valid datasource has been found
-      metadataCallback({ totalCount: totalCount, hasExactCount: hasExactCount });
+      destination.setProperty('metadata', { totalCount: totalCount, hasExactCount: hasExactCount });
       destination.close();
     }
     else {
       // Send query to first applicable datasource and optionally emit triples from consecutive datasources
-      metadataCallback({ totalCount: totalCount, hasExactCount: hasExactCount });
-      var emitted = 0;
+      destination.setProperty('metadata', { totalCount: totalCount, hasExactCount: hasExactCount });
 
       // Modify our triple stream so that if all results from one datasource have arrived,
       // check if we haven't reached the limit and if so, trigger a new query for the next datasource.
+      var emitted = 0;
       countItems(destination, function (localEmittedCount) {
         // This is called after the last element has been pushed
 
diff --git a/lib/datasources/Datasource.js b/lib/datasources/Datasource.js
@@ -112,20 +112,17 @@ Datasource.prototype.select = function (query, onError) {
     if (triple.object[0]  === '_') triple.object  = blankNodePrefix + triple.object.substr(2);
     return triple;
   });
+  outputTriples.copyProperties(destination, ['metadata']);
   onError && outputTriples.on('error', onError);
 
   // Execute the query
-  try {
-    this._executeQuery(query, destination, function (metadata) {
-      setImmediate(function () { outputTriples.emit('metadata', metadata); });
-    });
-  }
+  try { this._executeQuery(query, destination); }
   catch (error) { outputTriples.emit('error', error); }
   return outputTriples;
 };
 
 // Writes the results of the query to the given destination
-Datasource.prototype._executeQuery = function (query, destination, metadataCallback) {
+Datasource.prototype._executeQuery = function (query, destination) {
   throw new Error('_executeQuery has not been implemented');
 };
 
diff --git a/lib/datasources/ExternalHdtDatasource.js b/lib/datasources/ExternalHdtDatasource.js
@@ -33,7 +33,7 @@ ExternalHdtDatasource.prototype._initialize = function (done) {
 };
 
 // Writes the results of the query to the given triple stream
-ExternalHdtDatasource.prototype._executeQuery = function (query, destination, metadataCallback) {
+ExternalHdtDatasource.prototype._executeQuery = function (query, destination) {
   // Execute the external HDT utility
   var hdtFile = this._hdtFile, offset = query.offset || 0, limit = query.limit || Infinity,
       hdt = spawn(hdtUtility, ['--query', (query.subject   || '?s') + ' ' +
@@ -52,7 +52,7 @@ ExternalHdtDatasource.prototype._executeQuery = function (query, destination, me
       // Ensure the estimated total count is as least as large as the number of triples
       if (tripleCount && estimatedTotalCount < offset + tripleCount)
         estimatedTotalCount = offset + (tripleCount < query.limit ? tripleCount : 2 * tripleCount);
-      metadataCallback({ totalCount: estimatedTotalCount, hasExactCount: hasExactCount });
+      destination.setProperty('metadata', { totalCount: estimatedTotalCount, hasExactCount: hasExactCount });
       destination.close();
     }
   });
diff --git a/lib/datasources/HdtDatasource.js b/lib/datasources/HdtDatasource.js
@@ -28,7 +28,7 @@ HdtDatasource.prototype._initialize = function (done) {
 };
 
 // Writes the results of the query to the given triple stream
-HdtDatasource.prototype._executeQuery = function (query, destination, metadataCallback) {
+HdtDatasource.prototype._executeQuery = function (query, destination) {
   this._hdtDocument.search(query.subject, query.predicate, query.object,
                            { limit: query.limit, offset: query.offset },
     function (error, triples, estimatedTotalCount, hasExactCount) {
@@ -37,7 +37,7 @@ HdtDatasource.prototype._executeQuery = function (query, destination, metadataCa
       var tripleCount = triples.length, offset = query.offset || 0;
       if (tripleCount && estimatedTotalCount < offset + tripleCount)
         estimatedTotalCount = offset + (tripleCount < query.limit ? tripleCount : 2 * tripleCount);
-      metadataCallback({ totalCount: estimatedTotalCount, hasExactCount: hasExactCount });
+      destination.setProperty('metadata', { totalCount: estimatedTotalCount, hasExactCount: hasExactCount });
       // Add the triples to the output
       for (var i = 0; i < tripleCount; i++)
         destination._push(triples[i]);
diff --git a/lib/datasources/MemoryDatasource.js b/lib/datasources/MemoryDatasource.js
@@ -24,11 +24,11 @@ MemoryDatasource.prototype._getAllTriples = function (addTriple, done) {
 };
 
 // Writes the results of the query to the given triple stream
-MemoryDatasource.prototype._executeQuery = function (query, destination, metadataCallback) {
-  var offset =  query.offset || 0, limit = query.limit || Infinity,
+MemoryDatasource.prototype._executeQuery = function (query, destination) {
+  var offset = query.offset || 0, limit = query.limit || Infinity,
       triples = this._tripleStore.findByIRI(query.subject, query.predicate, query.object);
   // Send the metadata
-  metadataCallback({ totalCount: triples.length, hasExactCount: true });
+  destination.setProperty('metadata', { totalCount: triples.length, hasExactCount: true });
   // Send the requested subset of triples
   for (var i = offset, l = Math.min(offset + limit, triples.length); i < l; i++)
     destination._push(triples[i]);
diff --git a/lib/datasources/SparqlDatasource.js b/lib/datasources/SparqlDatasource.js
@@ -26,7 +26,7 @@ function SparqlDatasource(options) {
 Datasource.extend(SparqlDatasource, ['triplePattern', 'limit', 'offset', 'totalCount']);
 
 // Writes the results of the query to the given triple stream
-SparqlDatasource.prototype._executeQuery = function (query, destination, metadataCallback) {
+SparqlDatasource.prototype._executeQuery = function (query, destination) {
   // Create the HTTP request
   var sparqlPattern = this._createTriplePattern(query), self = this,
       constructQuery = this._createConstructQuery(sparqlPattern, query.offset, query.limit),
@@ -60,8 +60,10 @@ SparqlDatasource.prototype._executeQuery = function (query, destination, metadat
 
   // Determine the total number of matching triples
   this._getPatternCount(sparqlPattern, function (error, totalCount) {
-    if (error) emitError(error);
-    else if (typeof totalCount === 'number') metadataCallback({ totalCount: totalCount, hasExactCount: true });
+    if (error)
+      emitError(error);
+    else if (typeof totalCount === 'number')
+      destination.setProperty('metadata', { totalCount: totalCount, hasExactCount: true });
   });
 
   // Emits an error on the triple stream
diff --git a/lib/views/summary/SummaryRdfView.js b/lib/views/summary/SummaryRdfView.js
@@ -14,8 +14,8 @@ RdfView.extend(SummaryRdfView);
 // Generates triples and quads by sending them to the data and/or metadata callbacks
 SummaryRdfView.prototype._generateRdf = function (settings, data, metadata, done) {
   // Add summary triples
-  settings.resultStream.on('data', data);
-  settings.resultStream.on('end',  done);
+  settings.results.on('data', data);
+  settings.results.on('end',  done);
 };
 
 module.exports = SummaryRdfView;
diff --git a/lib/views/triplepatternfragments/TriplePatternFragmentsHtmlView.js b/lib/views/triplepatternfragments/TriplePatternFragmentsHtmlView.js
@@ -14,10 +14,13 @@ HtmlView.extend(TriplePatternFragmentsHtmlView);
 // Renders the view with the given settings to the response
 TriplePatternFragmentsHtmlView.prototype._render = function (settings, request, response, done) {
   // Read the data and metadata
-  var self = this, triples = [], resultStream = settings.resultStream;
-  resultStream.on('data', function (triple) { triples.push(triple); });
-  resultStream.on('end',  function () { settings.triples = triples; settings.metadata && renderHtml(); });
-  resultStream.on('metadata', function (m) { settings.metadata = m; settings.triples  && renderHtml(); });
+  var self = this, triples = settings.triples = [], results = settings.results;
+  results.on('data', function (triple) { triples.push(triple); });
+  results.on('end',  function () { settings.metadata && renderHtml(); });
+  results.getProperty('metadata', function (metadata) {
+    settings.metadata = metadata;
+    results.ended && renderHtml();
+  });
 
   // Generates the HTML after the data and metadata have been retrieved
   function renderHtml() {
diff --git a/lib/views/triplepatternfragments/TriplePatternFragmentsRdfView.js b/lib/views/triplepatternfragments/TriplePatternFragmentsRdfView.js
@@ -19,8 +19,8 @@ RdfView.extend(TriplePatternFragmentsRdfView);
 
 // Generates triples and quads by sending them to the data and/or metadata callbacks
 TriplePatternFragmentsRdfView.prototype._generateRdf = function (settings, data, metadata, done) {
-  var metadataDone = false, triplesDone = false,
-      datasource = settings.datasource, fragment = settings.fragment, query = settings.query;
+  var datasource = settings.datasource, fragment = settings.fragment, query = settings.query,
+      results = settings.results, metadataDone = false;
 
   // Add data source metadata
   metadata(datasource.index, hydra + 'member', datasource.url);
@@ -46,7 +46,7 @@ TriplePatternFragmentsRdfView.prototype._generateRdf = function (settings, data,
   metadata('_:object',    hydra + 'property', rdf + 'object');
 
   // Add fragment metadata
-  settings.resultStream.on('metadata', function (meta) {
+  results.getProperty('metadata', function (meta) {
     // General fragment metadata
     metadata(fragment.url, voID + 'subset', fragment.pageUrl);
     metadata(fragment.pageUrl, rdf + 'type', hydra + 'PartialCollectionView');
@@ -70,18 +70,14 @@ TriplePatternFragmentsRdfView.prototype._generateRdf = function (settings, data,
     if (totalCount >= query.limit + (query.offset || 0))
       metadata(fragment.pageUrl, hydra + 'next', fragment.nextPageUrl);
 
-    // End if both the metadata and the data have been written
+    // End if the data was also written
     metadataDone = true;
-    triplesDone && done();
+    results.ended && done();
   });
 
-  // Add data triples
-  settings.resultStream.on('data', data);
-  settings.resultStream.on('end',  function ()  {
-    // End if both the metadata and the data have been written
-    triplesDone = true;
-    metadataDone && done();
-  });
+  // Add fragment data
+  results.on('data', data);
+  results.on('end', function () { metadataDone && done(); });
 };
 
 module.exports = TriplePatternFragmentsRdfView;
diff --git a/test/controllers/TriplePatternFragmentsController-test.js b/test/controllers/TriplePatternFragmentsController-test.js
@@ -140,7 +140,7 @@ describe('TriplePatternFragmentsController', function () {
             nextPageUrl:     'https://example.org/my-datasource?a=b&c=d&page=2',
             previousPageUrl: null,
           },
-          resultStream: {
+          results: {
             stream: 'items',
           },
           prefixes: prefixes,
diff --git a/test/datasources/CompositeDatasource-test.js b/test/datasources/CompositeDatasource-test.js
@@ -119,7 +119,7 @@ function itShouldExecute(getDatasource, name, query,
     var resultsCount = 0, totalCount, triples = [];
     before(function (done) {
       var result = getDatasource().select(query);
-      result.on('metadata', function (metadata) { totalCount = metadata.totalCount; });
+      result.getProperty('metadata', function (metadata) { totalCount = metadata.totalCount; });
       result.on('data', function (triple) { resultsCount++; expectedTriples && triples.push(triple); });
       result.on('end', done);
     });
diff --git a/test/datasources/HdtDatasource-test.js b/test/datasources/HdtDatasource-test.js
@@ -193,7 +193,7 @@ function itShouldExecute(getDatasource, name, query,
     var resultsCount = 0, totalCount, triples = [];
     before(function (done) {
       var result = getDatasource().select(query);
-      result.on('metadata', function (metadata) { totalCount = metadata.totalCount; });
+      result.getProperty('metadata', function (metadata) { totalCount = metadata.totalCount; });
       result.on('data', function (triple) { resultsCount++; expectedTriples && triples.push(triple); });
       result.on('end', done);
     });
diff --git a/test/datasources/JsonLdDatasource-test.js b/test/datasources/JsonLdDatasource-test.js
@@ -87,7 +87,7 @@ function itShouldExecute(datasource, name, query, expectedResultsCount, expected
     var resultsCount = 0, totalCount;
     before(function (done) {
       var result = datasource.select(query);
-      result.on('metadata', function (metadata) { totalCount = metadata.totalCount; });
+      result.getProperty('metadata', function (metadata) { totalCount = metadata.totalCount; });
       result.on('data', function (triple) { resultsCount++; });
       result.on('end', done);
     });
diff --git a/test/datasources/SparqlDatasource-test.js b/test/datasources/SparqlDatasource-test.js
@@ -152,7 +152,7 @@ describe('SparqlDatasource', function () {
         request.onThirdCall().returns(test.createHttpResponse(turtleResult, 'text/turtle'));
 
         result = datasource.select({ subject: 'abc', features: { triplePattern: true } });
-        result.on('metadata', function (metadata) { totalCount = metadata.totalCount; });
+        result.getProperty('metadata', function (metadata) { totalCount = metadata.totalCount; });
         firstArgsCopy = JSON.parse(JSON.stringify(request.firstCall.args[0]));
       });
 
@@ -271,7 +271,7 @@ function itShouldExecute(datasource, request, name, query, constructQuery, count
       request.onSecondCall().returns(test.createHttpResponse(countResult, 'text/csv'));
 
       result = datasource.select(query);
-      result.on('metadata', function (metadata) { totalCount = metadata.totalCount; });
+      result.getProperty('metadata', function (metadata) { totalCount = metadata.totalCount; });
     });
 
     it('should request a matching CONSTRUCT query', function () {
diff --git a/test/datasources/TurtleDatasource-test.js b/test/datasources/TurtleDatasource-test.js
@@ -87,7 +87,7 @@ function itShouldExecute(datasource, name, query, expectedResultsCount, expected
     var resultsCount = 0, totalCount;
     before(function (done) {
       var result = datasource.select(query);
-      result.on('metadata', function (metadata) { totalCount = metadata.totalCount; });
+      result.getProperty('metadata', function (metadata) { totalCount = metadata.totalCount; });
       result.on('data', function (triple) { resultsCount++; });
       result.on('end', done);
     });
diff --git a/test/test-setup.js b/test/test-setup.js
@@ -46,16 +46,6 @@ test.createStreamCapture = function () {
   return stream;
 };
 
-// Creates a readable stream from an array
-test.streamFromArray = function (items) {
-  var stream = new Readable({ objectMode: true });
-  stream._read = function () {
-    var item = items.shift();
-    this.push(item || null);
-  };
-  return stream;
-};
-
 chai.use(function (chai, utils) {
   // Checks whether the stream contains the given number of elements
   chai.Assertion.addMethod('streamWithLength', function (expectedLength, callback) {
diff --git a/test/views/triplepatternfragments/TriplePatternFragmentsRdfView-test.js b/test/views/triplepatternfragments/TriplePatternFragmentsRdfView-test.js

Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,7 @@ SummaryController.prototype._handleRequest = function (request, response, next)`
`42`	`42`
`43`	`43`	`// Render the summary`
`44`	`44`	`var view = this._negotiateView('Summary', request, response);`
`45`		`- view.render({ prefixes: this._prefixes, resultStream: streamParser }, request, response);`
	`45`	`+ view.render({ prefixes: this._prefixes, results: streamParser }, request, response);`
`46`	`46`	`}`
`47`	`47`	`else`
`48`	`48`	`next();`