var pull     = require('pull-stream')
var toPull   = require('stream-to-pull-stream')
var pushable = require('pull-pushable')
var cat      = require('pull-cat')
var window   = require('pull-window')
var fixRange = require('level-fix-range')
var post     = require('level-post')

function read(db, opts) {
  return toPull(db.createReadStream(fixRange(opts)))
}

var live = 
exports.live = 
function (db, opts) {
  opts = opts || {}
  fixRange(opts)

  var l = pushable()

  var cleanup = post(db, opts, function (ch) {
    l.push(ch)
  })

  return l.pipe(pull.through(null, cleanup))

}

exports.read =
exports.readStream = 
exports.createReadStream = function (db, opts) {
  opts = opts || {}
  fixRange(opts)
  if(!opts.tail)
    return read(db, opts)

  //optionally notify when we switch from reading history to realtime
  var sync = opts.onSync && function (abort, cb) {
      opts.onSync(); cb(true)
    }

  return cat([read(db, opts), sync, live(db, opts)])
}

exports.write =
exports.writeStream = 
exports.createWriteStream = function (db, opts, done) {
  if('function' === typeof opts)
    done = opts, opts = null
  opts = opts || {}
  return pull.map(function (e) {
    if(e.type) return e
    return {
      key   : e.key, 
      value : e.value,
      type  : e.value == null ? 'del' : 'put'
    }
  })
  .pipe(window(opts.windowSize, opts.windowTime))
  .pipe(pull.asyncMap(function (batch, cb) {
    db.batch(batch, cb)
  }))
  .pipe(pull.onEnd(done))
}


@@ -158,22 +158,22 @@
 t')%0Avar 
-window
+pw    
    = req
@@ -507,24 +507,53 @@
 tion (ch) %7B%0A
+    console.log('post!', ch)%0A
     l.push(c
@@ -574,15 +574,16 @@
 urn 
-l.pipe(
+pull(l, 
 pull
@@ -921,14 +921,28 @@
 ync(
+abort
 ); cb(
+abort %7C%7C 
 true
@@ -1188,24 +1188,34 @@
 %7B%7D%0A  return 
+pull(%0A    
 pull.map(fun
@@ -1222,24 +1222,26 @@
 ction (e) %7B%0A
+  
     if(e.typ
@@ -1252,24 +1252,26 @@
 eturn e%0A    
+  
 return %7B%0A   
@@ -1267,16 +1267,18 @@
 eturn %7B%0A
+  
       ke
@@ -1297,16 +1297,18 @@
  %0A      
+  
 value : 
@@ -1316,16 +1316,18 @@
 .value,%0A
+  
       ty
@@ -1372,29 +1372,33 @@
     
+  
 %7D%0A  
+  
 %7D)
-%0A  .pipe(window
+,%0A    pw.recent
 (opt
@@ -1427,26 +1427,22 @@
 dowTime)
-)%0A  .pipe(
+,%0A    
 pull.asy
@@ -1466,24 +1466,66 @@
 atch, cb) %7B%0A
+      console.log('write batch', batch)%0A  
     db.batch
@@ -1540,38 +1540,45 @@
 cb)%0A
+  
   %7D)
-)%0A  .pipe(pull.onEnd(
+,%0A    pull.drain(null, 
 done)
+%0A  
 )%0A%7D%0A

update pull api