Update to scramjet version supporting pipeline method, add error hand…

…ling in pipeline in adapters so that parsing and fetch errors are handled gracefully.

adapters/caaqm.js

@@ -34,19 +34,18 @@ export async function fetchStream (source) {
     body: Buffer.from('{"region":"landing_dashboard"}').toString('base64')
   });
 
-  return DataStream
-    .from(
-      () => {
-        const response = request(options);
-        const parser = JSONStream.parse('map.station_list.*');
-        const output = new DataStream();
-
-        parser.on('error', (e) => output.raise(e));
-        response.on('error', (e) => output.raise(e));
+  const requestObject = request(options);
 
-        return response.pipe(parser).pipe(output);
-      }
+  return DataStream
+    .pipeline(
+      requestObject,
+      JSONStream.parse('map.station_list.*')
     )
+    .catch(e => {
+      requestObject.abort();
+      e.stream.end();
+      throw e;
+    })
     .setOptions({maxParallel: 5})
     .into(
       (siteStream, site) => {

adapters/chinaaqidata.js

@@ -30,10 +30,17 @@ exports.fetchStream = function (source) {
     Key: 'airnow.json'
   };
 
-  return s3.getObject(params)
-    .createReadStream()
-    .pipe(JSONStream.parse('features.*'))
-    .pipe(new DataStream())
+  const s3stream = s3.getObject(params).createReadStream();
+  return DataStream
+    .pipeline(
+      s3stream,
+      JSONStream.parse('features.*')
+    )
+    .catch(e => {
+      s3stream.abort();
+      e.stream.end();
+      throw e;
+    })
     .use(stream => {
       stream.name = 'unused';
       return stream;

adapters/eea-direct.js

@@ -39,9 +39,17 @@ export async function fetchData (source, cb) {
 let _battuta = null;
 function getBattutaStream () {
   if (!_battuta) {
-    _battuta = request({url: stationsLink})
-      .pipe(JSONStream.parse('*'))
-      .pipe(new DataStream())
+    const requestObject = request({url: stationsLink});
+    _battuta = DataStream
+      .pipeline(
+        requestObject,
+        JSONStream.parse('*')
+      )
+      .catch(e => {
+        requestObject.abort();
+        e.stream.end();
+        throw e;
+      })
       .keep(Infinity);
   }
 

adapters/gios-poland.js

@@ -19,8 +19,17 @@ export function fetchStream (source) {
   } = source;
 
   const stationUrl = `${url}/station/findAll`;
+  const requestObject = request.get(stationUrl);
   return DataStream
-    .from(() => request.get(stationUrl).pipe(JSONStream('*')))
+    .pipeline(
+      requestObject,
+      JSONStream('*')
+    )
+    .catch(e => {
+      requestObject.abort();
+      e.stream.end();
+      throw e;
+    })
     .map(
       ({
         id,

package.json

@@ -37,7 +37,7 @@
     "request-promise-native": "^1.0.5",
     "require-dir": "^1.0.0",
     "s3-upload-stream": "^1.0.7",
-    "scramjet": "^4.18.17",
+    "scramjet": "^4.19.0",
     "ssl-root-cas": "^1.2.5",
     "transliteration": "^1.6.6",
     "tz-lookup": "^6.1.8",