rhea - npm Package Compare versions

Comparing version 3.0.0 to 3.0.1

lib/connection.js

@@ -550,4 +550,15 @@ /*
         log.io('[%s] read %d bytes', this.options.id, buff.length);
-        if (this.previous_input) {
-            buffer = Buffer.concat([this.previous_input, buff], this.previous_input.length + buff.length);
+        if (this.frame_size) {
+            this.received_bytes += buff.length;
+            this.chunks.push(buff);
+            if (this.frame_size <= this.received_bytes) {
+                buffer = Buffer.concat(this.chunks, this.received_bytes);
+                this.chunks = null;
+                this.frame_size = undefined;
+            } else {
+                log.io('[%s] pushed %d bytes', this.options.id, buff.length);
+                return;
+            }
+        } else if (this.previous_input) {
+            buffer = Buffer.concat([this.previous_input, buff]);
             this.previous_input = null;
@@ -557,5 +568,13 @@ } else {
         }
-        var read = this.transport.read(buffer, this);
+        const read = this.transport.read(buffer, this);
         if (read < buffer.length) {
-            this.previous_input = buffer.slice(read);
+            const previous_input = buffer.slice(read);
+            this.frame_size = this.transport.peek_size(previous_input);
+            if (this.frame_size) {
+                this.chunks = [previous_input];
+                this.received_bytes = previous_input.length;
+                log.io('[%s] waiting frame_size %s', this.options.id, this.frame_size);
+            } else {
+                this.previous_input = previous_input;
+            }
         }
@@ -562,0 +581,0 @@ if (this.local.open.idle_time_out) this.heartbeat_in = setTimeout(this.idle.bind(this), this.local.open.idle_time_out);

lib/sasl.js

@@ -230,2 +230,10 @@ /*
 
+SaslServer.prototype.peek_size = function (buffer) {
+    if (this.transport.read_complete) {
+        return this.next.peek_size(buffer);
+    } else {
+        return this.transport.peek_size(buffer);
+    }
+};
+
 SaslServer.prototype.read = function (buffer) {
@@ -327,2 +335,10 @@ if (this.transport.read_complete) {
 
+SaslClient.prototype.peek_size = function (buffer) {
+    if (this.transport.read_complete) {
+        return this.next.peek_size(buffer);
+    } else {
+        return this.transport.peek_size(buffer);
+    }
+};
+
 SaslClient.prototype.read = function (buffer) {
@@ -357,2 +373,9 @@ if (this.transport.read_complete) {
 
+SelectiveServer.prototype.peek_size = function (buffer) {
+    if (this.header_received) {
+        return this.selected.peek_size(buffer);
+    }
+    return undefined;
+};
+
 SelectiveServer.prototype.read = function (buffer) {
@@ -359,0 +382,0 @@ if (!this.header_received) {

lib/session.js

@@ -162,2 +162,3 @@ /*
         'format':format ? format : 0,
+        'next_to_send': 0,
         'sent': false,
@@ -233,5 +234,7 @@ 'settled': false,
             if (d.link.has_credit()) {
-                if (this.transfer_window() >= d.data.length) {
-                    this.window -= d.data.length;
-                    for (var i = 0; i < d.data.length; i++) {
+                const num_to_send = Math.min(this.transfer_window(), d.data.length - d.next_to_send);
+                if (num_to_send > 0) {
+                    this.window -= num_to_send;
+                    const end_of_send = d.next_to_send + num_to_send;
+                    for (var i = d.next_to_send; i < end_of_send; i++) {
                         this.next_transfer_id++;
@@ -241,9 +244,14 @@ var more = (i+1) < d.data.length;
                         d.link.session.output(transfer, d.data[i]);
+                    }
+                    if (end_of_send < d.data.length) {
+                        d.next_to_send = end_of_send;
+                        break;
+                    } else {
                         if (d.settled) {
                             d.remote_settled = true;//if sending presettled, it can now be cleaned up
                         }
+                        d.link.credit--;
+                        d.link.delivery_count++;
+                        this.next_pending_delivery++;
                     }
-                    d.link.credit--;
-                    d.link.delivery_count++;
-                    this.next_pending_delivery++;
                 } else {
@@ -312,3 +320,2 @@ log.flow('[%s] Incoming window of peer preventing sending further transfers: remote_window=%d, remote_next_transfer_id=%d, next_transfer_id=%d',
         var current;
-        var data;
         if (receiver._incomplete) {
@@ -319,7 +326,3 @@ current = receiver._incomplete;
             }
-            if (frame.payload) {
-                data = Buffer.concat([current.data, frame.payload], current.data.length + frame.payload.length);
-            } else {
-                data = current.data;
-            }
+            current.frames.push(frame.payload);
         } else if (this.next_delivery_id === frame.performative.delivery_id) {
@@ -333,3 +336,5 @@ current = {'id':frame.performative.delivery_id,
                 'remote_settled': frame.performative.settled === undefined ? false : frame.performative.settled,
-                'remote_state': frame.performative.state};
+                'remote_state': frame.performative.state,
+                'frames': [frame.payload],
+            };
             var self = this;
@@ -356,3 +361,2 @@ current.update = function (settled, state) {
             this.next_delivery_id++;
-            data = frame.payload;
         } else {
@@ -365,5 +369,6 @@ //TODO: better error handling
             receiver._incomplete = current;
-            current.data = data;
         } else {
             receiver._incomplete = undefined;
+            const data = current.frames.length === 1 ? current.frames[0] : Buffer.concat(current.frames);
+            delete current.frames;
             if (receiver.credit > 0) receiver.credit--;
@@ -370,0 +375,0 @@ else console.error('Received transfer when credit was %d', receiver.credit);

lib/transport.js

@@ -67,2 +67,10 @@ /*
 
+Transport.prototype.peek_size = function (buffer) {
+    log.frames('[%s] peek_size %o, %d', this.identifier, this.header_received, buffer.length);
+    if (this.header_received && buffer.length >= 4) {
+        return buffer.readUInt32BE();
+    }
+    return undefined;
+};
+
 Transport.prototype.read = function (buffer) {
@@ -69,0 +77,0 @@ var offset = 0;

package.json

 {
     "name": "rhea",
-    "version": "3.0.0",
+    "version": "3.0.1",
     "description": "reactive AMQP 1.0 library",
@@ -38,3 +38,3 @@ "homepage": "http://github.com/amqp/rhea",
         "coverage": "nyc --reporter=json --reporter=lcov --reporter=text-summary npm run test",
-        "browserify": "browserify -r .:rhea -o dist/rhea.js",
+        "browserify": "browserify -r .:rhea -o dist/rhea.js && browserify -r .:rhea --standalone rhea-umd > dist/rhea-umd.js",
         "run-examples": "require-self && mocha examples/test_examples.js",
@@ -41,0 +41,0 @@ "uglify": "uglifyjs --source-map --output dist/rhea.min.js dist/rhea.js",

typings/sasl.d.ts

@@ -72,2 +72,3 @@ /// <reference types="node" />
   write(socket: Socket): void;
+  peek_size(buffer: Buffer): number | undefined;
   read(buffer: Buffer): number;
@@ -91,2 +92,3 @@ }
   write(socket: Socket): void;
+  peek_size(buffer: Buffer): number | undefined;
   read(buffer: Buffer): number;
@@ -105,2 +107,3 @@ }
   write(socket: Socket): void | number;
+  peek_size(buffer: Buffer): number | undefined;
   read(buffer: Buffer): number;
@@ -107,0 +110,0 @@ }

typings/transport.d.ts

@@ -19,3 +19,4 @@ /// <reference types="node" />
   write(socket: Socket): void;
+  peek_size(buffer: Buffer): number | undefined;
   read(buffer: Buffer): number;
 }

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

1350829

increased by32.44%

Number of package files

44

increased by2.33%

Lines of code

25027

increased by62.53%

Worsened metrics

Number of low supply chain risk alerts

30

increased by42.86%

Number of medium supply chain risk alerts

27

increased by35%

No dependency changes