Always attach an error listener to fetched resources.

A datasource loaded with an invalid source
now emits an error rather than crashing the server.

Author: RubenVerborgh

lib/datasources/Datasource.js

@@ -128,7 +128,8 @@ Datasource.prototype._executeQuery = function (query, destination) {
 
 // Retrieves a stream through HTTP or the local file system
 Datasource.prototype._fetch = function (options) {
-  var stream, url = options.url, protocolMatch = /^(?:([a-z]+):\/\/)?/.exec(url);
+  var self = this, stream,
+      url = options.url, protocolMatch = /^(?:([a-z]+):)?/.exec(url);
   switch (protocolMatch[1] || 'file') {
   // Fetch a representation through HTTP(S)
   case 'http':
@@ -147,8 +148,18 @@ Datasource.prototype._fetch = function (options) {
     stream = fs.createReadStream(url.substr(protocolMatch[0].length), { encoding: 'utf8' });
     break;
   default:
-    throw new Error('Unknown protocol: ' + url);
+    stream = new EventEmitter();
+    setImmediate(function () {
+      stream.emit('error', new Error('Unknown protocol: ' + protocolMatch[1]));
+    });
   }
+
+  // If the stream has no other error handlers attached (besides this one),
+  // emit the stream error as a datasource error
+  stream.on('error', function (error) {
+    if (stream.listenerCount('error') === 1)
+      self.emit('error', error);
+  });
   return stream;
 };
 

test/datasources/Datasource-test.js

@@ -1,7 +1,11 @@
 /*! @license MIT ©2013-2016 Ruben Verborgh, Ghent University - imec */
 var Datasource = require('../../lib/datasources/Datasource');
 
-var EventEmitter = require('events');
+var EventEmitter = require('events'),
+    fs = require('fs'),
+    path = require('path');
+
+var exampleFile = path.join(__dirname, '../assets/test.ttl');
 
 describe('Datasource', function () {
   describe('The Datasource module', function () {
@@ -54,6 +58,56 @@ describe('Datasource', function () {
       .should.throw('_executeQuery has not been implemented');
     });
 
+    describe('fetching a resource', function () {
+      it('fetches an existing resource', function (done) {
+        var result = datasource._fetch({ url: 'file://' + exampleFile }), buffer = '';
+        result.on('data', function (d) { buffer += d; });
+        result.on('end', function () {
+          buffer.should.equal(fs.readFileSync(exampleFile, 'utf8'));
+          done();
+        });
+        result.on('error', done);
+      });
+
+      it('assumes file:// as the default protocol', function (done) {
+        var result = datasource._fetch({ url: exampleFile }), buffer = '';
+        result.on('data', function (d) { buffer += d; });
+        result.on('end', function () {
+          buffer.should.equal(fs.readFileSync(exampleFile, 'utf8'));
+          done();
+        });
+        result.on('error', done);
+      });
+
+      it('emits an error when the protocol is unknown', function (done) {
+        var result = datasource._fetch({ url: 'myprotocol:abc' });
+        result.on('error', function (error) {
+          error.message.should.contain('Unknown protocol: myprotocol');
+          done();
+        });
+      });
+
+      it('emits an error on the datasource when no error listener is attached to the result', function (done) {
+        var result = datasource._fetch({ url: exampleFile + 'notfound' });
+        result.on('data', done);
+        datasource.on('error', function (error) {
+          error.message.should.contain('ENOENT: no such file or directory');
+          done();
+        });
+      });
+
+      it('does not emit an error on the datasource when an error listener is attached to the result', function (done) {
+        var result = datasource._fetch({ url: exampleFile + 'notfound' });
+        result.on('error', function (error) {
+          error.message.should.contain('ENOENT: no such file or directory');
+          done();
+        });
+        datasource.on('error', function (error) {
+          done(error);
+        });
+      });
+    });
+
     describe('when closed without a callback', function () {
       it('should do nothing', function () {
         datasource.close();