beanstalkd - npm Package Compare versions

Comparing version 1.0.0 to 1.0.1

lib/client.js

@@ -45,2 +45,40 @@ 'use strict';
     this.closed = false;
+
+    this.use = makeCommand(new _writer.BasicWriter('use', new _types.TubeType()), new _reader.BasicReader('USING', new _types.TubeType()));
+
+    this.listTubeUsed = makeCommand(new _writer.BasicWriter('list-tube-used'), new _reader.BasicReader('USING', new _types.TubeType()));
+
+    this.pauseTube = makeCommand(new _writer.BasicWriter('pause-tube', new _types.TubeType(), new _types.DelayType()), new _reader.BasicReader('PAUSED'));
+
+    this.put = makeCommand(new _writer.BodyWriter('put', new _types.PriorityType(), new _types.DelayType(), new _types.Type('ttr', Number), new _types.BodyType()), new _reader.BasicReader('INSERTED', new _types.IdType()));
+
+    this.watch = makeCommand(new _writer.BasicWriter('watch', new _types.TubeType()), new _reader.BasicReader('WATCHING', new _types.TubeType()));
+    this.ignore = makeCommand(new _writer.BasicWriter('ignore', new _types.TubeType()), new _reader.BasicReader('WATCHING', new _types.TubeType()));
+
+    /* Reserve commands */
+    this.reserve = makeCommand(new _writer.BasicWriter('reserve'), new _reader.BodyReader('RESERVED', new _types.IdType(), new _types.BodyType()));
+    this.reserveWithTimeout = makeCommand(new _writer.BasicWriter('reserve-with-timeout', new _types.Type('timeout', Number)), new _reader.BodyReader('RESERVED', new _types.IdType(), new _types.BodyType()));
+
+    /* Job commands */
+    this.destroy = makeCommand(new _writer.BasicWriter('delete', new _types.IdType()), new _reader.BasicReader('DELETED'));
+    this.bury = makeCommand(new _writer.BasicWriter('bury', new _types.IdType(), new _types.PriorityType()), new _reader.BasicReader('BURIED'));
+    this.release = makeCommand(new _writer.BasicWriter('release', new _types.IdType(), new _types.PriorityType(), new _types.DelayType()), new _reader.BasicReader('RELEASED'));
+    this.touch = makeCommand(new _writer.BasicWriter('touch', new _types.IdType()), new _reader.BasicReader('TOUCHED'));
+    this.kickJob = makeCommand(new _writer.BasicWriter('kick-job', new _types.IdType()), new _reader.BasicReader('KICKED'));
+
+    /* Peek commands */
+    this.peek = makeCommand(new _writer.BasicWriter('peek', new _types.IdType()), new _reader.BodyReader('FOUND', new _types.IgnoreType(), new _types.BodyType()));
+
+    this.peekReady = makeCommand(new _writer.BasicWriter('peek-ready'), new _reader.BodyReader('FOUND', new _types.IdType(), new _types.BodyType()));
+
+    this.peekDelayed = makeCommand(new _writer.BasicWriter('peek-delayed'), new _reader.BodyReader('FOUND', new _types.IdType(), new _types.BodyType()));
+
+    this.peekBuried = makeCommand(new _writer.BasicWriter('peek-buried'), new _reader.BodyReader('FOUND', new _types.IdType(), new _types.BodyType()));
+
+    /* Commands that returns YAML */
+    this.listTubesWatched = makeCommand(new _writer.BasicWriter('list-tubes-watched'), new _reader.YamlReader('OK', new _types.YamlBodyType()));
+    this.listTubes = makeCommand(new _writer.BasicWriter('list-tubes'), new _reader.YamlReader('OK', new _types.YamlBodyType()));
+    this.statsJob = makeCommand(new _writer.BasicWriter('stats-job', new _types.IdType()), new _reader.YamlReader('OK', new _types.YamlBodyType()));
+    this.statsTube = makeCommand(new _writer.BasicWriter('stats-tube', new _types.TubeType()), new _reader.YamlReader('OK', new _types.YamlBodyType()));
+    this.stats = makeCommand(new _writer.BasicWriter('stats'), new _reader.YamlReader('OK', new _types.YamlBodyType()));
   }
@@ -154,40 +192,2 @@
 }
-
-BeanstalkdClient.prototype.use = makeCommand(new _writer.BasicWriter('use', new _types.TubeType()), new _reader.BasicReader('USING', new _types.TubeType()));
-
-BeanstalkdClient.prototype.listTubeUsed = makeCommand(new _writer.BasicWriter('list-tube-used'), new _reader.BasicReader('USING', new _types.TubeType()));
-
-BeanstalkdClient.prototype.pauseTube = makeCommand(new _writer.BasicWriter('pause-tube', new _types.TubeType(), new _types.DelayType()), new _reader.BasicReader('PAUSED'));
-
-BeanstalkdClient.prototype.put = makeCommand(new _writer.BodyWriter('put', new _types.PriorityType(), new _types.DelayType(), new _types.Type('ttr', Number), new _types.BodyType()), new _reader.BasicReader('INSERTED', new _types.IdType()));
-
-BeanstalkdClient.prototype.watch = makeCommand(new _writer.BasicWriter('watch', new _types.TubeType()), new _reader.BasicReader('WATCHING', new _types.TubeType()));
-BeanstalkdClient.prototype.ignore = makeCommand(new _writer.BasicWriter('ignore', new _types.TubeType()), new _reader.BasicReader('WATCHING', new _types.TubeType()));
-
-/* Reserve commands */
-BeanstalkdClient.prototype.reserve = makeCommand(new _writer.BasicWriter('reserve'), new _reader.BodyReader('RESERVED', new _types.IdType(), new _types.BodyType()));
-BeanstalkdClient.prototype.reserveWithTimeout = makeCommand(new _writer.BasicWriter('reserve-with-timeout', new _types.Type('timeout', Number)), new _reader.BodyReader('RESERVED', new _types.IdType(), new _types.BodyType()));
-
-/* Job commands */
-BeanstalkdClient.prototype.destroy = makeCommand(new _writer.BasicWriter('delete', new _types.IdType()), new _reader.BasicReader('DELETED'));
-BeanstalkdClient.prototype.bury = makeCommand(new _writer.BasicWriter('bury', new _types.IdType(), new _types.PriorityType()), new _reader.BasicReader('BURIED'));
-BeanstalkdClient.prototype.release = makeCommand(new _writer.BasicWriter('release', new _types.IdType(), new _types.PriorityType(), new _types.DelayType()), new _reader.BasicReader('RELEASED'));
-BeanstalkdClient.prototype.touch = makeCommand(new _writer.BasicWriter('touch', new _types.IdType()), new _reader.BasicReader('TOUCHED'));
-BeanstalkdClient.prototype.kickJob = makeCommand(new _writer.BasicWriter('kick-job', new _types.IdType()), new _reader.BasicReader('KICKED'));
-
-/* Peek commands */
-BeanstalkdClient.prototype.peek = makeCommand(new _writer.BasicWriter('peek', new _types.IdType()), new _reader.BodyReader('FOUND', new _types.IgnoreType(), new _types.BodyType()));
-
-BeanstalkdClient.prototype.peekReady = makeCommand(new _writer.BasicWriter('peek-ready'), new _reader.BodyReader('FOUND', new _types.IdType(), new _types.BodyType()));
-
-BeanstalkdClient.prototype.peekDelayed = makeCommand(new _writer.BasicWriter('peek-delayed'), new _reader.BodyReader('FOUND', new _types.IdType(), new _types.BodyType()));
-
-BeanstalkdClient.prototype.peekBuried = makeCommand(new _writer.BasicWriter('peek-buried'), new _reader.BodyReader('FOUND', new _types.IdType(), new _types.BodyType()));
-
-/* Commands that returns YAML */
-BeanstalkdClient.prototype.listTubesWatched = makeCommand(new _writer.BasicWriter('list-tubes-watched'), new _reader.YamlReader('OK', new _types.YamlBodyType()));
-BeanstalkdClient.prototype.listTubes = makeCommand(new _writer.BasicWriter('list-tubes'), new _reader.YamlReader('OK', new _types.YamlBodyType()));
-BeanstalkdClient.prototype.statsJob = makeCommand(new _writer.BasicWriter('stats-job', new _types.IdType()), new _reader.YamlReader('OK', new _types.YamlBodyType()));
-BeanstalkdClient.prototype.statsTube = makeCommand(new _writer.BasicWriter('stats-tube', new _types.TubeType()), new _reader.YamlReader('OK', new _types.YamlBodyType()));
-BeanstalkdClient.prototype.stats = makeCommand(new _writer.BasicWriter('stats'), new _reader.YamlReader('OK', new _types.YamlBodyType()));
 module.exports = exports['default'];

lib/read-queue.js

@@ -38,2 +38,6 @@ 'use strict';
         this.current = this.queue.shift();
+
+        if (!this.current) {
+          throw new Error('No read queue item for item, length: ' + result.length);
+        }
       }
@@ -40,0 +44,0 @@

package.json

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

test/integration/client.test.js

@@ -142,3 +142,91 @@ import Client from 'client';
     });
+
+    it('should be able to reserve two large jobs in parallel', function () {
+      let workerA = new Client(host, port)
+        , workerB = new Client(host, port)
+        , tube = Math.random().toString()
+        , values = {};
+
+      for (let i = 0; i < 1750; i++) {
+        values[Math.random().toString()] = Math.random().toString();
+      }
+
+      return Promise.join(
+        workerA.connect(),
+        workerB.connect()
+      ).then(() => {
+        return Promise.join(
+          this.client.use(tube),
+          workerA.watch(tube).then(function () {
+            return workerA.ignore('default');
+          }),
+          workerB.watch(tube).then(function () {
+            return workerB.ignore('default');
+          })
+        );
+      }).then(() => {
+        return Promise.join(
+          this.client.put(0, 0, 180, JSON.stringify(values)),
+          this.client.put(0, 0, 180, JSON.stringify(values))
+        ).then(() => {
+          return Promise.join(
+            workerA.reserveWithTimeout(0).spread((reserveId, body) => {
+              expect(JSON.parse(body.toString())).to.deep.equal(values);
+            }),
+            workerB.reserveWithTimeout(0).spread((reserveId, body) => {
+              expect(JSON.parse(body.toString())).to.deep.equal(values);
+            })
+          );
+        });
+      }).finally(function () {
+        workerA.quit();
+        workerB.quit();
+      });
+    });
+
+    it('should be able to reserve two large jobs on different tubes in parallel', function () {
+      let workerA = new Client(host, port)
+        , workerB = new Client(host, port)
+        , tubeA = Math.random().toString()
+        , tubeB = Math.random().toString()
+        , values = {};
+
+      for (let i = 0; i < 1750; i++) {
+        values[Math.random().toString()] = Math.random().toString();
+      }
+
+      return Promise.join(
+        workerA.connect(),
+        workerB.connect()
+      ).then(() => {
+        return Promise.join(
+          this.client.use(tubeA).then(() => {
+            return this.client.put(0, 0, 180, JSON.stringify(values));
+          }).then(() => {
+            return this.client.use(tubeB).then(() => {
+              return this.client.put(0, 0, 180, JSON.stringify(values));
+            })
+          }),
+          workerA.watch(tubeA).then(function () {
+            return workerA.ignore('default');
+          }),
+          workerB.watch(tubeB).then(function () {
+            return workerB.ignore('default');
+          })
+        );
+      }).then(() => {
+        return Promise.join(
+          workerA.reserveWithTimeout(0).spread((reserveId, body) => {
+            expect(JSON.parse(body.toString())).to.deep.equal(values);
+          }),
+          workerB.reserveWithTimeout(0).spread((reserveId, body) => {
+            expect(JSON.parse(body.toString())).to.deep.equal(values);
+          })
+        );
+      }).finally(function () {
+        workerA.quit();
+        workerB.quit();
+      });
+    });
   });
 });

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

50492

increased by5.63%

Lines of code

1013

increased by9.16%

No dependency changes