Connection cleanup

README.rdoc

@@ -46,35 +46,6 @@ The application will pause at socket.connect and handle events from Pusher as th
 
   socket.connect
 
-== Asynchronous Usage
-The socket will remain open in the background as long as your main application thread is running,
-and you can continue to subscribe/unsubscribe to channels and bind new events.
-
-  require 'pusher-client'
-  socket = PusherClient::Socket.new(YOUR_APPLICATION_KEY)
-  socket.connect(true) # Connect asynchronously
-
-  # Subscribe to two channels
-  socket.subscribe('channel1')
-  socket.subscribe('channel2')
-
-  # Bind to a global event (can occur on either channel1 or channel2)
-  socket.bind('globalevent') do |data|
-    puts data
-  end
-
-  # Bind to a channel event (can only occur on channel1)
-  socket['channel1'].bind('channelevent') do |data|
-    puts data
-  end
-
-  loop do
-    sleep(1) # Keep your main thread running
-  end
-
-For further documentation, read the source & test suite. Some features of the JavaScript client
-are not yet implemented.
-
 == Native extension
 Pusher depends on the websocket[https://github.com/imanel/websocket-ruby] gem
 which is a pure Ruby implementation of websockets.

lib/pusher-client/channel.rb

@@ -14,7 +14,7 @@ def initialize(channel_name, user_data=nil, logger=PusherClient.logger)
     end
 
     def bind(event_name, &callback)
-      PusherClient.logger.debug "Binding #{event_name} to #{name}"
+      logger.debug "Binding #{event_name} to #{name}"
       @callbacks[event_name] = callbacks[event_name] || []
       @callbacks[event_name] << callback
       return self

lib/pusher-client/socket.rb

@@ -6,7 +6,7 @@ module PusherClient
   class Socket
 
     CLIENT_ID = 'pusher-ruby-client'
-    PROTOCOL = '6'
+    PROTOCOL = '7'
 
     attr_reader :path, :connected, :channels, :global_channel, :socket_id
 
@@ -45,8 +45,9 @@ def initialize(app_key, options={})
       end
 
       bind('pusher:connection_disconnected') do |data|
+        @connection = nil
         @connected = false
-        @channels.channels.each { |c| c.disconnect }
+        @socket_id = nil
       end
 
       bind('pusher:error') do |data|
@@ -61,33 +62,25 @@ def initialize(app_key, options={})
     end
 
     def connect(async = false)
+      raise "Async is not supported anymore" if async
       return if @connection
       logger.debug("Pusher : connecting : #{@url}")
 
-      if async
-        @connection_thread = Thread.new do
-          begin
-            connect_internal
-          rescue => ex
-            send_local_event "pusher:error", ex
-          end
-        end
-      else
-        connect_internal
-      end
+      connect_internal
       self
     end
 
-    def disconnect
+    def disconnect(ex = nil)
       return unless @connection
       logger.debug("Pusher : disconnecting")
-      @connected = false
-      @connection.close
-      @connection = nil
-      if @connection_thread
-        @connection_thread.kill
-        @connection_thread = nil
-      end
+
+      @connection.close rescue nil
+
+      send_local_event("pusher:connection_disconnected", ex)
+    end
+
+    def closed?
+      @connection && @connection.closed?
     end
 
     def subscribe(channel_name, user_data = nil)
@@ -100,15 +93,15 @@ def subscribe(channel_name, user_data = nil)
       end
 
       channel = @channels.add(channel_name, user_data)
-      if @connected
+      if connected
         authorize(channel, method(:authorize_callback))
       end
       return channel
     end
 
     def unsubscribe(channel_name)
       channel = @channels.remove channel_name
-      if channel && @connected
+      if channel && connected
         send_event('pusher:unsubscribe', {
           'channel' => channel_name
         })
@@ -206,12 +199,15 @@ def connect_internal
         @connection.receive.each do |msg|
           params = parser(msg)
 
-          # why ?
+          # ignore messages to self
           next if params['socket_id'] && params['socket_id'] == self.socket_id
 
           send_local_event(params['event'], params['data'], params['channel'])
         end
       end
+    rescue IOError, Errno::EBADF, SocketError => ex
+      disconnect(ex)
+      return ex
     end
 
     def send_local_event(event_name, event_data, channel_name=nil)
@@ -231,13 +227,13 @@ def parser(data)
       return JSON.parse(data)
     rescue => err
       logger.warn(err)
-      logger.warn("Pusher : data attribute not valid JSON - you may wish to implement your own Pusher::Client.parser")
+      logger.warn("Pusher : data attribute not valid JSON - you may wish to implement your own PusherClient::Socket.parser")
       return data
     end
 
     def hmac(secret, string_to_sign)
       digest = OpenSSL::Digest::SHA256.new
-      signature = OpenSSL::HMAC.hexdigest(digest, secret, string_to_sign)
+      OpenSSL::HMAC.hexdigest(digest, secret, string_to_sign)
     end
   end
 

lib/pusher-client/websocket.rb

@@ -12,8 +12,8 @@ class PusherWebSocket
     attr_accessor :socket
 
     def initialize(url, params = {})
-      @hs ||= WebSocket::Handshake::Client.new(:url => url)
-      @frame ||= WebSocket::Frame::Incoming::Server.new(:version => @hs.version)
+      @hs = WebSocket::Handshake::Client.new(:url => url)
+      @frame = WebSocket::Frame::Incoming::Client.new(:version => @hs.version)
       @socket = TCPSocket.new(@hs.host, @hs.port || 80)
       @cert_file = params[:cert_file]
       @logger = params[:logger] || PusherClient.logger
@@ -45,9 +45,13 @@ def initialize(url, params = {})
         @hs << data
 
         if @hs.finished?
-          raise @hs.error.to_s unless @hs.valid?
-          @handshaked = true
-          break
+          if @hs.valid?
+            @handshaked = true
+            break
+          else
+            close
+            raise @hs.error.to_s
+          end
         end
       end
     end
@@ -92,6 +96,10 @@ def close
       logger.debug error.message
     end
 
+    def closed?
+      @socket.closed?
+    end
+
     private
 
     attr_reader :logger