Replace pusher with action cable (#178)

closes #43

app/channels/application_cable/channel.rb

@@ -0,0 +1,2 @@
+class ApplicationCable::Channel < ActionCable::Channel::Base
+end

app/channels/application_cable/connection.rb

@@ -0,0 +1,2 @@
+class ApplicationCable::Connection < ActionCable::Connection::Base
+end

app/channels/room_channel.rb

@@ -0,0 +1,5 @@
+class RoomChannel < ApplicationCable::Channel
+  def subscribed
+    stream_from params[:pubsub_token]
+  end
+end

app/dispatchers/sync_dispatcher.rb

@@ -5,6 +5,6 @@ class SyncDispatcher < BaseDispatcher
   end
 
   def listeners
-    [PusherListener.instance]
+    [ActionCableListener.instance]
   end
 end

app/javascript/dashboard/constants.js

@@ -1,6 +1,5 @@
 export default {
   APP_BASE_URL: '/',
-  PUSHER: __PUSHER__,
   get apiURL() {
     return `${this.APP_BASE_URL}/`;
   },

app/javascript/dashboard/helper/actionCable.js

@@ -0,0 +1,70 @@
+import { createConsumer } from '@rails/actioncable';
+
+import AuthAPI from '../api/auth';
+
+class ActionCableConnector {
+  constructor(app, pubsubToken) {
+    const consumer = createConsumer();
+    consumer.subscriptions.create(
+      {
+        channel: 'RoomChannel',
+        pubsub_token: pubsubToken,
+      },
+      {
+        received: this.onReceived,
+      }
+    );
+    this.app = app;
+    this.events = {
+      'message.created': this.onMessageCreated,
+      'conversation.created': this.onConversationCreated,
+      'status_change:conversation': this.onStatusChange,
+      'user:logout': this.onLogout,
+      'page:reload': this.onReload,
+      'assignee.changed': this.onAssigneeChanged,
+    };
+  }
+
+  onAssigneeChanged = payload => {
+    const { meta = {}, id } = payload;
+    const { assignee } = meta || {};
+    if (id) {
+      this.app.$store.dispatch('updateAssignee', { id, assignee });
+    }
+  };
+
+  onConversationCreated = data => {
+    this.app.$store.dispatch('addConversation', data);
+  };
+
+  onLogout = () => AuthAPI.logout();
+
+  onMessageCreated = data => {
+    this.app.$store.dispatch('addMessage', data);
+  };
+
+  onReceived = ({ event, data } = {}) => {
+    if (this.events[event]) {
+      this.events[event](data);
+    }
+  };
+
+  onReload = () => window.location.reload();
+
+  onStatusChange = data => {
+    this.app.$store.dispatch('addConversation', data);
+  };
+}
+
+export default {
+  init() {
+    if (AuthAPI.isLoggedIn()) {
+      const actionCable = new ActionCableConnector(
+        window.WOOT,
+        AuthAPI.getPubSubToken()
+      );
+      return actionCable;
+    }
+    return null;
+  },
+};

app/javascript/dashboard/helper/pusher.js

@@ -1,71 +0,0 @@
-/* eslint-env browser */
-/* eslint no-console: 0 */
-import Pusher from 'pusher-js';
-import AuthAPI from '../api/auth';
-import CONSTANTS from '../constants';
-
-class VuePusher {
-  constructor(apiKey, options) {
-    this.app = options.app;
-    this.pusher = new Pusher(apiKey, options);
-    this.channels = [];
-  }
-
-  subscribe(channelName) {
-    const channel = this.pusher.subscribe(channelName);
-    if (!this.channels.includes(channel)) {
-      this.channels.push(channelName);
-    }
-    this.bindEvent(channel);
-  }
-
-  unsubscribe(channelName) {
-    this.pusher.unsubscribe(channelName);
-  }
-
-  bindEvent(channel) {
-    channel.bind('message.created', data => {
-      this.app.$store.dispatch('addMessage', data);
-    });
-
-    channel.bind('conversation.created', data => {
-      this.app.$store.dispatch('addConversation', data);
-    });
-
-    channel.bind('status_change:conversation', data => {
-      this.app.$store.dispatch('addConversation', data);
-    });
-
-    channel.bind('assignee.changed', payload => {
-      const { meta = {}, id } = payload;
-      const { assignee } = meta || {};
-      if (id) {
-        this.app.$store.dispatch('updateAssignee', { id, assignee });
-      }
-    });
-
-    channel.bind('user:logout', () => AuthAPI.logout());
-    channel.bind('page:reload', () => window.location.reload());
-  }
-}
-
-/* eslint no-param-reassign: ["error", { "props": false }] */
-export default {
-  init() {
-    // Log only if env is testing or development.
-    Pusher.logToConsole = CONSTANTS.PUSHER.logToConsole || true;
-    // Init Pusher
-    const options = {
-      encrypted: true,
-      app: window.WOOT,
-      cluster: CONSTANTS.PUSHER.cluster,
-    };
-    const pusher = new VuePusher(CONSTANTS.PUSHER.token, options);
-    // Add to global Obj
-    if (AuthAPI.isLoggedIn()) {
-      pusher.subscribe(AuthAPI.getPubSubToken());
-      return pusher.pusher;
-    }
-    return null;
-  },
-};

app/javascript/dashboard/routes/index.spec.js

@@ -14,7 +14,6 @@ jest.mock('./login/login.routes', () => ({
 jest.mock('../constants', () => {
   return {
     APP_BASE_URL: '/',
-    PUSHER: false,
     get apiUrl() {
       return `${this.APP_BASE_URL}/`;
     },

app/javascript/dashboard/store/modules/auth.js

@@ -8,7 +8,7 @@ import * as types from '../mutation-types';
 import router from '../../routes';
 import authAPI from '../../api/auth';
 import createAxios from '../../helper/APIHelper';
-import vuePusher from '../../helper/pusher';
+import actionCable from '../../helper/actionCable';
 // initial state
 const state = {
   currentUser: {
@@ -61,7 +61,7 @@ const actions = {
         .then(() => {
           commit(types.default.SET_CURRENT_USER);
           window.axios = createAxios(axios);
-          window.pusher = vuePusher.init(Vue);
+          actionCable.init(Vue);
           router.replace({ name: 'home' });
           resolve();
         })

app/javascript/dashboard/store/modules/conversations/index.js

@@ -168,7 +168,7 @@ const mutations = {
     _state.chatStatusFilter = data;
   },
 
-  // Update assignee on pusher message
+  // Update assignee on action cable message
   [types.default.UPDATE_ASSIGNEE](_state, payload) {
     const [chat] = _state.allConversations.filter(c => c.id === payload.id);
     chat.meta.assignee = payload.assignee;

app/javascript/packs/application.js

@@ -24,7 +24,7 @@ import createAxios from '../dashboard/helper/APIHelper';
 import commonHelpers from '../dashboard/helper/commons';
 import router from '../dashboard/routes';
 import store from '../dashboard/store';
-import vuePusher from '../dashboard/helper/pusher';
+import vueActionCable from '../dashboard/helper/actionCable';
 import constants from '../dashboard/constants';
 
 Vue.config.env = process.env;
@@ -58,7 +58,7 @@ window.onload = () => {
     components: { App },
     template: '<App/>',
   }).$mount('#app');
-  window.pusher = vuePusher.init();
+  vueActionCable.init();
 };
 
 if ('serviceWorker' in navigator) {

app/listeners/action_cable_listener.rb

@@ -1,46 +1,53 @@
-class PusherListener < BaseListener
+class ActionCableListener < BaseListener
   include Events::Types
 
   def conversation_created(event)
     conversation, account, timestamp = extract_conversation_and_account(event)
     members = conversation.inbox.members.pluck(:pubsub_token)
-    Pusher.trigger(members, CONVERSATION_CREATED, conversation.push_event_data) if members.present?
+    send_to_members(members, CONVERSATION_CREATED, conversation.push_event_data)
   end
 
   def conversation_read(event)
     conversation, account, timestamp = extract_conversation_and_account(event)
     members = conversation.inbox.members.pluck(:pubsub_token)
-    Pusher.trigger(members, CONVERSATION_READ, conversation.push_event_data) if members.present?
+    send_to_members(members, CONVERSATION_READ, conversation.push_event_data)
   end
 
   def message_created(event)
     message, account, timestamp = extract_message_and_account(event)
     conversation = message.conversation
     members = conversation.inbox.members.pluck(:pubsub_token)
-
-    Pusher.trigger(members, MESSAGE_CREATED, message.push_event_data) if members.present?
+    send_to_members(members, MESSAGE_CREATED, message.push_event_data)
   end
 
   def conversation_reopened(event)
     conversation, account, timestamp = extract_conversation_and_account(event)
     members = conversation.inbox.members.pluck(:pubsub_token)
-    Pusher.trigger(members, CONVERSATION_REOPENED, conversation.push_event_data) if members.present?
+    send_to_members(members, CONVERSATION_REOPENED, conversation.push_event_data)
   end
 
   def conversation_lock_toggle(event)
     conversation, account, timestamp = extract_conversation_and_account(event)
     members = conversation.inbox.members.pluck(:pubsub_token)
-    Pusher.trigger(members, CONVERSATION_LOCK_TOGGLE, conversation.lock_event_data) if members.present?
+    send_to_members(members, CONVERSATION_LOCK_TOGGLE, conversation.lock_event_data)
   end
 
   def assignee_changed(event)
     conversation, account, timestamp = extract_conversation_and_account(event)
     members = conversation.inbox.members.pluck(:pubsub_token)
-    Pusher.trigger(members, ASSIGNEE_CHANGED, conversation.push_event_data) if members.present?
+    send_to_members(members, ASSIGNEE_CHANGED, conversation.push_event_data)
   end
 
   private
 
+  def send_to_members(members, event_name, data)
+    return if members.blank?
+
+    members.each do |member|
+      ActionCable.server.broadcast(member, event: event_name, data: data)
+    end
+  end
+
   def push(pubsub_token, data)
     # Enqueue sidekiq job to push event to corresponding channel
   end

app/models/concerns/pubsubable.rb

@@ -4,7 +4,7 @@ module Pubsubable
   extend ActiveSupport::Concern
 
   included do
-    # Used by the pusher/PubSub Service we use for real time communications
+    # Used by the actionCable/PubSub Service we use for real time communications
     has_secure_token :pubsub_token
   end
 end

app/models/user.rb

@@ -12,7 +12,7 @@ class User < ApplicationRecord
          :validatable,
          :confirmable
 
-  # Used by the pusher/PubSub Service we use for real time communications
+  # Used by the actionCable/PubSub Service we use for real time communications
   has_secure_token :pubsub_token
 
   validates_uniqueness_of :email, scope: :account_id