From 9ea2a6dbd0fd2ea2319b10e240b140a2ad594ed6 Mon Sep 17 00:00:00 2001 From: James Coglan Date: Tue, 22 Nov 2011 01:07:15 +0000 Subject: [PATCH] Extract WebSocket classes from Faye 0.7. --- .gitignore | 2 + .gitmodules | 3 + Gemfile | 3 + faye-websocket.gemspec | 22 ++ lib/faye/thin_extensions.rb | 75 +++++ lib/faye/websocket.rb | 116 +++++++ lib/faye/websocket/api.rb | 103 ++++++ lib/faye/websocket/client.rb | 80 +++++ lib/faye/websocket/draft75_parser.rb | 53 +++ lib/faye/websocket/draft76_parser.rb | 53 +++ lib/faye/websocket/protocol8_parser.rb | 324 +++++++++++++++++++ lib/faye/websocket/publisher.rb | 31 ++ spec/faye/websocket/client_spec.rb | 121 +++++++ spec/faye/websocket/draft75_parser_spec.rb | 41 +++ spec/faye/websocket/protocol8_parser_spec.rb | 145 +++++++++ spec/spec_helper.rb | 43 +++ vendor/em-rspec | 1 + 17 files changed, 1216 insertions(+) create mode 100644 .gitignore create mode 100644 .gitmodules create mode 100644 Gemfile create mode 100644 faye-websocket.gemspec create mode 100644 lib/faye/thin_extensions.rb create mode 100644 lib/faye/websocket.rb create mode 100644 lib/faye/websocket/api.rb create mode 100644 lib/faye/websocket/client.rb create mode 100644 lib/faye/websocket/draft75_parser.rb create mode 100644 lib/faye/websocket/draft76_parser.rb create mode 100644 lib/faye/websocket/protocol8_parser.rb create mode 100644 lib/faye/websocket/publisher.rb create mode 100644 spec/faye/websocket/client_spec.rb create mode 100644 spec/faye/websocket/draft75_parser_spec.rb create mode 100644 spec/faye/websocket/protocol8_parser_spec.rb create mode 100644 spec/spec_helper.rb create mode 160000 vendor/em-rspec diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7ae6fcf --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +Gemfile.lock +*.gem diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..0db9be5 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "vendor/em-rspec"] + path = vendor/em-rspec + url = git://github.com/jcoglan/em-rspec.git diff --git a/Gemfile b/Gemfile new file mode 100644 index 0000000..6281f00 --- /dev/null +++ b/Gemfile @@ -0,0 +1,3 @@ +source "http://rubygems.org" +gemspec + diff --git a/faye-websocket.gemspec b/faye-websocket.gemspec new file mode 100644 index 0000000..afe2b2e --- /dev/null +++ b/faye-websocket.gemspec @@ -0,0 +1,22 @@ +Gem::Specification.new do |s| + s.name = "faye-websocket" + s.version = "0.1.0" + s.summary = "Robust general-purpose WebSocket server and client" + s.author = "James Coglan" + s.email = "jcoglan@gmail.com" + s.homepage = "http://github.com/jcoglan/faye-websocket-ruby" + + # s.extra_rdoc_files = %w[README.rdoc] + # s.rdoc_options = %w[--main README.rdoc] + + s.files = Dir.glob("{lib,spec}/**/*") + + s.require_paths = %w[lib] + + s.add_dependency "eventmachine", ">= 0.12.0" + s.add_dependency "thin", "~> 1.2" + + s.add_development_dependency "rspec", "~> 2.5.0" + s.add_development_dependency "rack" +end + diff --git a/lib/faye/thin_extensions.rb b/lib/faye/thin_extensions.rb new file mode 100644 index 0000000..e4f6ea3 --- /dev/null +++ b/lib/faye/thin_extensions.rb @@ -0,0 +1,75 @@ +# WebSocket extensions for Thin +# Based on code from the Cramp project +# http://github.com/lifo/cramp + +# Copyright (c) 2009-2011 Pratik Naik +# +# Permission is hereby granted, free of charge, to any person obtaining +# a copy of this software and associated documentation files (the +# "Software"), to deal in the Software without restriction, including +# without limitation the rights to use, copy, modify, merge, publish, +# distribute, sublicense, and/or sell copies of the Software, and to +# permit persons to whom the Software is furnished to do so, subject to +# the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE +# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION +# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + +class Thin::Connection + def receive_data(data) + trace { data } + + case @serving + when :websocket + callback = @request.env[Thin::Request::WEBSOCKET_RECEIVE_CALLBACK] + callback.call(data) if callback + else + if @request.parse(data) + if @request.websocket? + @request.env['em.connection'] = self + @response.persistent! + @response.websocket = true + @serving = :websocket + end + + process + end + end + rescue Thin::InvalidRequest => e + log "!! Invalid request" + log_error e + close_connection + end +end + +class Thin::Request + WEBSOCKET_RECEIVE_CALLBACK = 'websocket.receive_callback'.freeze + def websocket? + @env['HTTP_CONNECTION'] and + @env['HTTP_CONNECTION'].split(/\s*,\s*/).include?('Upgrade') and + ['WebSocket', 'websocket'].include?(@env['HTTP_UPGRADE']) + end +end + +class Thin::Response + # Headers for sending Websocket upgrade + attr_accessor :websocket + + def each + yield(head) unless websocket + if @body.is_a?(String) + yield @body + else + @body.each { |chunk| yield chunk } + end + end +end + diff --git a/lib/faye/websocket.rb b/lib/faye/websocket.rb new file mode 100644 index 0000000..6a400a8 --- /dev/null +++ b/lib/faye/websocket.rb @@ -0,0 +1,116 @@ +require 'base64' +require 'digest/md5' +require 'digest/sha1' +require 'net/http' +require 'uri' + +require 'thin' +require File.dirname(__FILE__) + '/thin_extensions' + +module Faye + class WebSocket + + root = File.expand_path('../websocket', __FILE__) + + autoload :API, root + '/api' + autoload :Client, root + '/client' + autoload :Draft75Parser, root + '/draft75_parser' + autoload :Draft76Parser, root + '/draft76_parser' + autoload :Protocol8Parser, root + '/protocol8_parser' + autoload :Publisher, root + '/publisher' + + # http://www.w3.org/International/questions/qa-forms-utf-8.en.php + UTF8_MATCH = /^([\x00-\x7F]|[\xC2-\xDF][\x80-\xBF]|\xE0[\xA0-\xBF][\x80-\xBF]|[\xE1-\xEC\xEE\xEF][\x80-\xBF]{2}|\xED[\x80-\x9F][\x80-\xBF]|\xF0[\x90-\xBF][\x80-\xBF]{2}|[\xF1-\xF3][\x80-\xBF]{3}|\xF4[\x80-\x8F][\x80-\xBF]{2})*$/ + + def self.encode(string, validate_encoding = false) + if Array === string + return nil if validate_encoding and !valid_utf8?(string) + string = string.pack('C*') + end + return string unless string.respond_to?(:force_encoding) + string.force_encoding('UTF-8') + end + + def self.valid_utf8?(byte_array) + UTF8_MATCH =~ byte_array.pack('C*') ? true : false + end + + def self.parser(env) + if env['HTTP_SEC_WEBSOCKET_VERSION'] + Protocol8Parser + elsif env['HTTP_SEC_WEBSOCKET_KEY1'] + Draft76Parser + else + Draft75Parser + end + end + + extend Forwardable + def_delegators :@parser, :version + + attr_reader :env + include API + + def initialize(env) + @env = env + @callback = @env['async.callback'] + @stream = Stream.new(self, @env['em.connection']) + @callback.call [200, {}, @stream] + + @url = determine_url + @ready_state = CONNECTING + @buffered_amount = 0 + + @parser = WebSocket.parser(@env).new(self) + @stream.write(@parser.handshake_response) + + @ready_state = OPEN + + event = Event.new('open') + event.init_event('open', false, false) + dispatch_event(event) + + @env[Thin::Request::WEBSOCKET_RECEIVE_CALLBACK] = @parser.method(:parse) + end + + private + + def determine_url + secure = if @env.has_key?('HTTP_X_FORWARDED_PROTO') + @env['HTTP_X_FORWARDED_PROTO'] == 'https' + else + @env['HTTP_ORIGIN'] =~ /^https:/i + end + + scheme = secure ? 'wss:' : 'ws:' + "#{ scheme }//#{ @env['HTTP_HOST'] }#{ @env['REQUEST_URI'] }" + end + end + + class WebSocket::Stream + include EventMachine::Deferrable + + extend Forwardable + def_delegators :@connection, :close_connection, :close_connection_after_writing + + def initialize(web_socket, connection) + @web_socket = web_socket + @connection = connection + end + + def each(&callback) + @data_callback = callback + end + + def fail + @web_socket.close(1006, '', false) + end + + def write(data) + return unless @data_callback + @data_callback.call(data) + end + end + +end + diff --git a/lib/faye/websocket/api.rb b/lib/faye/websocket/api.rb new file mode 100644 index 0000000..3887f50 --- /dev/null +++ b/lib/faye/websocket/api.rb @@ -0,0 +1,103 @@ +module Faye + class WebSocket + + CONNECTING = 0 + OPEN = 1 + CLOSING = 2 + CLOSED = 3 + + module API + attr_reader :url, :ready_state, :buffered_amount + attr_accessor :onopen, :onmessage, :onerror, :onclose + + include Publisher + + def receive(data) + return false unless ready_state == OPEN + event = Event.new('message') + event.init_event('message', false, false) + event.data = data + dispatch_event(event) + end + + def send(data, type = nil, error_type = nil) + return false if ready_state == CLOSED + data = WebSocket.encode(data) if String === data + frame = @parser.frame(data, type, error_type) + @stream.write(frame) if frame + end + + def close(code = nil, reason = nil, ack = true) + return if [CLOSING, CLOSED].include?(ready_state) + + @ready_state = CLOSING + + close = lambda do + @ready_state = CLOSED + @stream.close_connection_after_writing + event = Event.new('close', :code => code || 1000, :reason => reason || '') + event.init_event('close', false, false) + dispatch_event(event) + end + + if ack + if @parser.respond_to?(:close) + @parser.close(code, reason, &close) + else + close.call + end + else + @parser.close(code, reason) if @parser.respond_to?(:close) + close.call + end + end + + def add_event_listener(type, listener, use_capture) + bind(type, listener) + end + + def remove_event_listener(type, listener, use_capture) + unbind(type, listener) + end + + def dispatch_event(event) + event.target = event.current_target = self + event.event_phase = Event::AT_TARGET + + trigger(event.type, event) + callback = __send__("on#{ event.type }") + callback.call(event) if callback + end + end + + class Event + attr_reader :type, :bubbles, :cancelable + attr_accessor :target, :current_target, :event_phase, :data + + CAPTURING_PHASE = 1 + AT_TARGET = 2 + BUBBLING_PHASE = 3 + + def initialize(event_type, options = {}) + @type = event_type + metaclass = (class << self ; self ; end) + options.each do |key, value| + metaclass.__send__(:define_method, key) { value } + end + end + + def init_event(event_type, can_bubble, cancelable) + @type = event_type + @bubbles = can_bubble + @cancelable = cancelable + end + + def stop_propagation + end + + def prevent_default + end + end + + end +end diff --git a/lib/faye/websocket/client.rb b/lib/faye/websocket/client.rb new file mode 100644 index 0000000..4d78ff5 --- /dev/null +++ b/lib/faye/websocket/client.rb @@ -0,0 +1,80 @@ +module Faye + class WebSocket + + class Client + include API + attr_reader :uri + + def initialize(url) + @parser = Protocol8Parser.new(self, :masking => true) + @url = url + @uri = URI.parse(url) + + @ready_state = CONNECTING + @buffered_amount = 0 + + EventMachine.connect(@uri.host, @uri.port || 80, Connection) do |conn| + @stream = conn + conn.parent = self + end + end + + private + + def on_connect + @stream.start_tls if @uri.scheme == 'wss' + @handshake = @parser.create_handshake + @message = [] + @stream.write(@handshake.request_data) + end + + def receive_data(data) + data = WebSocket.encode(data) + + case @ready_state + when CONNECTING then + @message += @handshake.parse(data) + return unless @handshake.complete? + + if @handshake.valid? + @ready_state = OPEN + event = Event.new('open') + event.init_event('open', false, false) + dispatch_event(event) + + receive_data(@message) + else + @ready_state = CLOSED + event = Event.new('error') + event.init_event('error', false, false) + dispatch_event(event) + end + + when OPEN, CLOSING then + @parser.parse(data) + end + end + + module Connection + attr_accessor :parent + + def connection_completed + parent.__send__(:on_connect) + end + + def receive_data(data) + parent.__send__(:receive_data, data) + end + + def unbind + parent.close(1006, '', false) + end + + def write(data) + send_data(data) + end + end + end + + end +end diff --git a/lib/faye/websocket/draft75_parser.rb b/lib/faye/websocket/draft75_parser.rb new file mode 100644 index 0000000..4320c96 --- /dev/null +++ b/lib/faye/websocket/draft75_parser.rb @@ -0,0 +1,53 @@ +module Faye + class WebSocket + + class Draft75Parser + def initialize(web_socket) + @socket = web_socket + @buffer = [] + @buffering = false + end + + def version + 'draft-75' + end + + def handshake_response + upgrade = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" + upgrade << "Upgrade: WebSocket\r\n" + upgrade << "Connection: Upgrade\r\n" + upgrade << "WebSocket-Origin: #{@socket.env['HTTP_ORIGIN']}\r\n" + upgrade << "WebSocket-Location: #{@socket.url}\r\n" + upgrade << "\r\n" + upgrade + end + + def parse(data) + data.each_byte(&method(:handle_byte)) + end + + def frame(data, type = nil, error_type = nil) + ["\x00", data, "\xFF"].map(&WebSocket.method(:encode)) * '' + end + + private + + def handle_byte(data) + case data + when 0x00 then + @buffering = true + + when 0xFF then + @socket.receive(WebSocket.encode(@buffer)) + @buffer = [] + @buffering = false + + else + @buffer.push(data) if @buffering + end + end + end + + end +end + diff --git a/lib/faye/websocket/draft76_parser.rb b/lib/faye/websocket/draft76_parser.rb new file mode 100644 index 0000000..179ff10 --- /dev/null +++ b/lib/faye/websocket/draft76_parser.rb @@ -0,0 +1,53 @@ +module Faye + class WebSocket + + class Draft76Parser < Draft75Parser + def version + 'draft-76' + end + + def handshake_response + env = @socket.env + + key1 = env['HTTP_SEC_WEBSOCKET_KEY1'] + value1 = number_from_key(key1) / spaces_in_key(key1) + + key2 = env['HTTP_SEC_WEBSOCKET_KEY2'] + value2 = number_from_key(key2) / spaces_in_key(key2) + + hash = Digest::MD5.digest(big_endian(value1) + + big_endian(value2) + + env['rack.input'].read) + + upgrade = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" + upgrade << "Upgrade: WebSocket\r\n" + upgrade << "Connection: Upgrade\r\n" + upgrade << "Sec-WebSocket-Origin: #{env['HTTP_ORIGIN']}\r\n" + upgrade << "Sec-WebSocket-Location: #{@socket.url}\r\n" + upgrade << "\r\n" + upgrade << hash + upgrade + end + + private + + def number_from_key(key) + key.scan(/[0-9]/).join('').to_i(10) + end + + def spaces_in_key(key) + key.scan(/ /).size + end + + def big_endian(number) + string = '' + [24,16,8,0].each do |offset| + string << (number >> offset & 0xFF).chr + end + string + end + end + + end +end + diff --git a/lib/faye/websocket/protocol8_parser.rb b/lib/faye/websocket/protocol8_parser.rb new file mode 100644 index 0000000..5394e88 --- /dev/null +++ b/lib/faye/websocket/protocol8_parser.rb @@ -0,0 +1,324 @@ +module Faye + class WebSocket + + class Protocol8Parser + GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11' + + FIN = MASK = 0b10000000 + RSV1 = 0b01000000 + RSV2 = 0b00100000 + RSV3 = 0b00010000 + OPCODE = 0b00001111 + LENGTH = 0b01111111 + + OPCODES = { + :continuation => 0, + :text => 1, + :binary => 2, + :close => 8, + :ping => 9, + :pong => 10 + } + + FRAGMENTED_OPCODES = OPCODES.values_at(:continuation, :text, :binary) + OPENING_OPCODES = OPCODES.values_at(:text, :binary) + + ERRORS = { + :normal_closure => 1000, + :going_away => 1001, + :protocol_error => 1002, + :unacceptable => 1003, + :encoding_error => 1007, + :policy_violation => 1008, + :too_large => 1009, + :extension_error => 1010 + } + + ERROR_CODES = ERRORS.values + + class Handshake + def initialize(uri) + @uri = uri + @key = Base64.encode64((1..16).map { rand(255).chr } * '').strip + @accept = Base64.encode64(Digest::SHA1.digest(@key + GUID)).strip + @buffer = [] + end + + def request_data + hostname = @uri.host + (@uri.port ? ":#{@uri.port}" : '') + + handshake = "GET #{@uri.path}#{@uri.query ? '?' : ''}#{@uri.query} HTTP/1.1\r\n" + handshake << "Host: #{hostname}\r\n" + handshake << "Upgrade: websocket\r\n" + handshake << "Connection: Upgrade\r\n" + handshake << "Sec-WebSocket-Key: #{@key}\r\n" + handshake << "Sec-WebSocket-Version: 8\r\n" + handshake << "\r\n" + + handshake + end + + def parse(data) + message = [] + complete = false + data.each_byte do |byte| + if complete + message << byte + else + @buffer << byte + complete ||= complete? + end + end + message + end + + def complete? + @buffer[-4..-1] == [0x0D, 0x0A, 0x0D, 0x0A] + end + + def valid? + data = WebSocket.encode(@buffer) + response = Net::HTTPResponse.read_new(Net::BufferedIO.new(StringIO.new(data))) + return false unless response.code.to_i == 101 + + upgrade, connection = response['Upgrade'], response['Connection'] + + upgrade and upgrade =~ /^websocket$/i and + connection and connection.split(/\s*,\s*/).include?('Upgrade') and + response['Sec-WebSocket-Accept'] == @accept + end + end + + def initialize(web_socket, options = {}) + reset + @socket = web_socket + @stage = 0 + @masking = options[:masking] + end + + def version + 'protocol-8' + end + + def handshake_response + sec_key = @socket.env['HTTP_SEC_WEBSOCKET_KEY'] + return '' unless String === sec_key + + accept = Base64.encode64(Digest::SHA1.digest(sec_key + GUID)).strip + + upgrade = "HTTP/1.1 101 Switching Protocols\r\n" + upgrade << "Upgrade: websocket\r\n" + upgrade << "Connection: Upgrade\r\n" + upgrade << "Sec-WebSocket-Accept: #{accept}\r\n" + upgrade << "\r\n" + upgrade + end + + def create_handshake + Handshake.new(@socket.uri) + end + + def parse(data) + data.each_byte do |byte| + case @stage + when 0 then parse_opcode(byte) + when 1 then parse_length(byte) + when 2 then parse_extended_length(byte) + when 3 then parse_mask(byte) + when 4 then parse_payload(byte) + end + emit_frame if @stage == 4 and @length == 0 + end + end + + def frame(data, type = nil, code = nil) + return nil if @closed + + type ||= (String === data ? :text : :binary) + data = data.bytes.to_a if data.respond_to?(:bytes) + + if code + data = [code].pack('n').bytes.to_a + data + end + + frame = (FIN | OPCODES[type]).chr + length = data.size + masked = @masking ? MASK : 0 + + case length + when 0..125 then + frame << (masked | length).chr + when 126..65535 then + frame << (masked | 126).chr + frame << [length].pack('n') + else + frame << (masked | 127).chr + frame << [length >> 32, length & 0xFFFFFFFF].pack('NN') + end + + if @masking + mask = (1..4).map { rand 256 } + data.each_with_index do |byte, i| + data[i] = byte ^ mask[i % 4] + end + frame << mask.pack('C*') + end + + WebSocket.encode(frame) + WebSocket.encode(data) + end + + def close(code = nil, reason = nil, &callback) + return if @closed + @closing_callback ||= callback + @socket.send(reason || '', :close, code || ERRORS[:normal_closure]) + @closed = true + end + + private + + def parse_opcode(data) + if [RSV1, RSV2, RSV3].any? { |rsv| (data & rsv) == rsv } + return @socket.close(ERRORS[:protocol_error], nil, false) + end + + @final = (data & FIN) == FIN + @opcode = (data & OPCODE) + @mask = [] + @payload = [] + + unless OPCODES.values.include?(@opcode) + return @socket.close(ERRORS[:protocol_error], nil, false) + end + + unless FRAGMENTED_OPCODES.include?(@opcode) or @final + return @socket.close(ERRORS[:protocol_error], nil, false) + end + + if @mode and OPENING_OPCODES.include?(@opcode) + return @socket.close(ERRORS[:protocol_error], nil, false) + end + + @stage = 1 + end + + def parse_length(data) + @masked = (data & MASK) == MASK + @length = (data & LENGTH) + + if @length <= 125 + @stage = @masked ? 3 : 4 + else + @length_buffer = [] + @length_size = (@length == 126) ? 2 : 8 + @stage = 2 + end + end + + def parse_extended_length(data) + @length_buffer << data + return unless @length_buffer.size == @length_size + @length = integer(@length_buffer) + @stage = @masked ? 3 : 4 + end + + def parse_mask(data) + @mask << data + return if @mask.size < 4 + @stage = 4 + end + + def parse_payload(data) + @payload << data + return if @payload.size < @length + emit_frame + end + + def emit_frame + payload = unmask(@payload, @mask) + + case @opcode + when OPCODES[:continuation] then + return @socket.close(ERRORS[:protocol_error], nil, false) unless @mode + @buffer += payload + if @final + message = @buffer + message = WebSocket.encode(message, true) if @mode == :text + reset + if message + @socket.receive(message) + else + @socket.close(ERRORS[:encoding_error], nil, false) + end + end + + when OPCODES[:text] then + if @final + message = WebSocket.encode(payload, true) + if message + @socket.receive(message) + else + @socket.close(ERRORS[:encoding_error], nil, false) + end + else + @mode = :text + @buffer += payload + end + + when OPCODES[:binary] then + if @final + @socket.receive(payload) + else + @mode = :binary + @buffer += payload + end + + when OPCODES[:close] then + code = (payload.size >= 2) ? 256 * payload[0] + payload[1] : nil + + unless (payload.size == 0) or + (code && code >= 3000 && code < 5000) or + ERROR_CODES.include?(code) + code = ERRORS[:protocol_error] + end + + if payload.size > 125 or not WebSocket.valid_utf8?(payload[2..-1] || []) + code = ERRORS[:protocol_error] + end + + reason = (payload.size > 2) ? WebSocket.encode(payload[2..-1], true) : nil + @socket.close(code, reason, false) + @closing_callback.call if @closing_callback + + when OPCODES[:ping] then + return @socket.close(ERRORS[:protocol_error], nil, false) if payload.size > 125 + @socket.send(payload, :pong) + end + @stage = 0 + end + + def reset + @buffer = [] + @mode = nil + end + + def integer(bytes) + number = 0 + bytes.each_with_index do |data, i| + number += data << (8 * (bytes.size - 1 - i)) + end + number + end + + def unmask(payload, mask) + unmasked = [] + payload.each_with_index do |byte, i| + byte = byte ^ mask[i % 4] if mask.size > 0 + unmasked << byte + end + unmasked + end + end + + end +end + diff --git a/lib/faye/websocket/publisher.rb b/lib/faye/websocket/publisher.rb new file mode 100644 index 0000000..06e64e2 --- /dev/null +++ b/lib/faye/websocket/publisher.rb @@ -0,0 +1,31 @@ +class Faye::WebSocket + module Publisher + + def count_listeners(event_type) + return 0 unless @subscribers and @subscribers[event_type] + @subscribers[event_type].size + end + + def bind(event_type, &listener) + @subscribers ||= {} + list = @subscribers[event_type] ||= [] + list << listener + end + + def unbind(event_type, &listener) + return unless @subscribers and @subscribers[event_type] + return @subscribers.delete(event_type) unless listener + + @subscribers[event_type].delete_if(&listener.method(:==)) + end + + def trigger(event_type, *args) + return unless @subscribers and @subscribers[event_type] + @subscribers[event_type].each do |listener| + listener.call(*args) + end + end + + end +end + diff --git a/spec/faye/websocket/client_spec.rb b/spec/faye/websocket/client_spec.rb new file mode 100644 index 0000000..a8e822b --- /dev/null +++ b/spec/faye/websocket/client_spec.rb @@ -0,0 +1,121 @@ +# encoding=utf-8 + +require "spec_helper" + +WebSocketSteps = EM::RSpec.async_steps do + def server(port, &callback) + @server = EchoServer.new + @server.listen(port) + @port = port + EM.add_timer(0.1, &callback) + end + + def stop(&callback) + @server.stop + EM.next_tick(&callback) + end + + def open_socket(url, &callback) + done = false + + resume = lambda do |open| + unless done + done = true + @open = open + callback.call + end + end + + @ws = Faye::WebSocket::Client.new(url) + + @ws.onopen = lambda { |e| resume.call(true) } + @ws.onclose = lambda { |e| resume.call(false) } + end + + def close_socket(&callback) + @ws.onclose = lambda do |e| + @open = false + callback.call + end + @ws.close + end + + def check_open(&callback) + @open.should == true + callback.call + end + + def check_closed(&callback) + @open.should == false + callback.call + end + + def listen_for_message(&callback) + @ws.onmessage = lambda { |e| @message = e.data } + callback.call + end + + def send_message(&callback) + @ws.send("I expect this to be echoed") + EM.add_timer(0.1, &callback) + end + + def check_response(&callback) + @message.should == "I expect this to be echoed" + callback.call + end + + def check_no_response(&callback) + @message.should == nil + callback.call + end +end + +describe Faye::WebSocket::Client do + include WebSocketSteps + + before do + Thread.new { EM.run } + sleep(0.1) until EM.reactor_running? + + server 8000 + sync + end + + after { sync ; stop } + + it "can open a connection" do + open_socket "ws://localhost:8000/" + check_open + end + + it "can close the connection" do + open_socket "ws://localhost:8000/" + close_socket + check_closed + end + + describe "in the OPEN state" do + before { open_socket "ws://localhost:8000/" } + + it "can send and receive messages" do + listen_for_message + send_message + check_response + end + end + + describe "in the CLOSED state" do + before do + open_socket "ws://localhost:8000/" + close_socket + end + + it "cannot send and receive messages" do + listen_for_message + send_message + check_no_response + end + end +end + diff --git a/spec/faye/websocket/draft75_parser_spec.rb b/spec/faye/websocket/draft75_parser_spec.rb new file mode 100644 index 0000000..f0bbdd2 --- /dev/null +++ b/spec/faye/websocket/draft75_parser_spec.rb @@ -0,0 +1,41 @@ +# encoding=utf-8 + +require "spec_helper" + +describe Faye::WebSocket::Draft75Parser do + include EncodingHelper + + before do + @web_socket = mock Faye::WebSocket + @parser = Faye::WebSocket::Draft75Parser.new(@web_socket) + end + + describe :parse do + it "parses text frames" do + @web_socket.should_receive(:receive).with("Hello") + parse [0x00, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0xff] + end + + it "parses multibyte text frames" do + @web_socket.should_receive(:receive).with(encode "Apple = ") + parse [0x00, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf, 0xff] + end + + it "parses fragmented frames" do + @web_socket.should_receive(:receive).with("Hello") + parse [0x00, 0x48, 0x65, 0x6c] + parse [0x6c, 0x6f, 0xff] + end + end + + describe :frame do + it "returns the given string formatted as a WebSocket frame" do + bytes(@parser.frame "Hello").should == [0x00, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0xff] + end + + it "encodes multibyte characters correctly" do + message = encode "Apple = " + bytes(@parser.frame message).should == [0x00, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf, 0xff] + end + end +end diff --git a/spec/faye/websocket/protocol8_parser_spec.rb b/spec/faye/websocket/protocol8_parser_spec.rb new file mode 100644 index 0000000..2cf5352 --- /dev/null +++ b/spec/faye/websocket/protocol8_parser_spec.rb @@ -0,0 +1,145 @@ +# encoding=utf-8 + +require "spec_helper" + +describe Faye::WebSocket::Protocol8Parser do + include EncodingHelper + + before do + @web_socket = mock Faye::WebSocket + @parser = Faye::WebSocket::Protocol8Parser.new(@web_socket) + end + + describe :parse do + let(:mask) { (1..4).map { rand 255 } } + + def mask_message(*bytes) + output = [] + bytes.each_with_index do |byte, i| + output[i] = byte ^ mask[i % 4] + end + output + end + + it "parses unmasked text frames" do + @web_socket.should_receive(:receive).with("Hello") + parse [0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f] + end + + it "parses empty text frames" do + @web_socket.should_receive(:receive).with("") + parse [0x81, 0x00] + end + + it "parses fragmented text frames" do + @web_socket.should_receive(:receive).with("Hello") + parse [0x01, 0x03, 0x48, 0x65, 0x6c] + parse [0x80, 0x02, 0x6c, 0x6f] + end + + it "parses masked text frames" do + @web_socket.should_receive(:receive).with("Hello") + parse [0x81, 0x85] + mask + mask_message(0x48, 0x65, 0x6c, 0x6c, 0x6f) + end + + it "parses masked empty text frames" do + @web_socket.should_receive(:receive).with("") + parse [0x81, 0x80] + mask + mask_message() + end + + it "parses masked fragmented text frames" do + @web_socket.should_receive(:receive).with("Hello") + parse [0x01, 0x81] + mask + mask_message(0x48) + parse [0x80, 0x84] + mask + mask_message(0x65, 0x6c, 0x6c, 0x6f) + end + + it "closes the socket if the frame has an unrecognized opcode" do + @web_socket.should_receive(:close).with(1002, nil, false) + parse [0x83, 0x00] + end + + it "closes the socket if a close frame is received" do + @web_socket.should_receive(:close).with(1000, "Hello", false) + parse [0x88, 0x07, 0x03, 0xe8, 0x48, 0x65, 0x6c, 0x6c, 0x6f] + end + + it "parses unmasked multibyte text frames" do + @web_socket.should_receive(:receive).with(encode "Apple = ") + parse [0x81, 0x0b, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf] + end + + it "parses fragmented multibyte text frames" do + @web_socket.should_receive(:receive).with(encode "Apple = ") + parse [0x01, 0x0a, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3] + parse [0x80, 0x01, 0xbf] + end + + it "parses masked multibyte text frames" do + @web_socket.should_receive(:receive).with(encode "Apple = ") + parse [0x81, 0x8b] + mask + mask_message(0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf) + end + + it "parses masked fragmented multibyte text frames" do + @web_socket.should_receive(:receive).with(encode "Apple = ") + parse [0x01, 0x8a] + mask + mask_message(0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3) + parse [0x80, 0x81] + mask + mask_message(0xbf) + end + + it "parses unmasked medium-length text frames" do + @web_socket.should_receive(:receive).with("Hello" * 40) + parse [0x81, 0x7e, 0x00, 0xc8] + [0x48, 0x65, 0x6c, 0x6c, 0x6f] * 40 + end + + it "parses masked medium-length text frames" do + @web_socket.should_receive(:receive).with("Hello" * 40) + parse [0x81, 0xfe, 0x00, 0xc8] + mask + mask_message(*([0x48, 0x65, 0x6c, 0x6c, 0x6f] * 40)) + end + + it "replies to pings with a pong" do + @web_socket.should_receive(:send).with([0x4f, 0x48, 0x41, 0x49], :pong) + parse [0x89, 0x04, 0x4f, 0x48, 0x41, 0x49] + end + end + + describe :frame do + it "returns the given string formatted as a WebSocket frame" do + bytes(@parser.frame "Hello").should == [0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f] + end + + it "encodes multibyte characters correctly" do + message = encode "Apple = " + bytes(@parser.frame message).should == [0x81, 0x0b, 0x41, 0x70, 0x70, 0x6c, 0x65, 0x20, 0x3d, 0x20, 0xef, 0xa3, 0xbf] + end + + it "encodes medium-length strings using extra length bytes" do + message = "Hello" * 40 + bytes(@parser.frame message).should == [0x81, 0x7e, 0x00, 0xc8] + [0x48, 0x65, 0x6c, 0x6c, 0x6f] * 40 + end + + it "encodes long strings using extra length bytes" do + message = "Hello" * 13108 + bytes(@parser.frame message).should == [0x81, 0x7f] + + [0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x04] + + [0x48, 0x65, 0x6c, 0x6c, 0x6f] * 13108 + end + + it "encodes close frames with an error code" do + frame = @parser.frame "Hello", :close, 1002 + bytes(frame).should == [0x88, 0x07, 0x03, 0xea, 0x48, 0x65, 0x6c, 0x6c, 0x6f] + end + + it "encodes pong frames" do + bytes(@parser.frame '', :pong).should == [0x8a, 0x00] + end + end + + describe :utf8 do + it "detects valid UTF-8" do + Faye::WebSocket.valid_utf8?( [72, 101, 108, 108, 111, 45, 194, 181, 64, 195, 159, 195, 182, 195, 164, 195, 188, 195, 160, 195, 161, 45, 85, 84, 70, 45, 56, 33, 33] ).should be_true + end + + it "detects invalid UTF-8" do + Faye::WebSocket.valid_utf8?( [206, 186, 225, 189, 185, 207, 131, 206, 188, 206, 181, 237, 160, 128, 101, 100, 105, 116, 101, 100] ).should be_false + end + end +end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb new file mode 100644 index 0000000..848bec6 --- /dev/null +++ b/spec/spec_helper.rb @@ -0,0 +1,43 @@ +require 'rubygems' +require 'bundler/setup' +require File.expand_path('../../lib/faye/websocket', __FILE__) +require File.expand_path('../../vendor/em-rspec/lib/em-rspec', __FILE__) + +Thin::Logging.silent = true + +module EncodingHelper + def encode(message) + message.respond_to?(:force_encoding) ? + message.force_encoding("UTF-8") : + message + end + + def bytes(string) + string.bytes.to_a + end + + def parse(bytes) + @parser.parse(bytes.pack('C*')) + end +end + +class EchoServer + def call(env) + socket = Faye::WebSocket.new(env) + socket.onmessage = lambda do |event| + socket.send(event.data) + end + [-1, {}, []] + end + + def listen(port) + Rack::Handler.get('thin').run(self, :Port => port) do |server| + @server = server + end + end + + def stop + @server.stop + end +end + diff --git a/vendor/em-rspec b/vendor/em-rspec new file mode 160000 index 0000000..d2b6d33 --- /dev/null +++ b/vendor/em-rspec @@ -0,0 +1 @@ +Subproject commit d2b6d33f6f9c1a6addf2372b0c2b9983ae78d3c6