def connect(path, options = {}, &block)
      stream_parameters, callbacks = connection_options(path, options)

      @stream = EM::Twitter::Client.connect(stream_parameters)
      @stream.each do |item|
        begin
          hash = MultiJson.decode(item, :symbolize_keys => true)
        rescue MultiJson::DecodeError
          invoke_callback(callbacks['error'], "MultiJson::DecodeError occured in stream: #{item}")
          next
        end

        unless hash.is_a?(::Hash)
          invoke_callback(callbacks['error'], "Unexpected JSON object in stream: #{item}")
          next
        end

        respond_to(hash, callbacks, &block)

        yield_message_to(callbacks['anything'], hash)
      end

      @stream.on_close { invoke_callback(callbacks['close']) }

      @stream.on_error do |message|
        invoke_callback(callbacks['error'], message)
      end

      @stream.on_unauthorized { invoke_callback(callbacks['unauthorized']) }

      @stream.on_enhance_your_calm { invoke_callback(callbacks['enhance_your_calm']) }

      @stream.on_reconnect do |timeout, retries|
        invoke_callback(callbacks['reconnect'], timeout, retries)
      end

      @stream.on_max_reconnects do |timeout, retries|
        fail TweetStream::ReconnectError.new(timeout, retries)
      end

      @stream.on_no_data_received { invoke_callback(callbacks['no_data_received']) }

      @stream
    end
def connect(path, options = {}, &block)
      stream_parameters, callbacks = connection_options(path, options)

      @stream = EM::Twitter::Client.connect(stream_parameters)
      @stream.each do |item|
        begin
          hash = MultiJson.decode(item, :symbolize_keys => true)
        rescue MultiJson::DecodeError
          invoke_callback(callbacks['error'], "MultiJson::DecodeError occured in stream: #{item}")
          next
        end

        unless hash.is_a?(::Hash)
          invoke_callback(callbacks['error'], "Unexpected JSON object in stream: #{item}")
          next
        end

        respond_to(hash, callbacks, &block)

        yield_message_to(callbacks['anything'], hash)
      end

      @stream.on_close { invoke_callback(callbacks['close']) }

      @stream.on_error do |message|
        invoke_callback(callbacks['error'], message)
      end

      @stream.on_unauthorized { invoke_callback(callbacks['unauthorized']) }

      @stream.on_enhance_your_calm { invoke_callback(callbacks['enhance_your_calm']) }

      @stream.on_reconnect do |timeout, retries|
        invoke_callback(callbacks['reconnect'], timeout, retries)
      end

      @stream.on_max_reconnects do |timeout, retries|
        fail TweetStream::ReconnectError.new(timeout, retries)
      end

      @stream.on_no_data_received { invoke_callback(callbacks['no_data_received']) }

      @stream
    end
connect to twitter without starting a new EventMachine run loop