Day 30 -- Rails 实作 Action Cable 即时交易功能

Action Cable 毫无疑问地在 Rails的发展史上立下了ㄧ个重要的里程碑，它将 WebSocket 网路传输协定整合在 Rails 框架中，以 Pub/Sub 模式让开发者得以实现即时更新的功能。

接下来需要先了解以下主题：

• Action Cable 基础
• Sidekiq 基础
• Redis 的基本认识
• Paranoia
• AASM - Ruby state machines
• simple_form

预计 Model 之间的关系会是这样：

Step 1：建立使用者验证机制 

参考：快速实作使用者验证机制
Branch: "feature/Step_1_User_Authentication"

Step 2：建立商品的基本ＣＲＵＤ

Branch: "feature/Step_2_Commodity"

Step 3：加入Action Mailer 寄信功能

参考：实作开发模式 Action Mailer 寄信功能 with Sidekiq
Branch: "feature/Step_3_Action_Mailer"

Step 4：加入 Action Cable 即时功能

Branch: "feature/Step_4_Action_Cable"

Step 5: 部署 Heroku

Branch: "feature/Step_5_Deploy_to_Heroku"
完成後的 Demo

Connection Setup

这里要先设定服务器与客户端的连接的认证方式。

" 连接是客户端-服务器通信的基础。每当服务器接受一个WebSocket，就会实例化一个连接对象。所有频道订阅（channel subscription）都是在继承连接对象的基础上创建的。连接本身并不处理身份验证和授权之外的任何应用逻辑。WebSocket 连接的客户端被称为连接用户（connection consumer）。每当用户新打开一个浏览器标签、窗口或设备，对应地都会新建一个用户-连接对（consumer-connection pair）。" from Action Cable 概览

# app/channels/application_cable/connection.rb
module ApplicationCable
  class Connection < ActionCable::Connection::Base
    identified_by :current_user
    def connect
      self.current_user = find_verified_user
    end
    private
    def find_verified_user
      verified_user =  User.find_by(id: cookies.signed['user.id'])
      if verified_user && cookies.signed['user.expires_at'] > Time.now       
        verified_user
      else
        reject_unauthorized_connection
      end
    end
  end
end

Channel Setup

这里 Channel 的概念与 Rails MVC 的 Controller 相近。在预设情况下，Rails 将以 ApplicationCable::Channel为所有频道的上层，如果有共用的方法或是逻辑就可以放在这里。

# app/channels/application_cable/channel.rb
module ApplicationCable
  class Channel < ActionCable::Channel::Base
  end
end

接着建立交易平台的频道，在终端机输入：rails generate channel board

# app/channels/board_channel.rb
class BoardChannel < ApplicationCable::Channel
  def subscribed
    stream_from "board"，
  end
  def unsubscribed
    # Any cleanup needed when channel is unsubscribed
  end
end

频道建立完成後，登入後的会员就可以订阅交易平台的频道，成为频道订阅者(subscriber)便可以收到发布的消息。

Broadcasting

" 广播是指发布/订阅的链接，也就是说，当频道订阅者使用流接收某个广播时，发布者发布的内容会被直接发送给订阅者。
广播也是时间相关的在线队列。如果用户未使用流（即未订阅频道），稍後就无法接收到广播 " from Action Cable 概览

这次所实作的交易平台在三种情况下需要透过『广播』，让订阅者可即时更新平台上的内容：

1. 会员新增委托单时，交易平台会即时增加该笔委托单
2. 会员取消委托单时，交易平台会即时移除该笔委托单
3. 委托单成交时，交易平台会即时移除已成交委托单

因此需要在 Commodity controller 里的 create 、destroy 及 deal action 广播

def create
  @commodity = Commodity.new(commodity_params)
  @commodity.user_id = current_user.id
 
  if @commodity.save
    SendCommodityWorker.perform_async(@commodity.id)
    redirect_to commodities_path, notice: "新增成功"
  else
    render :new
  end
end  
--------------------------------------------------------------------
def destroy
  if @commodity.may_cancel?
    RemoveCommodityWorker.perform_async(@commodity.id)
    @commodity.destroy
    @commodity.cancel!
    redirect_to commodities_path, notice: "取消成功"
  else
    redirect_to commodities_path, notice: "此委托单已成交"
  end
end
--------------------------------------------------------------------
def deal
  @commodity = Commodity.find_by(id: params[:id])
  if @commodity && @commodity.trade!
    @commodity.closer_id = current_user.id
    @commodity.save
    RemoveCommodityWorker.perform_async(@commodity.id)
    # 寄给结单使用者
    MailWorker.perform_async(@commodity.closer_id)
    # 寄给挂单委托者
    MailWorker.perform_async(@commodity.user_id)
    # 寄给系统管理员
    MailAdminWorker.perform_async(1, @commodity.user_id,  
    @commodity.closer_id)
    redirect_to commodities_path, notice: "下单成功"
  else
    redirect_to commodities_path, notice: "下单失败"
  end
end

接着建立两个 worker 让 sidekiq 处理背景任务：

# app/workers/send_commodity_worker.rb
class SendCommodityWorker
  include Sidekiq::Worker
  sidekiq_options queue: :default, retry: 3
  def perform(commodity_id)
    commodity = Commodity.find(commodity_id)
    html = CommoditiesController.render(
      partial: 'commodity',
      locals: { commodity: commodity }
      ).squish
    ActionCable.server.broadcast "board", html: html
  end
end

# ---------------------------------------------------------
# app/workers/remove_commodity_worker.rb
class RemoveCommodityWorker
  include Sidekiq::Worker
  sidekiq_options queue: :default, retry: 3
  def perform(commodity_id)
    commodity = Commodity.find(commodity_id)
    html = "commodity_#{commodity.id}"
    closed = CommoditiesController.render(
      partial: 'closed_commodity',
      locals: { commodity: commodity }
    ).squish
    ActionCable.server.broadcast "board", deal: html, closed: closed
  end
end

Subscriptions

" 订阅频道的用户，称为订阅者。用户创建的连接称为（频道）订阅。订阅基於连接用户（订阅者）发送的标识符创建，收到的消息将被发送到这些订阅。" from Action Cable 概览

目前交易平台的设计是不允许订阅者互相传送讯息，因此不需要有另外的频道。登入系统後的订阅者都会收到相同的讯息：

# app > javascript > channels > board_channel.js
import consumer from "./consumer"
consumer.subscriptions.create("BoardChannel", {
  // Called when the subscription is ready for use on the server     
  connected() {
    console.log("Connected to Board channel")
  },
  // Called when incoming data on the websocket for this channel
  received(data) {
   const commodityContainer = document.getElementById('commodities')  
   const closed_commodityContainer = 
   document.getElementById('closed_commodities')
   const commodityRecords = 
   document.getElementsByClassName(data.deal)
   // data['deal'] is equal to data.deal
   // if the "deal" key can't be found in data hash, javascript will  
   // return "undefined"
   // it means the create action is required, and vice versa
   if (data['deal']!= undefined) {
     var i = 0;
     do {
       commodityRecords[i].innerText='';
       i += 1;
     } while (i < commodityRecords.length);
     closed_commodityContainer.innerHTML = data.closed +  
     closed_commodityContainer.innerHTML;
   }
   else
     commodityContainer.innerHTML = data.html + 
     commodityContainer.innerHTML;
  }
});

以上面的程序码为例：我们用 connected() 方法在 Chrome Console 看出是否有出现 " Connected to Board Channel"，有出现的话就代表成功连到 Board channel。

另一个 received(data) 方法则是代表当资料从频道取得後，要作如何的操作。在这里可以在 HTML 的 tr 元素里绑住新增的『商品委托单ID』，再以 class 的方式来选取该笔新增的资料。

 " <tr class='commodity_28'> 
   <td>28</td> 
   <td>买进</td> 
   <td>台积电</td> 
   <td>5</td> 
   <td>$100.0</td> 
   <td>$500.0</td> 
   <td>
     <a data-confirm="确认要卖出吗？" class="btn btn-outline-primary"  
      rel="nofollow" data-method="post" href="/commodities/28/deal">
      卖出
     </a>
   </td> 
   </tr> "

确认有抓下来该笔资料後，可用 data.html + commodityContainer.innerHTML 把新增的资料加在原有资料表里的最上方。

Step 5： 部署 Heroku

Branch: "feature/Step_5_Deploy_to_Heroku"

这边是完成後的 Demo

参考资料：

Action Cable Overview
Action Cable 概览
Ruby on Rails 中的 Sidekiq 的介绍与应用