text: var assert = require('assert')
    , http = require('http')
    , EventEmitter = require('events').EventEmitter
    , rewire = require('rewire')
    , sinon = require('sinon')
    , Twit = require('../lib/twitter')
    , config1 = require('../config1')
    , config2 = require('../config2')
    , colors = require('colors')
    , helpers = require('./helpers')
    , util = require('util')
    , zlib = require('zlib')
    , async = require('async')
    , restTest = require('./rest');
  
  /**
   * Stop the stream and check the tweet we got back.
   * Call @done on completion.
   *
   * @param  {object}   stream object returned by twit.stream()
   * @param  {Function} done   completion callback
   */
  exports.checkStream = function (stream, done) {
    stream.on('connected', function () {
      console.log('\nconnected'.grey)
    });
  
    stream.once('tweet', function (tweet) {
      assert.equal(null, stream._abortedBy)
  
      stream.stop()
  
      assert.equal('twit-client', stream._abortedBy)
      assert.ok(tweet)
      assert.equal('string', typeof tweet.text)
      assert.equal('string', typeof tweet.id_str)
  
      console.log(('\ntweet: '+tweet.text).grey)
  
      done()
    });
  
    stream.on('reconnecting', function (req, res, connectInterval) {
      console.log('Got disconnected. Scheduling reconnect! statusCode:', res.statusCode, 'connectInterval', connectInterval)
    });
  
    stream.on('error', function (err) {
      console.log('Stream emitted an error', err)
      return done(err)
    })
  }
  
  /**
   * Check the stream state is correctly set for a stopped stream.
   *
   * @param  {object}   stream object returned by twit.stream()
   */
  exports.checkStreamStopState = function (stream) {
    assert.strictEqual('twit-client', stream._abortedBy)
    assert.strictEqual(stream._connectInterval, 0)
    assert.strictEqual(stream._usedFirstReconnect, false)
    assert.strictEqual(stream._scheduledReconnect, undefined)
    assert.strictEqual(stream._stallAbortTimeout, undefined)
  }
  
  describe('Streaming API', function () {
  
    it('statuses/sample', function (done) {
      var twit = new Twit(config1);
      var stream = twit.stream('statuses/sample')
  
      exports.checkStream(stream, done)
    })
  
    it('statuses/filter using `track`', function (done) {
      this.timeout(120000)
      var twit = new Twit(config2);
      var stream = twit.stream('statuses/filter', { track: 'fun' })
  
      exports.checkStream(stream, done)
    })
  
    it('statuses/filter using `locations` string', function (done) {
      var twit = new Twit(config1);
      var world = '-180,-90,180,90';
      var stream = twit.stream('statuses/filter', { locations: world })
  
      exports.checkStream(stream, done)
    })
  
    it('statuses/filter using `locations` array for San Francisco and New York', function (done) {
      var twit = new Twit(config2);
      var params = {
        locations: [ '-122.75', '36.8', '121.75', '37.8', '-74', '40', '73', '41' ]
      }
  
      var stream = twit.stream('statuses/filter', params)
  
      exports.checkStream(stream, done)
    })
  
    it('statuses/filter using `track` array', function (done) {
      var twit = new Twit(config1);
      var params = {
        track: [ 'twitter', ':)', 'fun' ]
      }
  
      var stream = twit.stream('statuses/filter', params)
  
      exports.checkStream(stream, done)
    })
  
    it('statuses/filter using `track` and `language`', function (done) {
      var twit = new Twit(config1);
      var params = {
        track: [ 'twitter', '#apple', 'google', 'twitter', 'facebook', 'happy', 'party', ':)' ],
        language: 'en'
      }
  
      var stream = twit.stream('statuses/filter', params)
  
      exports.checkStream(stream, done)
    })
  
    it('stopping & restarting the stream works', function (done) {
      var twit = new Twit(config2);
      var stream = twit.stream('statuses/sample')
  
      //stop the stream after 2 seconds
      setTimeout(function () {
        assert.equal(null, stream._abortedBy)
        stream.stop()
  
        exports.checkStreamStopState(stream)
  
        console.log('\nstopped stream')
      }, 2000)
  
      //after 3 seconds, start the stream, and stop after 'connect'
      setTimeout(function () {
        stream.once('connected', function (req) {
          console.log('\nrestarted stream')
          stream.stop()
  
          exports.checkStreamStopState(stream)
  
          console.log('\nstopped stream')
          done()
        })
  
        //restart the stream
        stream.start()
      }, 3000)
    })
  
    it('stopping & restarting stream emits to previously assigned callbacks', function (done) {
      var twit = new Twit(config1);
      var stream = twit.stream('statuses/sample')
  
      var started = false
      var numTweets = 0
      stream.on('tweet', function (tweet) {
        process.stdout.write('.')
        if (!started) {
          started = true
          numTweets++
          console.log('received tweet', numTweets)
  
          console.log('stopping stream')
          stream.stop()
  
          exports.checkStreamStopState(stream)
  
          // we've successfully received a new tweet after restarting, test successful
          if (numTweets === 2) {
            done()
          } else {
            started = false
            console.log('restarting stream')
  
            setTimeout(function () {
              stream.start()
            }, 1000)
          }
        }
      })
  
      stream.on('limit', function (limitMsg) {
        console.log('limit', limitMsg)
      })
  
      stream.on('disconnect', function (disconnMsg) {
        console.log('disconnect', disconnMsg)
      })
  
      stream.on('reconnect', function (req, res, ival) {
        console.log('reconnect. statusCode:', res.statusCode, 'interval:', ival)
      })
  
      stream.on('connect', function (req) {
        console.log('connect')
      })
  
    })
  })
  
  describe('streaming API direct message events', function () {
      var senderScreenName;
      var receiverScreenName;
      var twitSender;
      var twitReceiver;
  
      // before we send direct messages the user receiving the DM
      // has to follow the sender. Make this so.
      before(function (done) {
        twitSender = new Twit(config1);
        twitReceiver = new Twit(config2);
  
        // get sender/receiver names in parallel, then make the receiver follow the sender
        async.parallel({
          // get sender screen name and set it for tests to use
          getSenderScreenName: function (parNext) {
            console.log('getting sender user screen_name')
  
            twitSender.get('account/verify_credentials', { twit_options: { retry: true } }, function (err, reply) {
              assert(!err, err)
  
              assert(reply)
              assert(reply.screen_name)
  
              senderScreenName = reply.screen_name
  
              return parNext()
            })
          },
          // get receiver screen name and set it for tests to use
          getReceiverScreenName: function (parNext) {
            console.log('getting receiver user screen_name')
            twitReceiver.get('account/verify_credentials', { twit_options: { retry: true } }, function (err, reply) {
              assert(!err, err)
  
              assert(reply)
              assert(reply.screen_name)
  
              receiverScreenName = reply.screen_name
  
              return parNext()
            })
          }
        }, function (err) {
          assert(!err, err)
  
          var followParams = { screen_name: senderScreenName }
          console.log('making receiver user follow the sender user')
          // make receiver follow sender
          twitReceiver.post('friendships/create', followParams, function (err, reply) {
            assert(!err, err)
            assert(reply.following)
  
            done()
          })
        })
      })
  
      it('user_stream `direct_message` event', function (done) {
        // User A follows User B
        // User A connects to their user stream
        // User B posts a DM to User A
        // User A receives it in their user stream
        this.timeout(0);
  
        // build out DM params
        function makeDmParams () {
          return {
            screen_name: receiverScreenName,
            text: helpers.generateRandomString(10) + ' direct message streaming event test! :-) ' + helpers.generateRandomString(20),
            twit_options: {
              retry: true
            }
          }
        }
  
        var dmIdsReceived = []
        var dmIdsSent = []
        var sentDmFound = false
  
        // start listening for user stream events
        var receiverStream = twitReceiver.stream('user')
  
        console.log('\nlistening for DMs')
        // listen for direct_message event and check DM once it's received
        receiverStream.on('direct_message', function (directMsg) {
          if (sentDmFound) {
            // don't call `done` more than once
            return
          }
  
          console.log('got DM event. id:', directMsg.direct_message.id_str)
          restTest.checkDm(directMsg.direct_message)
          dmIdsReceived.push(directMsg.direct_message.id_str)
  
          // make sure one of the DMs sent was found
          // (we can send multiple DMs if our stream has to reconnect)
          sentDmFound = dmIdsSent.some(function (dmId) {
            return dmId == directMsg.direct_message.id_str
          })
  
          if (!sentDmFound) {
            console.log('this DM doesnt match our test DMs - still waiting for a matching one.')
            console.log('dmIdsSent', dmIdsSent)
            return
          }
  
          receiverStream.stop()
          return done()
        })
  
        var lastTimeSent = 0
        var msToWait = 0
        var postDmInterval = null
  
        receiverStream.on('connected', function () {
          var dmParams = makeDmParams()
  
          console.log('sending a new DM:', dmParams.text)
          twitSender.post('direct_messages/new', dmParams, function (err, reply) {
            assert(!err, err)
            assert(reply)
            restTest.checkDm(reply)
            assert(reply.id_str)
            // we will check this dm against the reply recieved in the message event
            dmIdsSent.push(reply.id_str)
  
            console.log('successfully posted DM:', reply.text, reply.id_str)
            if (dmIdsReceived.indexOf(reply.id_str) !== -1) {
              // our response to the DM posting lost the race against the direct_message
              // listener (we already got the event). So we can finish the test.
              done()
            }
          })
        })
  
        after(function (done) {
          console.log('cleaning up DMs:', dmIdsSent)
          // delete the DMs we posted
          var deleteDms = dmIdsSent.map(function (dmId) {
            return function (next) {
              assert.equal(typeof dmId, 'string')
              console.log('\ndeleting DM', dmId)
              var params = { id: dmId, twit_options: { retry: true } }
              twitSender.post('direct_messages/destroy', params, function (err, reply) {
                assert(!err, err)
                restTest.checkDm(reply)
                assert.equal(reply.id, dmId)
                return next()
              })
            }
          })
          async.parallel(deleteDms, done)
        })
      })
  })
  
  describe('streaming API friends preamble', function () {
    it('returns an array of strings if stringify_friend_ids is true', function (done) {
      var twit = new Twit(config1);
      var stream = twit.stream('user', { stringify_friend_ids: true });
      stream.on('friends', function (friendsObj) {
        assert(friendsObj)
        assert(friendsObj.friends_str)
        if (friendsObj.friends_str.length) {
          assert.equal(typeof friendsObj.friends_str[0], 'string')
        } else {
          console.log('\nEmpty friends preamble:', friendsObj, '. Make some friends on Twitter! ^_^')
        }
        done()
      })
    })
  })
  
  describe('streaming API bad request', function (done) {
    it('emits an error for a 401 response', function (done) {
      var badCredentials = {
          consumer_key: 'a'
        , consumer_secret: 'b'
        , access_token: 'c'
        , access_token_secret: 'd'
      }
  
      var twit = new Twit(badCredentials);
  
      var stream = twit.stream('statuses/filter', { track : ['foo'] });
  
      stream.on('parser-error', function (err) {
        assert.equal(err.statusCode, 401)
        assert(err.twitterReply)
  
        return done()
      })
    })
  })
  
  describe('streaming API `messages` event', function (done) {
    var request = require('request');
    var originalPost = request.post;
    var RewiredTwit = rewire('../lib/twitter');
    var RewiredStreamingApiConnection = rewire('../lib/streaming-api-connection');
    var revertParser, revertTwit;
  
    var MockParser = function () {
      var self = this;
      EventEmitter.call(self);
      process.nextTick(function () {
        self.emit('element', {scrub_geo: 'bar'})
        self.emit('element', {limit: 'buzz'})
      });
    }
    util.inherits(MockParser, EventEmitter);
  
    before(function () {
      revertTwit = RewiredTwit.__set__('StreamingAPIConnection', RewiredStreamingApiConnection);
      revertParser = RewiredStreamingApiConnection.__set__('Parser', MockParser);
  
      request.post = function () { return new helpers.FakeRequest() }
    })
  
    after(function () {
      request.post = originalPost;
      revertTwit();
      revertParser();
    })
  
    it('is returned for 2 different event types', function (done) {
      var twit = new RewiredTwit(config1);
      var stream = twit.stream('statuses/sample');
      var gotScrubGeo = false;
      var gotLimit = false;
      var numMessages = 0;
  
      var maybeDone = function () {
        if (gotScrubGeo && gotLimit && numMessages == 2) {
          done()
        }
      }
  
      stream.on('limit', function () {
        gotLimit = true;
        maybeDone();
      });
      stream.on('scrub_geo', function () {
        gotScrubGeo = true;
        maybeDone();
      })
  
      stream.on('message', function (msg) {
        numMessages++;
        maybeDone();
      })
    })
  })
  
  describe('streaming reconnect', function (done) {
    it('correctly implements connection closing backoff', function (done) {
      var stubPost = function () {
        var fakeRequest = new helpers.FakeRequest()
        process.nextTick(function () {
          fakeRequest.emit('close')
        })
        return fakeRequest
      }
  
      var request = require('request')
      var stubPost = sinon.stub(request, 'post', stubPost)
  
      var twit = new Twit(config1);
      var stream = twit.stream('statuses/filter', { track: [ 'fun', 'yolo']});
  
      var reconnects = [0, 250, 500, 750]
      var reconnectCount = -1
  
      var testDone = false
  
      stream.on('reconnect', function () {
        if (testDone) {
          return
        }
        reconnectCount += 1
        var expectedInterval = reconnects[reconnectCount]
  
        // make sure our connect interval is correct
        assert.equal(stream._connectInterval, expectedInterval);
  
        // simulate immediate reconnect by forcing a new connection (`self._connectInterval` parameter unchanged)
        stream._startPersistentConnection();
  
        if (reconnectCount === reconnects.length -1) {
          // restore request.post
          stubPost.restore()
          testDone = true
          return done();
        }
      });
    });
  
    it('correctly implements 420 backoff', function (done) {
      var stubPost = function () {
        var fakeRequest = new helpers.FakeRequest()
        process.nextTick(function () {
          var fakeResponse = new helpers.FakeResponse(420)
          fakeRequest.emit('response', fakeResponse)
          fakeRequest.emit('close')
        })
        return fakeRequest
      }
  
      var request = require('request')
      var stubPost = sinon.stub(request, 'post', stubPost)
  
      var twit = new Twit(config1);
      var stream = twit.stream('statuses/filter', { track: [ 'fun', 'yolo']});
  
      var reconnects = [60000, 120000, 240000, 480000]
      var reconnectCount = -1
      var testComplete = false
  
      stream.on('reconnect', function (req, res, connectInterval) {
        if (testComplete) {
          // prevent race between last connection attempt firing a reconnect and us validating the final
          // reconnect value in `reconnects`
          return
        }
  
        reconnectCount += 1
         var expectedInterval = reconnects[reconnectCount]
  
        // make sure our connect interval is correct
        assert.equal(stream._connectInterval, connectInterval);
        assert.equal(stream._connectInterval, expectedInterval);
        // simulate immediate reconnect by forcing a new connection (`self._connectInterval` parameter unchanged)
        stream._startPersistentConnection();
  
        if (reconnectCount === reconnects.length -1) {
          // restore request.post
          stubPost.restore()
          testComplete = true
          return done();
        }
      });
    });
  });
  
  describe('Streaming API disconnect message', function (done) {
    it('results in stopping the stream', function (done) {
      var stubPost = function () {
        var fakeRequest = new helpers.FakeRequest()
        process.nextTick(function () {
          var body = zlib.gzipSync(JSON.stringify({disconnect: true}) + '\r\n')
          var fakeResponse = new helpers.FakeResponse(200, body)
          fakeRequest.emit('response', fakeResponse);
          fakeResponse.emit('close')
        });
        return fakeRequest
      }
  
      var request = require('request')
      var origRequest = request.post
      var stubs = sinon.collection
      stubs.stub(request, 'post', stubPost)
  
      var twit = new Twit(config1);
      var stream = twit.stream('statuses/filter', { track: ['fun']});
  
      stream.on('disconnect', function (disconnMsg) {
        stream.stop();
        // restore stub
        request.post = origRequest
        done();
      })
    })
  });
  
  describe('Streaming API Connection limit exceeded message', function (done) {
    it('results in an `error` event containing the message', function (done) {
      var errMsg = 'Exceeded connection limit for user';
  
      var stubPost = function () {
        var fakeRequest = new helpers.FakeRequest();
        process.nextTick(function () {
          var body = zlib.gzipSync(errMsg + '\r\n');
          var fakeResponse = new helpers.FakeResponse(200, body);
          fakeRequest.emit('response', fakeResponse);
          fakeResponse.emit('close');
        });
        return fakeRequest
      }
  
      var request = require('request');
      var origRequest = request.post;
      var stubs = sinon.collection;
      stubs.stub(request, 'post', stubPost);
  
      var twit = new Twit(config1);
      var stream = twit.stream('statuses/filter');
  
      stream.on('error', function (err) {
        assert(err.toString().indexOf(errMsg) !== -1, 'Unexpected error msg:' + errMsg + '.');;
        stream.stop();
        // restore stub
        request.post = origRequest;
        done();
      })
    })
  })
  
  describe('Streaming API connection management', function () {
    it('.stop() works in all states', function (done) {
      var stubPost = function () {
        var fakeRequest = new helpers.FakeRequest();
        process.nextTick(function () {
          var body = zlib.gzipSync('Foobar\r\n');
          var fakeResponse = new helpers.FakeResponse(200, body);
          fakeRequest.emit('response', fakeResponse);
        });
        return fakeRequest
      }
  
      var request = require('request');
      var origRequest = request.post;
      var stubs = sinon.collection;
      stubs.stub(request, 'post', stubPost);
  
      var twit = new Twit(config1);
  
      var stream = twit.stream('statuses/sample');
      stream.stop();
      console.log('\nStopped. Restarting..');
      stream.start();
      stream.once('connect', function(request) {
        console.log('Stream emitted `connect`. Stopping & starting stream..')
        stream.stop();
  
        stream.once('connected', function () {
          console.log('Stream emitted `connected`. Stopping stream.');
          stream.stop();
  
          stubs.restore();
          done();
        });
        stream.start();
      });
    })
  })
  