Rough STOMP Adapter

.travis.yml

@@ -1,16 +1,37 @@
 language: ruby
 
+bundler_args: --with-development-dependencies --jobs=3 --retry=3 --verbose
+
+cache:
+  bundler: true
+
+matrix:
+  fast_finish: true
+
 rvm:
   - 1.9.3
   - 2.0.0
-  - 2.1.0
+  - 2.1.8
+  - 2.2.4
 
 gemfile:
   - gemfiles/Gemfile.rails-3.x
   - gemfiles/Gemfile.rails-4.x
 
+# Set up for STOMP tests
+services:
+  - rabbitmq
+
+before_install:
+  - gem update bundler
+
+install:
+  - sudo rabbitmq-plugins enable rabbitmq_web_stomp
+  - sudo service rabbitmq-server restart
+
 # Set up and start Faye Server before tests are run
 before_script:
+  - bundle install
   - cp test/travis/sync.yml config/sync.yml
   - cp test/travis/sync.ru sync.ru
   - thin start -R sync.ru -p 9292 -t 1 2>&1 > test.log &

README.md

@@ -59,6 +59,13 @@ gem 'pusher'
 gem 'sync'
 ```
 
+#### Using STOMP
+
+```ruby
+gem 'stomp'
+gem 'sync'
+```
+
 #### Install
 
 ```bash
@@ -78,10 +85,15 @@ $ rails g sync:install
 <%= include_sync_config %>
 ```
 
-#### 4) Configure your pubsub server (Faye or Pusher)
+> NOTE: If you have multiple adapter_javascript_url that you need to include, such as in the STOMP example config below, add the following after:
+
+```erb
+<%= javascript_include_tag *Sync.adapter_javascript_url %>
+```
 
+#### 4) Configure your pubsub server (Faye, Pusher, or STOMP)
 
-#### Using [Faye](http://faye.jcoglan.com/) (self hosted)
+#### Using [Faye](http://faye.jcoglan.com/) (self-hosted)
 
 Set your configuration in the generated `config/sync.yml` file, using the Faye adapter. Then run Faye alongside your app.
 
@@ -93,6 +105,34 @@ rackup sync.ru -E production
 
 Set your configuration in the generated `config/sync.yml` file, using the Pusher adapter. No extra process/setup.
 
+#### Using [STOMP](https://stomp.github.io) with [RabbitMQ](https://www.rabbitmq.com) (self-hosted)
+
+> NOTE: Sync will probably play nicely with other STOMP servers besides RabbitMQ, but these have not been tested and are not guaranteed to be supported.
+
+- Install [RabbitMQ](https://www.rabbitmq.com/download.html).
+- Start the RabbitMQ server (see your installation guide in link above).
+- Enable the RabbitMQ STOMP plugin: `rabbitmq-plugins enable rabbitmq_web_stomp` ([see page](https://www.rabbitmq.com/stomp.html))
+- Set your configuration in the generated `config/sync.yml` file, using the Stomp adapter. Ensure the default `amq.topic` exchange is available on your RabbitMQ instance (although experienced users may adjust the `websocket` and `destination` endpoints to suit their needs).
+- The Stomp adapter needs access to vendor javascript assets for [`sockjs-client`](https://github.com/sockjs/sockjs-client) and [`stomp-websocket`](https://github.com/jmesnil/stomp-websocket). Either add them to your asset pipeline, and require them in your `application.js` manifest *before* the `//= require sync` declaration, or handle their inclusion in your layout as laid out in [Step #3](#3-add-the-pubsub-adapters-javascript-to-your-application-layout-appviewslayoutsapplicationhtmlerb).
+
+An example configuration for RabbitMQ servers (local, out-of-box, and STOMP-enabled):
+```yaml
+# config/sync.yml
+
+development:
+  adapter_javascript_url:
+    - 'https://cdnjs.cloudflare.com/ajax/libs/sockjs-client/1.0.3/sockjs.js'
+    - 'https://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.js'
+  server: 'stomp://guest:guest@localhost:61613'
+  api_key: 'guest'
+  auth_token: 'guest'
+  adapter: 'Stomp'
+  encryption: false
+  async: true
+  websocket: 'http://localhost:15674/stomp'
+  destination: '/topic/sync-'
+```
+
 ## Current Caveats
 The current implementation uses a DOM range query (jQuery's `nextUntil`) to match your partial's "element" in
 the DOM. The way this selector works requires your sync'd partial to be wrapped in a root level html tag for that partial file.

app/assets/javascripts/sync.coffee

@@ -6,14 +6,16 @@ $ = jQuery
   readyQueue: []
 
   init: ->
+    return unless SyncConfig? && Sync[SyncConfig.adapter]
+    @adapter ||= new Sync[SyncConfig.adapter]
+
     $ =>
-      return unless SyncConfig? && Sync[SyncConfig.adapter]
-      @adapter ||= new Sync[SyncConfig.adapter]
-      return if @isReady() || [email protected]()
-      @ready = true
-      @connect()
-      @flushReadyQueue()
-      @bindUnsubscribe()
+      onReadyInterval = setInterval (=>
+        return if @isReady() || [email protected]()
+        clearInterval(onReadyInterval)
+        @ready = true
+        @connect()
+      ), 250
 
 
   # Handle Turbolinks teardown, unsubscribe from all channels before transition
@@ -30,15 +32,29 @@ $ = jQuery
 
   onConnectFailure: (error) -> #noop
 
-  connect: -> @adapter.connect()
+  connect: ->
+    @adapter.connect()
+
+    onConnectInterval = setInterval (=>
+      return unless @isConnected()
+      clearInterval(onConnectInterval)
+      @adapter.onConnect()
+    ), 250
 
   isConnected: -> @adapter.isConnected()
 
-  onReady: (callback) ->
-    if @isReady()
-      callback()
-    else
-      @readyQueue.push callback
+  onDebug: (message) ->
+    return unless SyncConfig.debug_flag
+
+    window?.console?.log(message)
+
+  onReady: (callbacks) ->
+    callbackOrAddToQueue = (callback) =>
+      return callback() if @isReady()
+      @readyQueue.push(callback)
+
+    return callbackOrAddToQueue(callbacks) unless callbacks.constructor == Array
+    callbackOrAddToQueue(callback) for callback in callbacks
 
 
   flushReadyQueue: ->
@@ -89,14 +105,26 @@ class Sync.Adapter
       return
 
   subscribe: (channel, callback) ->
-    @unsubscribeChannel(channel)
-    subscription = new Sync[SyncConfig.adapter].Subscription(@client, channel, callback)
-    @subscriptions.push(subscription)
-    subscription
+    onReadyCallback = (channel, callback) =>
+      @unsubscribeChannel(channel)
+      subscription = new Sync[SyncConfig.adapter].Subscription(@client, channel, callback)
+      @subscriptions.push(subscription)
+      subscription
+
+    if @isConnected()
+      onReadyCallback(channel, callback)
+    else
+      Sync.onReady(onReadyCallback)
 
+  onConnect: ->
+    Sync.flushReadyQueue()
+    Sync.bindUnsubscribe()
+
+  onError: (error) ->
+    window?.console?.log({ error: error })
 
-class Sync.Faye extends Sync.Adapter
 
+class Sync.Faye extends Sync.Adapter
   subscriptions: []
 
   available: ->
@@ -126,8 +154,8 @@ class Sync.Pusher extends Sync.Adapter
     !!window.Pusher
 
   connect: ->
-    opts =
-      encrypted: SyncConfig.pusher_encrypted
+    # Pusher doesn't properly transform native boolean flags in options.
+    opts = { encrypted: SyncConfig.encryption_flag.toString() }
 
     opts.wsHost = SyncConfig.pusher_ws_host if SyncConfig.pusher_ws_host
     opts.wsPort = SyncConfig.pusher_ws_port if SyncConfig.pusher_ws_port
@@ -137,12 +165,44 @@ class Sync.Pusher extends Sync.Adapter
 
   isConnected: -> @client?.connection.state is "connected"
 
-  subscribe: (channel, callback) ->
-    @unsubscribeChannel(channel)
-    subscription = new Sync.Pusher.Subscription(@client, channel, callback)
-    @subscriptions.push(subscription)
-    subscription
+class Sync.Stomp extends Sync.Adapter
+
+  subscriptions: []
+
+  client: null
+  socket: null
 
+  headers: {
+    login: SyncConfig.api_key,
+    passcode: SyncConfig.auth_token
+  }
+
+  available: ->
+    !!window.Stomp
+
+  connect: ->
+    @socket = new window.SockJS(SyncConfig.websocket)
+    @client = window.Stomp.over(@socket)
+
+    @client.debug = Sync.onDebug
+
+    # SockJS does not support heart-beat: disable heart-beats
+    @client.heartbeat.outgoing = 0
+    @client.heartbeat.incoming = 0
+
+    @client.connect(
+      @headers['login'],
+      @headers['passcode'],
+      @onConnect,
+      @onError,
+      '/'
+    )
+
+  isConnected: ->
+    try
+      @client.connected
+    catch error
+      false
 
 class Sync.Pusher.Subscription
   constructor: (@client, channel, callback) ->
@@ -154,6 +214,18 @@ class Sync.Pusher.Subscription
   cancel: ->
     @client.unsubscribe(@channel) if @client.channel(@channel)?
 
+class Sync.Stomp.Subscription
+  constructor: (client, channel, callback) ->
+    @channel = channel
+    @client = client
+
+    @subscription = @client.subscribe channel, ( (e) ->
+      callback(JSON.parse(e.body))
+    )
+
+  cancel: ->
+    @subscription.unsubscribe()
+
 
 class Sync.View
 

lib/generators/sync/templates/sync.yml

@@ -15,6 +15,20 @@ development:
 #   adapter: "Pusher"
 #   async: true
 
+# Stomp
+# development:
+#   adapter_javascript_url:
+#     - 'https://cdnjs.cloudflare.com/ajax/libs/sockjs-client/1.0.3/sockjs.js'
+#     - 'https://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.js'
+#   server: 'stomp://YOUR_STOMP_SERVER_LOGIN:YOUR_STOMP_SERVER_PASSCODE@HOST:STOMP_PORT'
+#   api_key: 'YOUR_STOMP_CLIENT_LOGIN'
+#   auth_token: 'YOUR_STOMP_CLIENT_PASSCODE'
+#   adapter: 'Stomp'
+#   encryption: false
+#   async: true
+#   websocket: 'http://HOST:WEBSOCKET_PORT/stomp'
+#   destination: '/topic/sync'
+
 # Disabled
 # development:
 #   adapter: false

lib/sync.rb

@@ -27,6 +27,7 @@
 require 'sync/resource'
 require 'sync/clients/faye'
 require 'sync/clients/pusher'
+require 'sync/clients/stomp'
 require 'sync/clients/dummy'
 require 'sync/reactor'
 if defined? Rails
@@ -46,13 +47,16 @@ def config
     def config_json
       @config_json ||= begin
         {
-          server: server,
+          adapter: adapter,
           api_key: api_key,
+          auth_token: auth_token,
+          debug_flag: debug_flag,
+          encryption_flag: encryption_flag,
           pusher_ws_host: pusher_ws_host,
           pusher_ws_port: pusher_ws_port,
           pusher_wss_port: pusher_wss_port,
-          pusher_encrypted: pusher_encrypted,
-          adapter: adapter
+          server: server,
+          websocket: websocket
         }.reject { |k, v| v.nil? }.to_json
       end
     end
@@ -101,6 +105,16 @@ def server
       config[:server]
     end
 
+    def websocket
+      config[:websocket]
+    end
+
+    def destination
+      config[:destination] ||= '/'
+      return "/#{config[:destination]}" unless config[:destination][0] == '/'
+      config[:destination]
+    end
+
     def adapter_javascript_url
       config[:adapter_javascript_url]
     end
@@ -145,12 +159,12 @@ def pusher_wss_port
       config[:pusher_wss_port]
     end
 
-    def pusher_encrypted
-      if config[:pusher_encrypted].nil?
-        true
-      else
-        config[:pusher_encrypted]
-      end
+    def encryption_flag
+      config[:encryption]
+    end
+
+    def debug_flag
+      !!config[:debug]
     end
 
     def reactor