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

@@ -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() || !@adapter.available()
-      @ready = true
-      @connect()
-      @flushReadyQueue()
-      @bindUnsubscribe()
+      onReadyInterval = setInterval (=>
+        return if @isReady() || !@adapter.available()
+        clearInterval(onReadyInterval)
+        @ready = true
+        @connect()
+      ), 250
 
 
   # Handle Turbolinks teardown, unsubscribe from all channels before transition
@@ -30,15 +32,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: ->
@@ -89,14 +103,23 @@ 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()
 
-class Sync.Faye extends Sync.Adapter
 
+class Sync.Faye extends Sync.Adapter
   subscriptions: []
 
   available: ->
@@ -126,8 +149,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 +160,6 @@ 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.Pusher.Subscription
   constructor: (@client, channel, callback) ->
@@ -154,6 +171,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: SyncConfig.api_key,
+    passcode: SyncConfig.auth_token
+  }
+
+  available: ->
+    !!window.Stomp
+
+  connect: ->
+    @socket = new window.SockJS(SyncConfig.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 SyncConfig.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

@@ -15,6 +15,17 @@ development:
 #   adapter: "Pusher"
 #   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
@@ -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,14 +159,19 @@ def pusher_wss_port
       config[:pusher_wss_port]
     end
 
-    def pusher_encrypted
-      if config[:pusher_encrypted].nil?
+    def encryption_flag
+      if config[:encryption].nil?
         true
       else
-        config[:pusher_encrypted]
+        config[:encryption]
       end
     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-'