ActionCableコードリーディングその1


[その2](http://qiita.com/wakaba260/items/2f578b8e08ef6ac0c1c4)
[その3](http://qiita.com/wakaba260/items/bc03a306604f2f543a15)

## 概要

ActionCableのコードリーディングをしつつメモ書き。
今回は初期化部分を見ていきます。

## ActionCable::Server::Base#call

ActionCableはRails::Engineとして実装されているため、ここがエンドポイントになる。

```ruby
def call(env)
  setup_heartbeat_timer
  config.connection_class.new(self, env).process
end
```

setup_heartbeat_timerはActionCable::Server::Connectionモジュールのメソッドです。

```ruby
# WebSocket connection implementations differ on when they'll mark a connection as stale. We basically never want a connection to go stale, as you
# then can't rely on being able to receive and send to it. So there's a 3 second heartbeat running on all connections. If the beat fails, we automatically
# disconnect.
def setup_heartbeat_timer
  EM.next_tick do
    @heartbeat_timer ||= EventMachine.add_periodic_timer(BEAT_INTERVAL) do
      EM.next_tick { connections.map(&:beat) }
    end
  end
end
```

コメントに書いてある通り、3秒ごとにすべてのコネクションにheartbeatを送って、失敗したらdisconnectするメソッドですね。

### ActionCable::Server::Configuration

configはActionCable::Server::Configurationを見ています。

```ruby
cattr_accessor(:config, instance_accessor: true) { ActionCable::Server::Configuration.new }
```

connection_classはAppllicatinCable::Connectionのようです。

```ruby
# ActionCable::Server::Configuration
def initialize
  @log_tags = []

  @connection_class  = ApplicationCable::Connection
  @worker_pool_size  = 100

  @channels_path = Rails.root.join('app/channels')

  @disable_request_forgery_protection = false
end
```

### ApplicationCable::Connection

ApplicationCable::ConnectionはActionCable::Connection::Baseを継承しています。

```ruby
module ApplicationCable
  class Connection < ActionCable::Connection::Base
  end
end
```

コンストラクタはこのようになっています。
serverにActionCable::Server::Baseが、
envにActionCable::Server::Base#callに渡されたenvがそのまま渡されます。
どうやらwebsocketの初期化もここで行っているようです。

```ruby
def initialize(server, env)
  @server, @env = server, env

  @logger = new_tagged_logger

  @websocket      = ActionCable::Connection::WebSocket.new(env)
  @subscriptions  = ActionCable::Connection::Subscriptions.new(self)
  @message_buffer = ActionCable::Connection::MessageBuffer.new(self)

  @_internal_redis_subscriptions = nil
  @started_at = Time.now
end
```

### ActionCable::Connection::Base#process

initializeの後に呼ばれるprocessメソッドは次のようになっています。

```ruby
def process
  logger.info started_request_message

  if websocket.possible? && allow_request_origin?
    websocket.on(:open)    { |event| send_async :on_open   }
    websocket.on(:message) { |event| on_message event.data }
    websocket.on(:close)   { |event| send_async :on_close  }

    respond_to_successful_request
  else
    respond_to_invalid_request
  end
end
```

websocketが可能かどうか・request_originが許可されているかどうかをチェックして、成功・失敗を返すようです。
また、websocketへ on/message/close のイベントハンドラの登録もここで行われています。

respond_to_successful_requestの中身はrackのレスポンスを返すだけです。

```ruby
def respond_to_successful_request
  websocket.rack_response
end
```

respond_to_invalid_requestの中身はwebsocketがまだ生きてたらcloseして404を返すようになっています。

```ruby
def respond_to_invalid_request
  close if websocket.alive?

  logger.info finished_request_message
  [ 404, { 'Content-Type' => 'text/plain' }, [ 'Page not found' ] ]
end
```

## ActionCable::Connection::WebSocket

ActionCableのwebsocketの部分はfeya-websocketというgemのラッパーとして実装されています。

```ruby
require 'faye/websocket'

module ActionCable
  module Connection
    # Decorate the Faye::WebSocket with helpers we need.
    class WebSocket
      delegate :rack_response, :close, :on, to: :websocket

      def initialize(env)
        @websocket = Faye::WebSocket.websocket?(env) ? Faye::WebSocket.new(env) : nil
      end

      def possible?
        websocket
      end

      # ...
```

[feya-websocket](https://github.com/faye/faye-websocket-ruby)は[feya](https://github.com/faye/faye)というシンプルなPub/Subメッセージングを行うためのgemのwebsocketの部分を切り出したgemだそうです。
単品で使ったことがありますが、簡単にwebsocketを実装できたので、websocketサーバ単独で動かすときは選択肢の1つになると思います。

先ほど出てきたpossible?メソッドはwebsocketのインスタンスがあるかどうか判定してるだけですね。
Rackから渡されたenvがwebsocketのrequestかどうか判定して、websocketでなければnilが設定されるようになっています。


### ActionCable::Connection::Subscriptions

重要そうな名前なのでこちらも覗き見。

```ruby
module ActionCable
  module Connection
    # Collection class for all the channel subscriptions established on a given connection. Responsible for routing incoming commands that arrive on
    # the connection to the proper channel. Should not be used directly by the user.
    class Subscriptions
      def initialize(connection)
        @connection = connection
        @subscriptions = {}
      end
```

コメントに書いてある通り、接続が確立しているすべてのchannel subscriptionsのコレクションクラスのようです。
クライアントから送信されたコマンドによって適切なチャンネルへとルーティングする責務を担うみたいです。

### ActionCable::Connection::MessageBuffer

ついでにこちらも。

```ruby
module ActionCable
  module Connection
    # Allows us to buffer messages received from the WebSocket before the Connection has been fully initialized and is ready to receive them.
    # Entirely internal operation and should not be used directly by the user.
    class MessageBuffer
      def initialize(connection)
        @connection = connection
        @buffered_messages = []
      end
```

少なくとも接続が初期化される前にはwebsocketからのメッセージをバッファリング・受信する準備が出来ているようにするためのクラス……ってことですかね。
中身見てもappendとかreceiveとかprocessとか、process_buffered_messagesといったメソッドがあるのでそれっぽいです。

## 今回のまとめ

ActionCableがリクエストを受け取り、WebSocketのconnectionが確立されるまでの流れを見ることが出来ました。
次はメッセージングやPubSubの部分を見ていけたらと思います。