add put and del streams

Author: e-e-e

index.js

@@ -1,4 +1,5 @@
 const PassThrough = require('readable-stream').PassThrough
+const Transform = require('readable-stream').Transform
 const pump = require('pump')
 
 const utils = require('./lib/utils')
@@ -34,13 +35,34 @@ function doAction (action) {
   }
 }
 
+function doActionStream (action) {
+  return function () {
+    const transform = new Transform({
+      objectMode: true,
+      transform (triples, encoding, done) {
+        if (!triples) return done()
+        let entries = (!triples.reduce) ? [triples] : triples
+        entries = entries.reduce((prev, triple) => {
+          return prev.concat(utils.generateBatch(triple, action))
+        }, [])
+        this.push(entries.reverse())
+        done()
+      }
+    })
+    const writeStream = this.db.createWriteStream()
+    transform.pipe(writeStream)
+    return transform
+  }
+}
+
 // this is not implemented in hyperdb yet
 // for now we just put a null value in the db
-Graph.prototype.del = doAction('del')
 
 Graph.prototype.put = doAction('put')
+Graph.prototype.putStream = doActionStream('put')
 
-Graph.prototype.putStream = function (triple) { }
+Graph.prototype.del = doAction('del')
+Graph.prototype.delStream = doActionStream('del')
 
 Graph.prototype.searchStream = function (query, options) {
   const result = new PassThrough({ objectMode: true })

readme.md

@@ -44,7 +44,7 @@ Inserts **Hexastore** formated entries for triple into the graph database.
 
 #### `var stream = db.putStream(triple)`
 
-Returns a writable stream. **This is not yet implemented!**
+Returns a writable stream.
 
 #### `db.get(triple, [callback])`
 
@@ -56,7 +56,7 @@ Remove triples indices from the graph database.
 
 #### `var stream = db.delStream(triple)`
 
-Returns a writable stream for removing entries. **This is not yet implemented!**
+Returns a writable stream for removing entries.
 
 #### `var stream = db.getStream(triple)`
 

test/triple-store.spec.js

@@ -514,25 +514,25 @@ describe('a basic triple store', function () {
     }
   })
 
-  xit('should put triples using a stream', function (done) {
+  it('should put triples using a stream', function (done) {
     var t1 = { subject: 'a', predicate: 'b', object: 'c' }
     var t2 = { subject: 'a', predicate: 'b', object: 'd' }
     var stream = db.putStream()
-    stream.on('close', done)
+    stream.on('end', done)
 
     stream.write(t1)
     stream.end(t2)
   })
 
-  xit('should store the triples written using a stream', function (done) {
+  it('should store the triples written using a stream', function (done) {
     var t1 = { subject: 'a', predicate: 'b', object: 'c' }
     var t2 = { subject: 'a', predicate: 'b', object: 'd' }
     var stream = db.putStream()
 
     stream.write(t1)
     stream.end(t2)
 
-    stream.on('close', function () {
+    stream.on('end', function () {
       var triples = [t1, t2]
       var readStream = db.getStream({ predicate: 'b' })
 
@@ -544,20 +544,20 @@ describe('a basic triple store', function () {
     })
   })
 
-  xit('should del the triples using a stream', function (done) {
+  it('should del the triples using a stream', function (done) {
     var t1 = { subject: 'a', predicate: 'b', object: 'c' }
     var t2 = { subject: 'a', predicate: 'b', object: 'd' }
     var stream = db.putStream()
 
     stream.write(t1)
     stream.end(t2)
 
-    stream.on('close', function () {
+    stream.on('end', function () {
       var delStream = db.delStream()
       delStream.write(t1)
       delStream.end(t2)
 
-      delStream.on('close', function () {
+      delStream.on('end', function () {
         var readStream = db.getStream({ predicate: 'b' })
 
         var results = []