Merge branch 'feature-datasource-asynciterator' into develop

* feature-datasource-asynciterator:
  Pass datasource metadata as a property.
  Switch datasource to AsyncIterator.

Author: RubenVerborgh

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);
 

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();

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;

lib/datasources/CompositeDatasource.js

@@ -59,19 +59,16 @@ CompositeDatasource.prototype._getExactCount = function (datasource, query, call
   // Otherwise, count all the triples manually
   var emptyQuery = { offset: 0, subject: query.subject, predicate: query.predicate, object: query.object };
   var exactCount = 0;
-  var countingTripleStream = {
-    push: function (triple) {
-      if (triple)
-        exactCount++;
-      else {
-        // Cache large values; small ones are calculated fast anyway
-        if (exactCount > 1000)
-          cache.set(cacheKey, exactCount);
-        callback(exactCount);
-      }
+  var triplesCounter = {
+    _push: function (triple) { exactCount++; },
+    close: function () {
+      // Cache large values; small ones are calculated fast anyway
+      if (exactCount > 1000)
+        cache.set(cacheKey, exactCount);
+      callback(exactCount);
     },
   };
-  datasource._executeQuery(emptyQuery, countingTripleStream, noop);
+  datasource._executeQuery(emptyQuery, triplesCounter, noop);
 };
 
 // Recursively find all required datasource composition info to perform a query.
@@ -92,61 +89,65 @@ CompositeDatasource.prototype._getDatasourceInfo = function (query, absoluteOffs
       // We checked all datasources, return our accumulated information
       callback(chosenDatasource, chosenOffset, totalCount, hasExactCount);
     else {
-      var emptyTripleStream = { push: noop };
       var datasource = self._getDatasourceById(datasourceIndex);
-      datasource._executeQuery(emptyQuery, emptyTripleStream, 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, tripleStream, 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 });
-      tripleStream.push(null);
+      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.
-      tripleStream.push = makeSmartPush(tripleStream, function (localEmittedCount) {
+      var emitted = 0;
+      countItems(destination, function (localEmittedCount) {
         // This is called after the last element has been pushed
 
         // If we haven't reached our limit, try to fill it with other datasource query results.
@@ -156,36 +157,31 @@ CompositeDatasource.prototype._executeQuery = function (query, tripleStream, met
           var localLimit = limit - emitted;
           var subQuery = { offset: 0, limit: localLimit,
             subject: query.subject, predicate: query.predicate, object: query.object };
-          self._getDatasourceById(datasourceIndex)._executeQuery(subQuery, tripleStream, noop);
-          return true;
+          self._getDatasourceById(datasourceIndex)._executeQuery(subQuery, destination, noop);
+          return false;
         }
         else
-          return false;
+          return true;
       });
 
       // Initiate query to the first datasource.
       var subQuery = { offset: relativeOffset, limit: limit,
         subject: query.subject, predicate: query.predicate, object: query.object };
-      self._getDatasourceById(datasourceIndex)._executeQuery(subQuery, tripleStream, noop);
+      self._getDatasourceById(datasourceIndex)._executeQuery(subQuery, destination, noop);
     }
   });
 
-  // Replaces a tripleStream.push
-  // It takes the tripleStream as first argument and a callback as second argument.
-  // The callback will be called when the push function is called with a falsy value.
-  // Returning a falsy value inside the callback will delegate the falsy value to the original
-  // push function anyways.
-  function makeSmartPush(self, nullCb) {
-    var count = 0, originalPush = self.push;
-    return function (element) {
-      if (!element) {
-        if (!nullCb(count))
-          originalPush.call(self, element);
-      }
-      else {
-        count++;
-        originalPush.call(self, element);
-      }
+  // Counts the number of triples and sends them through the callback,
+  // only closing the iterator when the callback returns true.
+  function countItems(destination, closeCallback) {
+    var count = 0, originalPush = destination._push, originalClose = destination.close;
+    destination._push = function (element) {
+      count++;
+      originalPush.call(destination, element);
+    };
+    destination.close = function () {
+      if (closeCallback(count))
+        originalClose.call(destination);
     };
   }
 };

lib/datasources/Datasource.js

@@ -3,7 +3,7 @@
 
 var fs = require('fs'),
     _ = require('lodash'),
-    Readable = require('stream').Readable,
+    BufferedIterator = require('asynciterator').BufferedIterator,
     EventEmitter = require('events');
 
 // Creates a new Datasource
@@ -105,32 +105,24 @@ Datasource.prototype.select = function (query, onError) {
   if (query.object  && query.object.indexOf(blankNodePrefix) === 0)
     (query = _.clone(query)).object  = '_:' + query.object.substr(blankNodePrefixLength);
 
-  // Create the triple stream
-  var tripleStream = new Readable({ objectMode: true });
-  tripleStream._read = noop;
-  tripleStream._push = tripleStream.push;
-  tripleStream.push = function (triple) {
-    // Translate blank nodes to IRIs
-    if (triple) {
-      if (triple.subject[0] === '_') triple.subject = blankNodePrefix + triple.subject.substr(2);
-      if (triple.object[0]  === '_') triple.object  = blankNodePrefix + triple.object.substr(2);
-    }
-    this._push(triple);
-  };
-  onError && tripleStream.on('error', onError);
+  // Translate blank nodes in the result to blank node IRIs
+  var destination = new BufferedIterator(), outputTriples;
+  outputTriples = destination.map(function (triple) {
+    if (triple.subject[0] === '_') triple.subject = blankNodePrefix + triple.subject.substr(2);
+    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, tripleStream, function (metadata) {
-      setImmediate(function () { tripleStream.emit('metadata', metadata); });
-    });
-  }
-  catch (error) { tripleStream.emit('error', error); }
-  return tripleStream;
+  try { this._executeQuery(query, destination); }
+  catch (error) { outputTriples.emit('error', error); }
+  return outputTriples;
 };
 
-// Writes the results of the query to the given triple stream
-Datasource.prototype._executeQuery = function (query, tripleStream, metadataCallback) {
+// Writes the results of the query to the given destination
+Datasource.prototype._executeQuery = function (query, destination) {
   throw new Error('_executeQuery has not been implemented');
 };
 
@@ -165,7 +157,4 @@ Datasource.prototype.close = function (callback) {
   callback && callback();
 };
 
-// The empty function
-function noop() {}
-
 module.exports = Datasource;

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, tripleStream, 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') + ' ' +
@@ -45,15 +45,15 @@ ExternalHdtDatasource.prototype._executeQuery = function (query, tripleStream, m
   var parser = new N3Parser(), tripleCount = 0, estimatedTotalCount = 0, hasExactCount = true;
   parser.parse(hdt.stdout, function (error, triple) {
     if (error)
-      tripleStream.emit('error', new Error('Invalid query result: ' + error.message));
+      destination.emit('error', new Error('Invalid query result: ' + error.message));
     else if (triple)
-      tripleCount++, tripleStream.push(triple);
+      tripleCount++, destination._push(triple);
     else {
       // 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 });
-      tripleStream.push(null);
+      destination.setProperty('metadata', { totalCount: estimatedTotalCount, hasExactCount: hasExactCount });
+      destination.close();
     }
   });
   parser._prefixes._ = '_:'; // Ensure blank nodes are named consistently
@@ -66,7 +66,7 @@ ExternalHdtDatasource.prototype._executeQuery = function (query, tripleStream, m
 
   // Report query errors
   hdt.on('exit', function (exitCode) {
-    exitCode && tripleStream.emit('error', new Error('Could not query ' + hdtFile));
+    exitCode && destination.emit('error', new Error('Could not query ' + hdtFile));
   });
 };
 

lib/datasources/HdtDatasource.js

@@ -28,20 +28,20 @@ HdtDatasource.prototype._initialize = function (done) {
 };
 
 // Writes the results of the query to the given triple stream
-HdtDatasource.prototype._executeQuery = function (query, tripleStream, 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) {
-      if (error) return tripleStream.emit('error', error);
+      if (error) return destination.emit('error', error);
       // Ensure the estimated total count is as least as large as the number of triples
       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 });
-      // Add the triples to the stream
+      destination.setProperty('metadata', { totalCount: estimatedTotalCount, hasExactCount: hasExactCount });
+      // Add the triples to the output
       for (var i = 0; i < tripleCount; i++)
-        tripleStream.push(triples[i]);
-      tripleStream.push(null);
+        destination._push(triples[i]);
+      destination.close();
     });
 };
 

lib/datasources/MemoryDatasource.js

@@ -24,15 +24,15 @@ MemoryDatasource.prototype._getAllTriples = function (addTriple, done) {
 };
 
 // Writes the results of the query to the given triple stream
-MemoryDatasource.prototype._executeQuery = function (query, tripleStream, 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++)
-    tripleStream.push(triples[i]);
-  tripleStream.push(null);
+    destination._push(triples[i]);
+  destination.close();
 };
 
 module.exports = MemoryDatasource;