                     ck.end()
                     
 
+            read: (filter, callback) ->
+                reader = @createReader()
+                events = []
+                reader.on 'error', => 
+                    events = []
+                    callback.apply  @, arguments
+                reader.on 'data', (data) =>
+                    concat events, data
+                reader.on 'end', =>
+                    args = slice.call arguments
+                    args.unshift events
+                    args.unshift null
+                    callback.apply @, args
+                reader.read filter
+
             write: (commit, callback) ->
                 committer = @createCommitter()
                 committer.on 'commit', =>

                    streams = [
                        client.stream('zrangebyscore', id, filter.minRevision)
                        es.parse()
                        flatten
                    ]
                    pipe = es.pipeline.apply @, streams
                    #proxy stream commands to our pipe

                    pipe.write filter.maxRevision
                    pipe.end()
                reader

            createCommitter: ->
                emitter = es.map (commit, next) ->
