Connection to private channels

examples/private.rb

@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+require './lib/em/pusher/client'
+
+pusher_key = 'pusher_key'
+channel = 'private-channel'
+pusher_secret = 'pusher_secret'
+
+EM.run do
+  opts = {
+    key: pusher_key,
+    cluster: 'us2',
+    port: 80,
+    scheme: 'ws',
+  }
+  EM::Pusher::Client.connect(opts) do |conn|
+    conn.connected do
+      puts 'connected'
+    end
+
+    conn.connection_established do |socket_id|
+      puts 'callback'
+      auth = EM::Pusher::Client.build_auth(pusher_secret, pusher_key, socket_id, channel)
+      msg = {
+        event: 'pusher:subscribe',
+        data: {
+          channel: channel,
+          auth: auth,
+        },
+      }
+      conn.send_msg(msg)
+    end
+
+    conn.errback do |e|
+      puts "Got error: #{e}"
+    end
+
+    conn.stream do |msg|
+      case msg.event
+      when 'pusher:connection_established'
+        puts 'Connection Established'
+      when 'pusher_internal:subscription_succeeded'
+        puts "Subscribed to #{msg.json['channel']}"
+      when 'someevent'
+        puts "#{msg.type}: #{msg.event} >> #{msg.data}"
+      else
+        puts "unexpected event (#{msg.type}): <#{msg}>"
+      end
+      # conn.close_connection if closed?
+    end
+
+    conn.disconnect do
+      puts 'gone'
+      EM.stop_event_loop
+    end
+  end
+end

lib/em/pusher/client.rb

@@ -27,6 +27,17 @@ def self.connect(options)
         end
       end
 
+      def self.sign(secret, socket_id, channel_name)
+        digest = OpenSSL::Digest::SHA256.new
+        string_to_sign = "#{socket_id}:#{channel_name}"
+        OpenSSL::HMAC.hexdigest(digest, secret, string_to_sign)
+      end
+
+      def self.build_auth(secret, key, socket_id, channel_name)
+        sig = sign(secret, socket_id, channel_name)
+        "#{key}:#{sig}"
+      end
+
       def self.url(options)
         opts = DEFAULT_OPTIONS.merge(options)
         REQUIRED_OPTIONS.each { |opt| fail ArgumentError, "option #{opt} is required" unless opts[opt] }

lib/em/pusher/client/connection.rb

@@ -4,6 +4,7 @@
 require 'uri'
 require 'json'
 require 'websocket'
+require 'openssl'
 
 module EM
   module Pusher
@@ -53,6 +54,14 @@ def disconnect(&cback)
           @disconnect = cback
         end
 
+        def connection_established(&cback)
+          @connection_established = cback
+        end
+
+        def ping(&cback)
+          @ping = cback
+        end
+
         # https://pusher.com/docs/pusher_protocol#subscription-events
         def subscribe(channel, auth = nil, channel_data = nil)
           msg = {
@@ -98,8 +107,34 @@ def unbind
         def handle_received_data(data)
           @frame << data
           while (msg = @frame.next)
-            @stream.call(EM::Pusher::Client::MsgParser.new(msg)) if @stream
+            process_data(EM::Pusher::Client::MsgParser.new(msg))
+          end
+        end
+
+        def process_data(parser)
+          if parser.event == 'pusher:connection_established'
+            process_connection(parser)
+          elsif parser.type == :ping
+            process_ping
+          elsif parser.type == :close
+            process_close
           end
+          @stream.call(parser) if @stream
+        end
+
+        def process_connection(parser)
+          socket_id = JSON.parse(parser.json['data'])['socket_id']
+          @connection_established.yield(socket_id) if @connection_established
+        end
+
+        def process_ping
+          @ping.yield if @ping
+          send_msg(event: 'pusher:pong')
+        end
+
+        def process_close
+          unbind
+          close_connection
         end
       end
     end

lib/em/pusher/client/msg_parser.rb

@@ -4,35 +4,38 @@ module EM
   module Pusher
     module Client
       class MsgParser
-        attr_reader :json,
-                    :data
-
         def initialize(msg)
-          parse(msg)
+          @msg = msg
         end
 
         def event
+          return '{}' unless json
           @event ||= json['event']
         end
 
         def to_s
           @msg.to_s
         end
 
-      private
+        def type
+          @msg.type
+        end
+
+        def json
+          @json ||= JSON.parse(@msg.data)
+        rescue JSON::ParserError
+          {}
+        end
 
-        def parse(msg)
-          @msg = msg
-          @json = JSON.parse(msg.data)
+        def data
           @data =
             if json['data'].is_a?(String)
               JSON.parse(json['data'])
             else
               json['data']
             end
-        rescue JSON::ParserError => e
-          puts "Error parsing event '#{event}': '#{e.message}'"
-          # logger.error("Error parsing msg #{e}")
+        rescue JSON::ParserError
+          {}
         end
       end
     end

spec/em/pusher/client/msg_parser_spec.rb

@@ -51,7 +51,7 @@ def build_msg(msg)
       let!(:parser) { described_class.new(build_msg(malformed)) }
 
       it { expect(parser.event).to eq('pusher:someevent') }
-      it { expect(parser.data).to be_nil }
+      it { expect(parser.data).to eq({}) }
       it { expect(parser.to_s).to eq(malformed) }
     end
   end