node-webhdfs - npm Package Compare versions

Comparing version 0.1.0 to 0.2.0

package.json

 {
   "name": "node-webhdfs",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "description": "A WebHDFS module for Node.js.",
@@ -24,3 +24,3 @@ "author": "Ryan Cole <[e-mail]> (http://rycole.com)",
     "request": "^2.45.0",
-    "underscore": "^1.7.0"
+    "lodash": "^4.17.4"
   },
@@ -27,0 +27,0 @@ "devDependencies": {

src/webhdfs.js

 var querystring = require('querystring'),
     request = require('request'),
-    _ = require('underscore'),
+    _ = require('lodash'),
     RemoteException = exports.RemoteException = require('./remoteexception.js');
@@ -15,6 +15,9 @@
         namenode_host: 'localhost',
-        path_prefix: '/webhdfs/v1'
-        
+        path_prefix: '/webhdfs/v1',
+        high_availability: false
     });
-    
+
+    if (Array.isArray(this.options.namenode_list) && this.options.namenode_list.length > 1) {
+        this.options.high_availability = true
+    }
     // save formatted base api url
@@ -25,3 +28,41 @@ this.base_url = 'http://' + this.options.namenode_host + ':' + this.options.namenode_port + this.options.path_prefix;
 
+WebHDFSClient.prototype._makeBaseUrl = function () {
+    this.base_url = 'http://' + this.options.namenode_host + ':' + this.options.namenode_port + this.options.path_prefix;
+};
 
+WebHDFSClient.prototype._changeNameNodeHost = function () {
+    var host = this.options.namenode_host;
+    var list = this.options.namenode_list;
+    var index = list.indexOf(host) + 1;
+    //if empty start from the beginning of the
+    this.options.namenode_host = list[index] ? list[index] : list[0];
+    this._makeBaseUrl();
+};
+
+function _parseResponse(self, fnName, args, bodyArgs, callback, justCheckErrors){
+    // forward request error
+    return function(error, response, body) {
+        if (error) return callback(error);
+
+        // exception handling
+        if (typeof body === 'object' && 'RemoteException' in body) {
+            if(self.options.high_availability && body.RemoteException.exception === 'StandbyException'){
+                //change client
+                self._changeNameNodeHost();
+                return self[fnName].apply(self, args)
+            }
+            else {
+                return callback(new RemoteException(body));
+            }
+        }
+
+        if (justCheckErrors) {
+            return
+        }
+        // execute callback
+        return callback(null, _.get(body, bodyArgs, body));
+    }
+}
+
+
 // ref: http://hadoop.apache.org/common/docs/r1.0.2/webhdfs.html#DELETE
@@ -32,6 +73,4 @@ WebHDFSClient.prototype.del = function (path, hdfsoptions, requestoptions, callback) {
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -41,8 +80,10 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
-
     }
     
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'del', originalArgs, 'boolean', callback);
+    
     // format request args
@@ -59,16 +100,3 @@ var args = _.defaults({
     // send http request
-    request.del(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // exception handling
-        if ('RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-        
-        // execute callback
-        return callback(null, body.boolean);
-        
-    });
+    request.del(args, parseResponse);
     
@@ -83,3 +111,2 @@ };
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
@@ -91,8 +118,10 @@ requestoptions = undefined;
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
+    }
 
-    }
-    
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'listStatus', originalArgs, 'FileStatuses.FileStatus', callback);
+
     // format request args
@@ -108,18 +137,5 @@ var args = _.defaults({
     // send http request
-    request.get(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // exception handling
-        if ('RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-
-        // execute callback
-        return callback(null, body.FileStatuses.FileStatus)
-        
-    })
+    request.get(args, parseResponse)
     
-}
+};
 
@@ -132,6 +148,4 @@
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -141,8 +155,10 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
-
     }
     
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'getFileStatus', originalArgs, 'FileStatus', callback);
+
     // format request args
@@ -158,16 +174,3 @@ var args = _.defaults({
     // send http request
-    request.get(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // exception handling
-        if ('RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-        
-        // execute callback
-        return callback(null, body.FileStatus);
-        
-    });
+    request.get(args, parseResponse);
     
@@ -182,3 +185,2 @@ };
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
@@ -190,8 +192,10 @@ requestoptions = undefined;
     if (callback===undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
-
     }
     
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'getContentSummary', originalArgs, 'ContentSummary', callback);
+
     // format request args
@@ -207,16 +211,3 @@ var args = _.defaults({
     // send http request
-    request.get(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // exception handling
-        if ('RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-        
-        // execute callback
-        return callback(null, body.ContentSummary);
-        
-    });
+    request.get(args, parseResponse);
     
@@ -231,6 +222,4 @@ };
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -240,7 +229,9 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
+    }
 
-    }
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'getFileChecksum', originalArgs, 'FileChecksum', callback);
     
@@ -257,16 +248,3 @@ // format request args
     // send http request
-    request.get(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // exception handling
-        if ('RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-        
-        // execute callback
-        return callback(null, body.FileChecksum);
-        
-    });
+    request.get(args, parseResponse);
 };
@@ -280,6 +258,4 @@
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -289,8 +265,10 @@
     if (callback===undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
-
     }
     
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'getHomeDirectory', originalArgs, 'Path', callback);
+    
     // format request args
@@ -307,15 +285,6 @@ var args = _.defaults({
     // send http request
-    request.get(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // execute callback
-        return callback(null, body.Path);
-        
-    });
+    request.get(args, parseResponse);
     
 };
 
-
 // ref: http://hadoop.apache.org/common/docs/r1.0.2/webhdfs.html#OPEN
@@ -326,6 +295,4 @@ WebHDFSClient.prototype.open = function (path, hdfsoptions, requestoptions, callback) {
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -335,8 +302,9 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
-
     }
 
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'open', originalArgs, null, callback);
     // format request args
@@ -352,14 +320,3 @@ var args = _.defaults({
     // send http request
-    return request.get(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-
-        if (typeof body === 'object' && 'RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-        // execute callback
-        return callback(null, body);
-        
-    });
+    return request.get(args, parseResponse);
     
@@ -374,6 +331,4 @@ };
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -383,7 +338,9 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
+    }
 
-    }
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'rename', originalArgs, 'boolean', callback);
     
@@ -402,16 +359,3 @@ // format request args
     // send http request
-    request.put(args, function (error, res, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // exception handling
-        if ('RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-        
-        // execute callback
-        return callback(null, body.boolean);
-        
-    });
+    request.put(args, parseResponse);
     
@@ -423,9 +367,7 @@ };
 WebHDFSClient.prototype.mkdirs = function (path, hdfsoptions, requestoptions, callback) {
-    
+
     // requestoptions may be omitted
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -435,7 +377,9 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
+    }
 
-    }
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'mkdirs', originalArgs, 'boolean', callback);
     
@@ -451,18 +395,5 @@ // generate query string
     }, requestoptions || {});
-    
+
     // send http request
-    request.put(args, function (error, response, body) {
-        
-        // forward request error
-        if (error) return callback(error);
-        
-        // exception handling
-        if ('RemoteException' in body){
-            return callback(new RemoteException(body));
-        }
-        
-        // execute callback
-        return callback(null, body.boolean);
-        
-    });
+    request.put(args, parseResponse);
     
@@ -477,6 +408,4 @@ };
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -486,7 +415,9 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
+    }
 
-    }
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'append', originalArgs, null, callback, true);
     
@@ -509,4 +440,3 @@ // format request args
         
-        // forward request error
-        if (error) return callback(error);
+       parseResponse(error, response, body);
         
@@ -518,6 +448,4 @@ // check for expected redirect
             args = _.defaults({
-                
                 body: data,
                 uri: response.headers.location
-                
             }, requestoptions || {});
@@ -529,25 +457,16 @@
                 // forward request error
-                if (error) return callback(error);
+                parseResponse(error, response, body);
         
                 // check for expected response
                 if (response.statusCode == 200) {
-                    
                     return callback(null, true);
-                    
                 } else {
-                    
                     return callback(new Error('expected http 200: ' + response.body));
-                    
                 }
-                
             });
             
         } else {
-            
             return callback(new Error('expected redirect'));
-            
         }
-        
-    }.bind(this));
-    
+    });
 };
@@ -561,6 +480,4 @@
     if (callback === undefined && typeof(requestoptions) === 'function') {
-
         callback = requestoptions;
         requestoptions = undefined;
-
     }
@@ -570,7 +487,9 @@
     if (callback === undefined && typeof(hdfsoptions) === 'function') {
-
         callback = hdfsoptions;
         hdfsoptions = undefined;
+    }
 
-    }
+    var self = this;
+    var originalArgs = [path, hdfsoptions, requestoptions, callback];
+    var parseResponse = _parseResponse(self, 'create', originalArgs, null, callback, true);
     
@@ -595,3 +514,3 @@ // generate query string
         // forward request error
-        if (error) return callback(error);
+        parseResponse(error, response, body);
         
@@ -611,26 +530,19 @@ // check for expected redirect
                 // forward request error
-                if (error) return callback(error);
+                parseResponse(error, response, body);
         
                 // check for expected created response
                 if (response.statusCode == 201) {
-                    
                     // execute callback
                     return callback(null, response.headers.location);
-                    
                 } else {
-                    
                     return callback(new Error('expected http 201 created'));
-                    
                 }
-                
             });
             
         } else {
-            
             return callback(new Error('expected redirect'));
-            
         }
         
-    }.bind(this));
+    });
     
 };

test/webhdfs.js

@@ -7,4 +7,41 @@
     
-    var client = new (require('..')).WebHDFSClient({ user: 'ryan' });
+    var client = new (require('..')).WebHDFSClient({
+        "user": "hdfs",
+        "namenode_port": 50070,
+        "path_prefix": "/webhdfs/v1",
+        namenode_host: "c1b5s2.tera4.terascope.io",
+        "namenode_list": ["c1b5s2.tera4.terascope.io", "c1b5s1.tera4.terascope.io"]
+    });
+
+    var client2 = new (require('..')).WebHDFSClient({
+        namenode_host: "endpoint1"
+    });
+
+    var client3 = new (require('..')).WebHDFSClient({
+        namenode_host: "endpoint1",
+        namenode_list: ["endpoint1", "endpoint2"]
+    });
+
+    describe('change endpoint', function () {
+
+        it('should set high_availability to false if a list is not provided', function (done) {
+            client2.should.have.property('base_url', 'http://endpoint1:50070/webhdfs/v1');
+            client2.options.should.have.property('high_availability', false);
+
+            return done()
+        });
+
+        it('should change endpoint if a list is provided', function (done) {
+            client3.should.have.property('base_url', 'http://endpoint1:50070/webhdfs/v1');
+            client3.options.should.have.property('high_availability', true);
+
+            client3._changeNameNodeHost();
+            client3.should.have.property('base_url', 'http://endpoint2:50070/webhdfs/v1');
+
+            return done()
+        });
+
+    });
     
+    
     describe('#mkdirs', function () {
@@ -131,3 +168,3 @@
                 
-                JSON.parse(data).should.have.property('foo', 'bar');
+              (data).should.have.property('foo', 'bar');
                 
@@ -134,0 +171,0 @@ return done();

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

24119

increased by6.06%

Lines of code

519

increased by4.01%

Dependency changes

• + Addedlodash@^4.17.4

• + Addedlodash@4.17.21(transitive)

• - Removedunderscore@^1.7.0

• - Removedunderscore@1.13.7(transitive)