beanstalkd - npm Package Compare versions

Comparing version 1.0.2 to 1.0.3

lib/client.js

@@ -95,2 +95,4 @@ 'use strict';
         connection.on('error', function (err) {
+          _this.closed = true;
+          _this.error = err;
           reject(err);
@@ -161,3 +163,3 @@ });
 
-    if (connection.closed) throw new Error('Connection is closed');
+    if (this.closed) throw new Error('Connection is closed');
 
@@ -164,0 +166,0 @@ return new _bluebird2['default'](function (resolve, reject) {

lib/read-queue.js

@@ -40,3 +40,5 @@ 'use strict';
         if (!this.current) {
-          throw new Error('No read queue item for item, length: ' + data.length);
+          this.connection.emit('error', new Error('No read queue item for item, length: ' + data.length));
+          this.connection.destroy();
+          return;
         }
@@ -43,0 +45,0 @@ }

package.json

 {
   "name": "beanstalkd",
-  "version": "1.0.2",
+  "version": "1.0.3",
   "description": "A beanstalkd client for Node.js with promises",
@@ -5,0 +5,0 @@ "main": "lib/client.js",

test/integration/client.test.js

@@ -144,2 +144,4 @@ import Client from 'client';
     it('should be able to reserve two large jobs in parallel', function () {
+      this.timeout(10000);
+
       let workerA = new Client(host, port)
@@ -150,3 +152,3 @@ , workerB = new Client(host, port)
 
-      for (let i = 0; i < 1750; i++) {
+      for (let i = 0; i < 10000; i++) {
         values[Math.random().toString()] = Math.random().toString();
@@ -188,3 +190,87 @@ }
 
+    it('should be able to put and reserve large jobs serially', function () {
+      this.timeout(10000);
+
+      let worker = new Client(host, port)
+        , tube = Math.random().toString()
+        , values = {};
+
+      for (let i = 0; i < 100000; i++) {
+        values[Math.random().toString()] = Math.random().toString();
+      }
+
+      return worker.connect().then(() => {
+        return Promise.join(
+          this.client.use(tube),
+          worker.watch(tube).then(function () {
+            return worker.ignore('default');
+          })
+        );
+      }).then(() => {
+        return this.client.put(0, 0, 180, JSON.stringify(values)).then(() => {
+          return worker.reserveWithTimeout(0).spread((reserveId, body) => {
+            expect(JSON.parse(body.toString())).to.deep.equal(values);
+          });
+        });
+      }).then(() => {
+        return this.client.put(0, 0, 180, JSON.stringify(values)).then(() => {
+          return worker.reserveWithTimeout(0).spread((reserveId, body) => {
+            expect(JSON.parse(body.toString())).to.deep.equal(values);
+          });
+        });
+      }).then(() => {
+        return this.client.put(0, 0, 180, JSON.stringify(values)).then(() => {
+          return worker.reserveWithTimeout(0).spread((reserveId, body) => {
+            expect(JSON.parse(body.toString())).to.deep.equal(values);
+          });
+        });
+      }).finally(function () {
+        worker.quit();
+      });
+    });
+
+    it('should be able to put and reserve large jobs serially (2)', function () {
+      this.timeout(10000);
+
+      let worker = new Client(host, port)
+        , tube = Math.random().toString()
+        , values = {};
+
+      for (let i = 0; i < 100000; i++) {
+        values[Math.random().toString()] = Math.random().toString();
+      }
+
+      return worker.connect().then(() => {
+        return Promise.join(
+          this.client.use(tube),
+          worker.watch(tube).then(function () {
+            return worker.ignore('default');
+          })
+        );
+      }).then(() => {
+        return Promise.join(
+          this.client.put(0, 0, 180, JSON.stringify(values)),
+          this.client.put(0, 0, 180, JSON.stringify(values)),
+          this.client.put(0, 0, 180, JSON.stringify(values))
+        );
+      }).then(() => {
+        return worker.reserveWithTimeout(0).spread((reserveId, body) => {
+          expect(JSON.parse(body.toString())).to.deep.equal(values);
+        });
+      }).then(() => {
+        return worker.reserveWithTimeout(0).spread((reserveId, body) => {
+          expect(JSON.parse(body.toString())).to.deep.equal(values);
+        });
+      }).then(() => {
+        return worker.reserveWithTimeout(0).spread((reserveId, body) => {
+          expect(JSON.parse(body.toString())).to.deep.equal(values);
+        });
+      }).finally(function () {
+        worker.quit();
+      });
+    });
+
     it('should be able to reserve two large jobs on different tubes in parallel', function () {
+      this.timeout(10000);
+
       let workerA = new Client(host, port)
@@ -196,3 +282,3 @@ , workerB = new Client(host, port)
 
-      for (let i = 0; i < 1750; i++) {
+      for (let i = 0; i < 10000; i++) {
         values[Math.random().toString()] = Math.random().toString();
@@ -199,0 +285,0 @@ }

test/unit/client.test.js

@@ -48,3 +48,3 @@ import BeanstalkdClient from 'client';
 
-    it('should resolve on error event', function () {
+    it('should reject on error event', function () {
       var promise = this.client.connect()
@@ -128,2 +128,17 @@ , callback = this.connectionStub.on.withArgs('error').getCall(0).args[1];
   });
+
+  describe('readQueue', function () {
+    it('should close connection if read queue errors', function () {
+      let client = client = new BeanstalkdClient(Math.random().toString(), Math.floor(Math.random() * 9999));
+      client.connect();
+      this.connectionStub.emit('connect');
+
+      expect(() => {
+        this.connectionStub.emit('data', new Buffer(Math.random().toString()));
+      }).to.throw();
+
+      expect(client.closed).to.equal(true);
+      expect(client.error).to.be.ok;
+    });
+  });
 });

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

54045

increased by7.04%

Lines of code

1105

increased by9.08%

No dependency changes