Rough STOMP adapter + associated configuration/test changes.

Author: Adam Eberlin

.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 &

app/assets/javascripts/sync.coffee.erb

@@ -4,20 +4,26 @@ $ = jQuery
 
   ready: false
   readyQueue: []
-  FAYE_HOST: "<%= Sync.server %>"
-  PUSHER_API_KEY: "<%= Sync.api_key %>"
+
+  SERVER: "<%= Sync.server %>"
+  API_KEY: "<%= Sync.api_key %>"
+  AUTH_TOKEN: "<%= Sync.auth_token %>"
   CLIENT_ADAPTER: "<%= Sync.adapter %>"
-  ENCRYPTION_FLAG: "<%= Sync.encryption_flag %>"
+  ENCRYPTION_FLAG: <%= Sync.encryption_flag %>
+  WEBSOCKET: "<%= Sync.websocket %>"
+  DEBUG_FLAG: <%= Sync.debug_flag %>
 
   init: ->
     return unless Sync[@CLIENT_ADAPTER]
     @adapter = new Sync[@CLIENT_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
@@ -33,15 +39,27 @@ $ = 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) ->
+    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: ->
@@ -92,21 +110,30 @@ class Sync.Adapter
       return
 
   subscribe: (channel, callback) ->
-    @unsubscribeChannel(channel)
-    subscription = new Sync[Sync.CLIENT_ADAPTER].Subscription(@client, channel, callback)
-    @subscriptions.push(subscription)
-    subscription
+    onReadyCallback = (channel, callback) =>
+      @unsubscribeChannel(channel)
+      subscription = new Sync[Sync.CLIENT_ADAPTER].Subscription(@client, channel, callback)
+      @subscriptions.push(subscription)
+      subscription
+
+    if @isConnected()
+      onReadyCallback(channel, callback)
+    else
+      Sync.onReady(onReadyCallback)
 
+  onConnect: ->
+    Sync.flushReadyQueue()
+    Sync.bindUnsubscribe()
 
-class Sync.Faye extends Sync.Adapter
 
+class Sync.Faye extends Sync.Adapter
   subscriptions: []
 
   available: ->
     !!window.Faye
 
   connect: ->
-    @client = new window.Faye.Client(Sync.FAYE_HOST)
+    @client = new window.Faye.Client(Sync.SERVER)
 
   isConnected: -> @client?.getState() is "CONNECTED"
 
@@ -129,16 +156,11 @@ class Sync.Pusher extends Sync.Adapter
     !!window.Pusher
 
   connect: ->
-    @client = new window.Pusher(Sync.PUSHER_API_KEY, encrypted: Sync.ENCRYPTION_FLAG)
+    # Pusher doesn't properly transform native boolean flags in options.
+    @client = new window.Pusher(Sync.API_KEY, encrypted: Sync.ENCRYPTION_FLAG.toString())
 
   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.Pusher.Subscription
   constructor: (@client, channel, callback) ->
@@ -150,6 +172,60 @@ class Sync.Pusher.Subscription
   cancel: ->
     @client.unsubscribe(@channel) if @client.channel(@channel)?
 
+class Sync.Stomp extends Sync.Adapter
+
+  subscriptions: []
+
+  client: null
+  socket: null
+
+  headers: {
+    login: Sync.API_KEY,
+    passcode: Sync.AUTH_TOKEN
+  }
+
+  available: ->
+    !!window.Stomp
+
+  connect: ->
+    @socket = new window.SockJS(Sync.WEBSOCKET)
+    @client = window.Stomp.over(@socket)
+
+    # SockJS does not support heart-beat: disable heart-beats
+    @client.heartbeat.outgoing = 0
+    @client.heartbeat.incoming = 0
+    @client.debug = Sync.onDebug if Sync.DEBUG_FLAG
+
+    @client.connect(
+      @headers['login'],
+      @headers['passcode'],
+      @onConnect,
+      @onError,
+      '/'
+    )
+
+  onError: (error) ->
+    console.log({ error: error })
+
+  isConnected: ->
+    try
+      @client.connected
+    catch error
+      false
+
+
+class Sync.Stomp.Subscription
+  constructor: (client, channel, callback) ->
+    @channel = channel
+    @client = client
+    # @subscription = @client.subscribe(channel, callback)
+    @subscription = @client.subscribe channel, ( (e) ->
+      callback(JSON.parse(e.body))
+    )
+
+  cancel: ->
+    @subscription.unsubscribe()
+
 
 class Sync.View
 

lib/generators/sync/templates/sync.yml

@@ -16,6 +16,17 @@ development:
 #   encryption: true
 #   async: true
 
+# Stomp
+# development:
+#   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
@@ -86,6 +87,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
@@ -110,6 +121,11 @@ def encryption_flag
       config[:encryption] || false
     end
 
+    def debug_flag
+      return config[:debug] unless config.fetch(:debug, nil).nil?
+      defined?(::Rails) && !::Rails.env.production?
+    end
+
     def reactor
       @reactor ||= Reactor.new
     end

lib/sync/clients/stomp.rb

@@ -0,0 +1,67 @@
+module Sync
+  module Clients
+    class Stomp
+      attr_reader :adapter
+
+      def setup
+        require 'stomp'
+
+        @adapter ||= ::Stomp::Client.new(Sync.server)
+      end
+
+      def batch_publish(*args)
+        Message.batch_publish(@adapter, *args)
+      end
+
+      def build_message(*args)
+        Message.new(*args)
+      end
+
+      # Public: Normalize channel to adapter supported format
+      #
+      # channel - The string channel name
+      #
+      # Returns The normalized channel prefixed with supported format for Stomp
+      def normalize_channel(channel)
+        return "#{Sync.destination}#{channel}" unless Sync.destination[-1] == '/'
+        "#{Sync.destination}/#{channel}"
+      end
+
+
+      class Message
+
+        attr_accessor :channel, :client, :data
+
+        def self.batch_publish(client, messages)
+          messages.each do |message|
+            message.client = client
+            message.publish
+          end
+        end
+
+        def initialize(channel, data)
+          self.channel = channel
+          self.data = data
+        end
+
+        def publish
+          if Sync.async?
+            publish_asynchronous
+          else
+            publish_synchronous
+          end
+        end
+
+        def publish_synchronous
+          client.publish(channel, data.to_json)
+        end
+
+        def publish_asynchronous
+          Sync.reactor.perform do
+            client.publish(channel, data.to_json)
+          end
+        end
+      end
+    end
+  end
+end

sync.gemspec

@@ -23,6 +23,7 @@ Gem::Specification.new do |s|
   s.add_development_dependency 'pry'
   s.add_development_dependency 'minitest', '< 5.0.0'
   s.add_development_dependency 'codeclimate-test-reporter'
+  s.add_development_dependency 'stomp', '~> 1.3'
 
   s.required_rubygems_version = ">= 1.3.4"
 end

test/fixtures/sync_stomp.yml

@@ -0,0 +1,10 @@
+# Stomp
+test:
+  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-'