Merge pull request #2 from basecamp/create-channel

Channel factory

9 files changed, 176 insertions, 163 deletions

diff --git a/lib/action_cable/channel/base.rb b/lib/action_cable/channel/base.rb
index 83ba2cb3d2..6c55a8ed65 100644
--- a/lib/action_cable/channel/base.rb
+++ b/lib/action_cable/channel/base.rb
@@ -10,16 +10,10 @@ module ActionCable
       attr_reader :params, :connection
       delegate :logger, to: :connection
 
-      class_attribute :channel_name
-
       class << self
         def matches?(identifier)
           raise "Please implement #{name}#matches? method"
         end
-
-        def find_name
-          @name ||= channel_name || to_s.demodulize.underscore
-        end
       end
 
       def initialize(connection, channel_identifier, params = {})
@@ -138,4 +132,4 @@ module ActionCable
         end
     end
   end
-end
\ No newline at end of file
+end
diff --git a/lib/action_cable/connection/subscriptions.rb b/lib/action_cable/connection/subscriptions.rb
index 9e7a8a5f73..ae191c7795 100644
--- a/lib/action_cable/connection/subscriptions.rb
+++ b/lib/action_cable/connection/subscriptions.rb
@@ -23,7 +23,7 @@ module ActionCable
         id_options = ActiveSupport::JSON.decode(id_key).with_indifferent_access
 
         subscription_klass = connection.server.registered_channels.detect do |channel_klass|
-          channel_klass.find_name == id_options[:channel]
+          channel_klass == id_options[:channel].safe_constantize
         end
 
         if subscription_klass
@@ -66,4 +66,4 @@ module ActionCable
         end
     end
   end
-end
\ No newline at end of file
+end
diff --git a/lib/assets/javascripts/cable.js.coffee b/lib/assets/javascripts/cable.js.coffee
index 7c033d3b08..0bd1757505 100644
--- a/lib/assets/javascripts/cable.js.coffee
+++ b/lib/assets/javascripts/cable.js.coffee
@@ -1,124 +1,8 @@
 #= require_self
-#= require_tree .
+#= require cable/consumer
 
-class @Cable
-  MAX_CONNECTION_INTERVAL: 5 * 1000
-  PING_STALE_INTERVAL: 8
+@Cable =
+  PING_IDENTIFIER: "_ping"
 
-  constructor: (@cableUrl) ->
-    @subscribers = {}
-    @resetPingTime()
-    @resetConnectionAttemptsCount()
-    @connect()
-
-  connect: ->
-    @connection = @createConnection()
-
-  createConnection: ->
-    connection = new WebSocket(@cableUrl)
-    connection.onmessage = @receiveData
-    connection.onopen    = @connected
-    connection.onclose   = @reconnect
-
-    connection.onerror   = @reconnect
-    connection
-
-  isConnected: =>
-    @connection?.readyState is 1
-
-  sendData: (identifier, data) =>
-    if @isConnected()
-      @connection.send JSON.stringify { command: 'message', identifier: identifier, data: data }
-
-  receiveData: (message) =>
-    data = JSON.parse message.data
-
-    if data.identifier is '_ping'
-      @pingReceived(data.message)
-    else
-      @subscribers[data.identifier]?.onReceiveData(data.message)
-
-  connected: =>
-    @startWaitingForPing()
-    @resetConnectionAttemptsCount()
-
-    for identifier, callbacks of @subscribers
-      @subscribeOnServer(identifier)
-      callbacks['onConnect']?()
-
-  reconnect: =>
-    @removeExistingConnection()
-
-    @resetPingTime()
-    @disconnected()
-
-    setTimeout =>
-      @incrementConnectionAttemptsCount()
-      @connect()
-    , @generateReconnectInterval()
-
-  removeExistingConnection: =>
-    if @connection?
-      @clearPingWaitTimeout()
-
-      @connection.onclose = -> # no-op
-      @connection.onerror = -> # no-op
-      @connection.close()
-      @connection = null
-
-  resetConnectionAttemptsCount: =>
-    @connectionAttempts = 1
-
-  incrementConnectionAttemptsCount: =>
-    @connectionAttempts += 1
-
-  generateReconnectInterval: () ->
-    interval = (Math.pow(2, @connectionAttempts) - 1) * 1000
-    if interval > @MAX_CONNECTION_INTERVAL then @MAX_CONNECTION_INTERVAL else interval
-
-  startWaitingForPing: =>
-    @clearPingWaitTimeout()
-
-    @waitForPingTimeout = setTimeout =>
-      console.log "Ping took too long to arrive. Reconnecting.."
-      @reconnect()
-    , @PING_STALE_INTERVAL * 1000
-
-  clearPingWaitTimeout: =>
-    clearTimeout(@waitForPingTimeout)
-
-  resetPingTime: =>
-    @lastPingTime = null
-
-  disconnected: =>
-    callbacks['onDisconnect']?() for identifier, callbacks of @subscribers
-
-  giveUp: =>
-    # Show an error message
-
-  subscribe: (identifier, callbacks) =>
-    @subscribers[identifier] = callbacks
-
-    if @isConnected()
-      @subscribeOnServer(identifier)
-      @subscribers[identifier]['onConnect']?()
-
-  unsubscribe: (identifier) =>
-    @unsubscribeOnServer(identifier, 'unsubscribe')
-    delete @subscribers[identifier]
-
-  subscribeOnServer: (identifier) =>
-    if @isConnected()
-      @connection.send JSON.stringify { command: 'subscribe', identifier: identifier }
-
-  unsubscribeOnServer: (identifier) =>
-    if @isConnected()
-      @connection.send JSON.stringify { command: 'unsubscribe', identifier: identifier }
-
-  pingReceived: (timestamp) =>
-    if @lastPingTime? and (timestamp - @lastPingTime) > @PING_STALE_INTERVAL
-      console.log "Websocket connection is stale. Reconnecting.."
-      @reconnect()
-    else
-      @startWaitingForPing()
-      @lastPingTime = timestamp
+  createConsumer: (url) ->
+    new Cable.Consumer url
diff --git a/lib/assets/javascripts/cable/connection.js.coffee b/lib/assets/javascripts/cable/connection.js.coffee
new file mode 100644
index 0000000000..cd9539a6aa
--- /dev/null
+++ b/lib/assets/javascripts/cable/connection.js.coffee
@@ -0,0 +1,53 @@
+#= require cable/connection_monitor
+
+class Cable.Connection
+  constructor: (@consumer) ->
+    new Cable.ConnectionMonitor @consumer
+    @open()
+
+  send: (data) ->
+    if @isOpen()
+      @websocket.send(JSON.stringify(data))
+      true
+    else
+      false
+
+  open: ->
+    @websocket = new WebSocket(@consumer.url)
+    @websocket.onmessage = @onMessage
+    @websocket.onopen    = @onOpen
+    @websocket.onclose   = @onClose
+    @websocket.onerror   = @onError
+    @websocket
+
+  close: ->
+    @websocket.close() unless @isClosed()
+
+  reopen: ->
+    @close()
+    @open()
+
+  isOpen: ->
+    @websocket.readyState is WebSocket.OPEN
+
+  isClosed: ->
+    @websocket.readyState in [ WebSocket.CLOSED, WebSocket.CLOSING ]
+
+  onMessage: (message) =>
+    data = JSON.parse message.data
+    @consumer.subscribers.notify(data.identifier, "received", data.message)
+
+  onOpen: =>
+    @consumer.subscribers.reload()
+
+  onClose: =>
+    @disconnect()
+
+  onError: =>
+    @disconnect()
+    @websocket.onclose = -> # no-op
+    @websocket.onerror = -> # no-op
+    try @close()
+
+  disconnect: ->
+    @consumer.subscribers.notifyAll("disconnected")
diff --git a/lib/assets/javascripts/cable/connection_monitor.js.coffee b/lib/assets/javascripts/cable/connection_monitor.js.coffee
new file mode 100644
index 0000000000..bb4ee8f7f6
--- /dev/null
+++ b/lib/assets/javascripts/cable/connection_monitor.js.coffee
@@ -0,0 +1,41 @@
+class Cable.ConnectionMonitor
+  MAX_CONNECTION_INTERVAL: 5 * 1000
+  PING_STALE_INTERVAL: 8 * 1000
+
+  identifier: Cable.PING_IDENTIFIER
+
+  constructor: (@consumer) ->
+    @reset()
+    @consumer.subscribers.add(this)
+    @pollConnection()
+
+  connected: ->
+    @reset()
+    @pingedAt = now()
+
+  received: ->
+    @pingedAt = now()
+
+  reset: ->
+    @connectionAttempts = 1
+
+  pollConnection: ->
+    setTimeout =>
+      @reconnect() if @connectionIsStale()
+      @pollConnection()
+    , @getPollTimeout()
+
+  getPollTimeout: ->
+    interval = (Math.pow(2, @connectionAttempts) - 1) * 1000
+    if interval > @MAX_CONNECTION_INTERVAL then @MAX_CONNECTION_INTERVAL else interval
+
+  connectionIsStale: ->
+    @pingedAt? and (now() - @pingedAt) > @PING_STALE_INTERVAL
+
+  reconnect: ->
+    console.log "Ping took too long to arrive. Reconnecting.."
+    @connectionAttempts += 1
+    @consumer.connection.reopen()
+
+  now = ->
+    new Date().getTime()
diff --git a/lib/assets/javascripts/cable/consumer.js.coffee b/lib/assets/javascripts/cable/consumer.js.coffee
new file mode 100644
index 0000000000..a9abd6256a
--- /dev/null
+++ b/lib/assets/javascripts/cable/consumer.js.coffee
@@ -0,0 +1,16 @@
+#= require cable/connection
+#= require cable/subscription
+#= require cable/subscriber_manager
+
+class Cable.Consumer
+  constructor: (@url) ->
+    @subscribers = new Cable.SubscriberManager this
+    @connection = new Cable.Connection this
+
+  createSubscription: (channelName, mixin) ->
+    channel = channelName
+    params = if typeof channel is "object" then channel else {channel}
+    new Cable.Subscription this, params, mixin
+
+  send: (data) ->
+    @connection.send(data)
diff --git a/lib/assets/javascripts/cable/subscriber_manager.js.coffee b/lib/assets/javascripts/cable/subscriber_manager.js.coffee
new file mode 100644
index 0000000000..922c74808c
--- /dev/null
+++ b/lib/assets/javascripts/cable/subscriber_manager.js.coffee
@@ -0,0 +1,36 @@
+class Cable.SubscriberManager
+  constructor: (@consumer) ->
+    @subscribers = []
+
+  add: (subscriber) ->
+    @subscribers.push(subscriber)
+    @notify(subscriber, "initialized")
+    if @sendCommand(subscriber, "subscribe")
+      @notify(subscriber, "connected")
+
+  reload: ->
+    for subscriber in @subscribers
+      if @sendCommand(subscriber, "subscribe")
+        @notify(subscriber, "connected")
+
+  remove: (subscriber) ->
+    @sendCommand(subscriber, "unsubscribe")
+    @subscribers = (s for s in @subscribers when s isnt subscriber)
+
+  notifyAll: (callbackName, args...) ->
+    for subscriber in @subscribers
+      @notify(subscriber, callbackName, args...)
+
+  notify: (subscriber, callbackName, args...) ->
+    if typeof subscriber is "string"
+      subscribers = (s for s in @subscribers when s.identifier is subscriber)
+    else
+      subscribers = [subscriber]
+
+    for subscriber in subscribers
+      subscriber[callbackName]?(args...)
+
+  sendCommand: (subscriber, command) ->
+    {identifier} = subscriber
+    return true if identifier is Cable.PING_IDENTIFIER
+    @consumer.send({command, identifier})
diff --git a/lib/assets/javascripts/cable/subscription.js.coffee b/lib/assets/javascripts/cable/subscription.js.coffee
new file mode 100644
index 0000000000..74cc35a7a7
--- /dev/null
+++ b/lib/assets/javascripts/cable/subscription.js.coffee
@@ -0,0 +1,22 @@
+class Cable.Subscription
+  constructor: (@consumer, params = {}, mixin) ->
+    @identifier = JSON.stringify(params)
+    extend(this, mixin)
+    @consumer.subscribers.add(this)
+
+  # Perform a channel action with the optional data passed as an attribute
+  perform: (action, data = {}) ->
+    data.action = action
+    @send(data)
+
+  send: (data) ->
+    @consumer.send(command: "message", identifier: @identifier, data: JSON.stringify(data))
+
+  unsubscribe: ->
+    @consumer.subscribers.remove(this)
+
+  extend = (object, properties) ->
+    if properties?
+      for key, value of properties
+        object[key] = value
+    object
diff --git a/lib/assets/javascripts/channel.js.coffee b/lib/assets/javascripts/channel.js.coffee
deleted file mode 100644
index c972334140..0000000000
--- a/lib/assets/javascripts/channel.js.coffee
+++ /dev/null
@@ -1,33 +0,0 @@
-class @Cable.Channel
-  constructor: (params = {}) ->
-    {channelName} = @constructor
-
-    if channelName?
-      params['channel'] = channelName
-      @channelIdentifier = JSON.stringify params
-    else
-      throw new Error "This channel's constructor is missing the required 'channelName' property"
-
-    cable.subscribe(@channelIdentifier, {
-      onConnect: @connected
-      onDisconnect: @disconnected
-      onReceiveData: @received
-    })
-
-
-  connected: =>
-    # Override in the subclass
-
-  disconnected: =>
-    # Override in the subclass
-
-  received: (data) =>
-    # Override in the subclass
-
-  # Perform a channel action with the optional data passed as an attribute
-  perform: (action, data = {}) ->
-    data.action = action
-    cable.sendData @channelIdentifier, JSON.stringify data
-
-  send: (data) ->
-    cable.sendData @channelIdentifier, JSON.stringify data