Làm chủ Push và Pub/Sub với Ruby: Xây dựng Real-Time WebSocket Apps

Xây dựng real-time features trong Rails dễ dàng hơn nhiều với thư viện như Action Cable. Trong episode AppSignal Academy này, chúng ta sẽ đi sâu vào real-time updating và thử xây dựng minimal WebSocket server để xem nó hoạt động under the hood như thế nào.

Chúng ta sẽ build ứng dụng push data, dùng Pub/Sub qua WebSocket. Trước khi code, hãy hiểu 3 khái niệm này:

• Push là push data đến receiver thay vì receiver poll data. Bắt buộc cho realtime updates như giá cổ phiếu, chat app, hoặc operations console.

• Pub/Sub (publish/subscribe) là interaction model push data phổ biến nhờ TIBCO trên Wall Street năm 1990. Receiver subscribe subject và chờ publisher push data đến subject. Thường có wildcard pattern matching match published message đến listener, tuy một số implementation đơn giản chỉ dùng named channels thay vì wildcard. Tôi bắt đầu tại TIBCO những ngày đầu nên thích sự linh hoạt của wildcard pattern matching.

• WebSocket là protocol exchange data - thường giữa web browser và application. HTTP connection upgraded thành WebSocket connection rồi data gửi 2 chiều giữa 2 endpoint. WebSocket push data từ app đến browser. Cũng cung cấp mechanism khác POST/PUT gửi data từ JavaScript browser về app. Quite nice, huh?

Under the Hood (Bên trong)

Hãy xem ví dụ WebSocket server có thể hoạt động như thế nào. Từ browser, client cố gắng tạo WebSocket connection đến server với JavaScript.

var sock = new WebSocket("ws://" + document.URL.split("/")[2] + "/upgrade");

Server nhận HTTP request với indicator upgrade requested. Thường server để app quyết định upgrade hay không. Cách làm phụ thuộc API cung cấp cho app. Server support Rack cung cấp option hijack socket để developer handle tất cả protocol details hoặc, theo proposed PR, response upgrade là đủ.

Upgrade là tập exchange giữa server và client. Tất cả browser và một số server gem che giấu details. Khi connection established, message exchange theo WebSocket protocol.

Magic under the hood handle encoding, decoding, message exchange protocol. Messages là binary, fixed-width structures với trailing payload, encrypted dùng SHA1. WebSocket protocol bao gồm several message types và exchanges, như ping/pong heartbeats và opening/closing message exchanges. Đó là phép màu server perform bằng cách không dùng connection-hijack approach.

Diving In (Đi sâu)

Chúng ta dùng ví dụ clock thread start publish current time đến all listening clients. Dùng Agoo build server vì fast và keep complexity minimum.

Bắt đầu với JavaScript client show current time trên HTML page. Sau tạo WebSocket mới, onopen callback set thay đổi status HTML element. onmessage callback update message HTML element. Callbacks là design pattern phổ biến khi làm async calls như publish/subscribe exchanges.

<!-- websocket.html -->
<html>
<body>
<p id="status">...</p>
<p id="message">... waiting ...</p>
<script type="text/javascript">
var sock = new WebSocket(
  "ws://" + document.URL.split("/")[2] + "/upgrade"
);
sock.onopen = function () {
  document.getElementById("status").textContent = "connected";
};
sock.onmessage = function (msg) {
  document.getElementById("message").textContent = msg.data;
};
</script>
</body>
</html>

Client xong, implement server - Ruby app dùng Rack API. Class Clock sẽ handler cho tất cả HTTP request trên path /upgrade. Nếu request upgrade ta return Success với HTTP status 200, ngược lại 404 Page Not Found. Bước duy nhất khác trong #call là assign WebSocket handler.

class Clock
  def self.call(env)
    unless env['rack.upgrade?'].nil?
      env['rack.upgrade'] = Clock
      [ 200, { }, [ ] ]
    else
      [ 404, { }, [ ] ]
    end
  end
end

API dựa trên callbacks. Callback duy nhất quan trọng cho server là #on_open cho phép tạo subscription đến subject "time". Message exchange qua channels identified bởi subjects/topics. #on_open gọi khi web socket connection established.

class Clock
  # ...
  def self.on_open(client)
    client.subscribe('time')
  end
end

Giờ start publish với thread publish time mỗi giây. Gọi Agoo.publish gửi message trên subject "time", tất cả subscriber nhận message. Server track subscriptions và connections, deliver message đến JavaScript client update HTML element.

require 'agoo'
Thread.new {
  loop do
    now = Time.now
    Agoo.publish('time', "%02d:%02d:%02d" % [now.hour, now.min, now.sec])
    sleep(1)
  end
}

Code còn lại chỉ init và start server. Agoo::Server.handle(:GET, '/upgrade', Clock) bảo server listen HTTP GET trên path /upgrade pass request cho class Clock. Cho phép routing xảy ra ngoài Ruby để performance và flexibility.

Agoo::Server.init(6464, '.', thread_count: 0)
Agoo::Server.handle(:GET, '/upgrade', Clock)
Agoo::Server.start

Gần xong. Run server lệnh:

$ ruby pubsub.rb

Log entry hiện server running listening port 6464.

I 2018/08/14 19:49:45.170618000 INFO: Agoo 2.5.0 with pid 40366 is listening on https://:6464.

Time To See If It's Working

Mở https://localhost:6464/websocket.html. Sau flicker ban đầu khi connection established, connection status và time hiển thị. Time increment mỗi giây khi clock tick.

connected
19:50:12

Chúc mừng bạn làm publish/subscribe web app ;-)

Hôm nay chúng ta xem WebSocket. Server Side Events (SSE) option khác làm tương tự, included trong full source code. Muốn biết thêm, xem Agoo server hoặc Iodine WebSocket Server.

Nếu có câu hỏi hoặc comment, đừng ngần ngại drop line @AppSignal.