Revert "Revert "Merge pull request #11 from neo4j/chunking-fixes""

This reverts commit 5da2482.

Author: jakewins

lib/internal/buf.js

@@ -390,7 +390,18 @@ class HeapBuffer extends BaseBuffer {
   }
 
   getSlice ( start, length ) {
-    return new HeapBuffer( this._buffer.slice( start, start + length ) );
+    if( this._buffer.slice ) {
+      return new HeapBuffer( this._buffer.slice( start, start + length ) );
+    } else {
+      // Some platforms (eg. phantomjs) don't support slice, so fall back to a copy
+      // We do this rather than return a SliceBuffer, because sliceBuffer cannot
+      // be passed to native network write ops etc - we need ArrayBuffer for that
+      let copy = new HeapBuffer(length);
+      for (var i = 0; i < length; i++) {
+        copy.putUInt8( i, this.getUInt8( i + start ) );
+      };
+      return copy;
+    }
   }
 
   /** 
@@ -463,20 +474,28 @@ class CombinedBuffer extends BaseBuffer {
         return buffer.getUInt8(position);
       }
     };
-  };
+  }
 
   getInt8 ( position ) {
     // Surely there's a faster way to do this.. some sort of lookup table thing?
     for (let i = 0; i < this._buffers.length; i++) {
       let buffer = this._buffers[i];
       // If the position is not in the current buffer, skip the current buffer
-      if( position > buffer.length ) {
+      if( position >= buffer.length ) {
         position -= buffer.length;
       } else {
         return buffer.getInt8(position);
       }
     };
-  };
+  }
+
+  getFloat64 ( position ) {
+    // At some point, a more efficient impl. For now, we copy the 8 bytes
+    // we want to read and depend on the platform impl of IEEE 754.
+    let b = alloc(8);
+    for (var i = 0; i < 8; i++) { b.putUInt8(i, this.getUInt8( position + i )); };
+    return b.getFloat64(0);
+  }
 }
 
 /**
@@ -492,23 +511,23 @@ class NodeBuffer extends BaseBuffer {
 
   getUInt8 (position) {
     return this._buffer.readUInt8( position );
-  };
+  }
 
   getInt8 (position) {
     return this._buffer.readInt8( position );
-  };
+  }
 
   getFloat64 (position) {
     return this._buffer.readDoubleBE(position);
-  };
+  }
 
   putUInt8 (position, val) {
     this._buffer.writeUInt8( val, position );
-  };
+  }
 
   putInt8 (position, val) {
     this._buffer.writeInt8( val, position );
-  };
+  }
 
   putFloat64 ( position, val ) {
     this._buffer.writeDoubleBE( val, position );
@@ -530,7 +549,7 @@ class NodeBuffer extends BaseBuffer {
 
   getSlice ( start, length ) {
     return new NodeBuffer( this._buffer.slice( start, start + length ) );
-  };
+  }
 }
 
 // Use HeapBuffer by default, unless Buffer API is available, see below

lib/internal/chunking.js

@@ -156,19 +156,14 @@ class Dechunker {
   }
 
   IN_CHUNK ( buf ) {
-
-    if ( this._chunkSize < buf.remaining() ) {
-      // Current packet is larger than current chunk, slice of the chunk
+    if ( this._chunkSize <= buf.remaining() ) {
+      // Current packet is larger than current chunk, or same size:
       this._currentMessage.push( buf.readSlice( this._chunkSize ) );
       return this.AWAITING_CHUNK;
-    } else if ( this._chunkSize == buf.remaining() ) {
-      // Current packet perfectly maps to current chunk
-      this._currentMessage.push( buf.readSlice( buf.length ) );
-      return this.AWAITING_CHUNK;
     } else {
       // Current packet is smaller than the chunk we're reading, split the current chunk itself up
-      this._chunkSize -= data.remaining();
-      this._currentMessage.push( buf.readSlice( buf.length ) );
+      this._chunkSize -= buf.remaining();
+      this._currentMessage.push( buf.readSlice( buf.remaining() ) );
       return this.IN_CHUNK;
     }
   }

test/internal/buf.test.js

@@ -18,6 +18,7 @@
  */
 
 var alloc = require('../../build/node/internal/buf').alloc;
+var CombinedBuffer = require('../../build/node/internal/buf').CombinedBuffer;
 var utf8 = require('../../build/node/internal/utf8');
 var Unpacker = require("../../build/node/internal/packstream.js").Unpacker;
 
@@ -63,19 +64,55 @@ describe('buffers', function() {
   });
 
   it('should decode list correctly', function() {
-    //Given
-    
+    // Given
     var b = alloc(5);
     b.writeUInt8(0x92);
     b.writeUInt8(0x81);
     b.writeUInt8(0x61);
     b.writeUInt8(0x81);
     b.writeUInt8(0x62);
     b.reset();
+
+    // When
     var data = new Unpacker().unpack( b );
+
+    // Then
     expect(data[0]).toBe('a');
     expect(data[1]).toBe('b');
+  });
+});
+
+describe('CombinedBuffer', function() {
+  it('should read int8', function() {
+    // Given
+    var b1 = alloc(1);
+    var b2 = alloc(1);
+    b1.putInt8(0, 1);
+    b2.putInt8(0, 2);
+
+    var b = new CombinedBuffer([b1,b2]);
 
+    // When
+    var first = b.readInt8();
+    var second = b.readInt8();
+
+    // Then
+    expect(first).toBe(1);
+    expect(second).toBe(2);
+  });
+
+  it('should read divided float64', function() {
+    // Given
+    var inner = alloc(8);
+    inner.putFloat64(0, 0.1);
+
+    var b = new CombinedBuffer([inner.readSlice(4),inner.readSlice(4)]);
+    
+    // When
+    var read = b.readFloat64();
+
+    // Then
+    expect(read).toBe(0.1);
   });
 });
 

test/internal/chunking.test.js

@@ -20,6 +20,7 @@
 var Chunker = require('../../build/node/internal/chunking').Chunker;
 var Dechunker = require('../../build/node/internal/chunking').Dechunker;
 var alloc = require('../../build/node/internal/buf').alloc;
+var CombinedBuffer = require('../../build/node/internal/buf').CombinedBuffer;
 
 describe('Chunker', function() {
   it('should chunk simple data', function() {
@@ -83,6 +84,41 @@ describe('Dechunker', function() {
     expect( messages.length ).toBe( 1 );
     expect( messages[0].toHex() ).toBe( "00 01 00 02 00 03 " );
   });
+
+  it('should handle message split at any point', function() {
+    // Given
+    var ch = new TestChannel();
+    var chunker = new Chunker(ch);
+
+    // And given the following message
+    chunker.writeInt8(1);
+    chunker.writeInt16(2);
+    chunker.writeInt32(3);
+    chunker.writeUInt8(4);
+    chunker.writeUInt32(5);
+    chunker.messageBoundary();
+    chunker.flush();
+
+    var chunked = ch.toBuffer();
+
+    // When I try splitting this chunked data at every possible position
+    // into two separate buffers, and send those to the dechunker
+    for (var i = 1; i < chunked.length; i++) {
+      var slice1 = chunked.getSlice( 0, i );
+      var slice2 = chunked.getSlice( i, chunked.length - i );
+
+      // Dechunk the slices
+      var messages = [];
+      var dechunker = new Dechunker();
+      dechunker.onmessage = function(buffer) { messages.push(buffer); };
+      dechunker.write( slice1 );
+      dechunker.write( slice2 );
+
+      // Then, the output should be correct
+      expect( messages.length ).toBe( 1 );
+      expect( messages[0].toHex() ).toBe( "01 00 02 00 00 00 03 04 00 00 00 05 " );
+    };
+  });
 });
 
 function TestChannel() {
@@ -101,6 +137,18 @@ TestChannel.prototype.toHex = function() {
   return out;
 };
 
+TestChannel.prototype.toBuffer = function() {
+  return new CombinedBuffer( this._written );
+};
+
+TestChannel.prototype.toHex = function() {
+  var out = "";
+  for( var i=0; i<this._written.length; i++ ) {
+    out += this._written[i].toHex();
+  }
+  return out;
+};
+
 function bytes() {
   var b = alloc( arguments.length );
   for( var i=0; i<arguments.length; i++ ) {