var fs = require('fs');
var path = require('path');
var events = require('events');
var http = require('http');
var util = require('util');
var Primus = require('primus');
var levelAgile = require('level-agile');

module.exports = Twittery;

util.inherits(Twittery, events.EventEmitter);

function Twittery (options) {
  if (!(this instanceof Twittery)) {
    return new Twittery(options);
  }
  events.EventEmitter.call(this);

  options = options || {};

  this.dbOpts = options.db || { host: '127.0.0.1', port: 4444 };
  this.level = levelAgile(this.dbOpts);
  this.level.on('error', this.emit.bind(this, 'error'));

  this.server = http.createServer(this._requestHandler.bind(this));
  this.primus = new Primus(this.server);

  this.primus.on('connection', this._onSocketConnect.bind(this));
  this.primus.on('disconnection', this._onSocketDisconnect.bind(this));

}

Twittery.prototype.listen = function (port, callback) {
  this.server.listen(port, callback);
};

Twittery.prototype._requestHandler = function (req, res) {
  var index = fs.createReadStream(path.join(__dirname, '..', 'public', 'index.html'));
  index.pipe(res);
};

Twittery.prototype._onSocketConnect = function (spark) {
  if (!this.stream) {
    this.stream = this.level.liveStream();
  }
  this.stream.pipe(spark);

};

Twittery.prototype._onSocketDisconnect = function (spark) {
  //
  // If its a single stream we only want to kill it if there are no listeners
  //

};

@@ -73,24 +73,69 @@
 ('events');%0A
+var Transform = require('stream').Transform;%0A
 var http = r
@@ -1238,39 +1238,15 @@
 ) %7B%0A
-  if (!this.stream) %7B%0A    this.
+%0A  var 
 stre
@@ -1279,25 +1279,50 @@
 ();%0A
+%0A
   
-%7D%0A  
+stream%0A    .pipe(
 this.
-s
 tr
-eam
+ansform())%0A    
 .pip
@@ -1484,12 +1484,324 @@
 rs%0A  //%0A%0A%7D;%0A
+%0ATwittery.prototype.transform = function () %7B%0A  var stream = new Transform(%7B objectMode: true %7D);%0A%0A  stream._transform = function (chunk, encoding, done) %7B%0A    chunk = %7B%0A      user: chunk.value.user%5B'screen_name'%5D,%0A      tweet: chunk.value.text%0A    %7D;%0A    stream.push(chunk)%0A    done();%0A  %7D;%0A  return stream;%0A%7D;%0A

add transform to limit data and create a new stream each time to show the prefetch -> live