support offset in pull

Refs: Fishrock123/bob#23

index.js

@@ -56,7 +56,7 @@ class FileSource {
     this.sink = sink
   }
 
-  pull (error, buffer) {
+  pull (error, buffer, offset) {
     if (error) {
       if (typeof this.fd === 'number') {
         fs.close(this.fd, (closeError) => {
@@ -76,27 +76,29 @@ class FileSource {
 
         this.fd = fd
 
-        this.readFromFile(buffer)
+        this.readFromFile(buffer, offset)
       })
     } else {
-      this.readFromFile(buffer)
+      this.readFromFile(buffer, offset)
     }
   }
 
-  readFromFile (buffer) {
+  readFromFile (buffer, offset) {
     if (typeof this.fd !== 'number') {
       return this.pull(null, buffer)
     }
 
-    fs.read(this.fd, buffer, 0, buffer.length, this.pos, (error, bytesRead) => {
+    const pos = offset !== undefined ? offset : this.pos
+
+    fs.read(this.fd, buffer, 0, buffer.length, pos, (error, bytesRead) => {
       if (error) {
         fs.close(this.fd, (closeError) => {
           this.fd = null
           this.sink.next(status.error, closeError || error)
         })
       } else {
         if (bytesRead > 0) {
-          this.pos += bytesRead;
+          if (offset === undefined) this.pos += bytesRead;
           this.sink.next(status.continue, null, buffer, bytesRead)
         } else {
           fs.close(this.fd, (closeError) => {

package.json

@@ -4,7 +4,7 @@
   "description": "A file system source for the BOB streaming protocol.",
   "main": "index.js",
   "scripts": {
-    "test": "node ./test/test-basic.js"
+    "test": "node ./test/test-basic.js && node ./test/test-offset.js"
   },
   "repository": {
     "type": "git",

test/assertion-helper-sink.js

@@ -5,10 +5,12 @@ const assert = require('assert').strict
 const status_type = require('bob-status')
 
 class AssertionSink {
-  constructor (assertions_queue) {
+  constructor (assertions_queue, options = {}) {
     this.source = null
     this.bindCb = null
 
+    this.offset = options.offset
+
     this._queue_index = 0
     this._assertions_queue = assertions_queue
   }
@@ -19,7 +21,7 @@ class AssertionSink {
 
     this.source.bindSink(this)
 
-    this.source.pull(null, Buffer.alloc(this._assertions_queue[this._queue_index].length))
+    this.source.pull(null, Buffer.alloc(this._assertions_queue[this._queue_index].length), this.offset)
   }
 
   next (status, error, buffer, bytes) {
@@ -33,7 +35,7 @@ class AssertionSink {
       return this.source.pull(null, Buffer.alloc(0))
     }
 
-    this.source.pull(null, Buffer.alloc(this._assertions_queue[this._queue_index].length))
+    this.source.pull(null, Buffer.alloc(this._assertions_queue[this._queue_index].length), this.offset)
   }
 }
 

test/test-offset.js

@@ -0,0 +1,24 @@
+'use strict'
+
+const FileSource = require('../index.js')
+const AssertionSink = require('./assertion-helper-sink')
+
+const fileSource = new FileSource('./test/fixtures/file')
+const assertionSink = new AssertionSink(
+  [
+    'Chunk two\ndata data\n\n',
+    'Chunk two\ndata data\n\n',
+    'Chunk two\ndata data\n\n'
+  ],
+  {
+    offset: 'Chunk one\ndata data\n\n'.length
+  }
+)
+
+assertionSink.bindSource(fileSource, error => {
+  if (error)
+    console.error('Stream returned ->', error.stack)
+  else {
+    console.log('ok')
+  }
+})