2018年11月12日

Fragile Base Class

這是物件導向程式語言中,類別繼承所造成的問題,當有類別被其他類別繼承時,常會遇到一些 method 被子類別覆寫 override 的狀況,但覆寫 method 後,卻可能會因為不了解父類別實作的細節,導致程式出錯。

以下是實際的例子

Super 提供兩個 method,但 inc1 會在method 裡面呼叫 inc2,Sub 繼承了 Super,但並不知道 inc1 的實作內容,在 Sub 裡面,覆寫了 inc2,將 inc2 改成呼叫 inc1,因此在使用 Sub 的 inc2 時,會一直重複呼叫函數,最後導致發生 java.lang.StackOverflowError

class Super {

  private int counter = 0;

  void inc1() {
    inc2();
  }

  void inc2() {
    counter++;
  }

}

class Sub extends Super {

  @Override
  void inc2() {
    inc1();
  }

  public static void main(String[] args) {
    Sub sub = new Sub();
    sub.inc2();
  }
}
$ java Sub
Exception in thread "main" java.lang.StackOverflowError
    at Super.inc1(Super.java:6)
    at Sub.inc2(Super.java:21)

這跟上面的例子類似,一樣是會發生重複呼叫函數的問題,最後導致發生 java.lang.StackOverflowError

class Base{
    protected int x;
    protected void m(){
        x++;
    }

    protected void n(){
        x++;      // <- defect
        m();
    }

}

class SubBase extends Base{
    protected void m(){
        n();
    }
    public static void main(String[] args) {
        SubBase sub = new SubBase();
        sub.m();
    }
}
$ java SubBase
Exception in thread "main" java.lang.StackOverflowError
    at Base.n(Base.java:9)
    at SubBase.m(Base.java:16)
    at Base.n(Base.java:9)

Java 提供一種解決方式,就是將無法被覆寫的 method 加上 final。

class Super {

  private int counter = 0;

  void inc1() {
    inc2();
  }

  final void inc2() {
    counter++;
  }
}

class Sub extends Super {

  @Override
  void inc2() {
    inc1();
  }
  public static void main(String[] args) {
    Sub sub = new Sub();
    sub.inc2();
  }
}

在編譯時,就會因為 final 的關係,造成編譯錯誤。

$ javac Super.java
Super.java:20: error: inc2() in Sub cannot override inc2() in Super
  void inc2() {
       ^
  overridden method is final
1 error

Kotlin 提供更積極的解決方案,所有 method 在不加上任何 modifier 的狀況下,預設都是有 final 特性的,如果可以讓子類別覆寫的 method,必須要加上 open 這個 modifier。

這樣的調整,可避免 method 任意被覆寫的問題,但相對的,programmer 要承擔更多責任,判斷什麼時候該加上 open,這有時候會造成一些困擾,就是不知道什麼時候要加上 open,就變成不寫 open。

open class Super {
    open fun v() {}
    fun nv() {}
}
class Sub() : Super() {
    override fun v() {}
}

最後只能說,沒有兩全其美的解決方案,就看程式語言的設計者,認定哪一種想法比較重要。

References

Fragile base class wiki

What is the fragile base class problem?

2018年11月5日

Phoenix_9_OTP

Managing State with Processes

Functional programs are stateless,改用 Process 保存 state

新增 /rumbl/ib/rumbl/counter.ex

defmodule Rumbl.Counter do

  def inc(pid), do: send(pid, :inc)

  def dec(pid), do: send(pid, :dec)

  def val(pid, timeout \\ 5000) do 
    ref = make_ref()
    send(pid, {:val, self(), ref})
    receive do
      {^ref, val} -> val
    after timeout -> exit(:timeout)
    end
  end

  def start_link(initial_val) do 
    {:ok, spawn_link(fn -> listen(initial_val) end)}
  end

  defp listen(val) do 
    receive do
      :inc -> listen(val + 1)
      :dec -> listen(val - 1)
      {:val, sender, ref} ->
        send sender, {ref, val}
        listen(val)
    end
  end
end

這是一個獨立的 counter service,Counter API 有三個

  1. :inc
  2. :dec
  3. :val

:inc :dec 是非同步的呼叫

:val 不同,發送 message 後,會用 receive 等待回應。

make_ref() 是一個 global unique reference,可在 global (cluster) 環境中運作。

^ref 表示我們是以 pattern matching 的方式,判斷是不是回傳了正確的 process reference

OTP 需要一個 startlink function,並以 initialval 為 counter 初始的 state

程式中沒有看到任何 global variable 儲存 state,而是呼叫 listen,listen 會以 receive 去 block 並等待 message,而 val 就是放在這個 function 的參數上。process state 是用 recursive function 的方式不斷重複發送給下一個 listen,這是 tail recursive。


測試 Counter

$ iex -S mix

iex(1)> alias Rumbl.Counter
Rumbl.Counter
iex(2)> {:ok, counter} = Counter.start_link(0)
{:ok, #PID<0.270.0>}
iex(3)> Counter.inc(counter)
:inc
iex(4)> Counter.inc(counter)
:inc

iex(5)> Counter.val(counter)
2
iex(6)> Counter.dec(counter)
:dec
iex(7)> Counter.val(counter)
1

Building GenServer for OTP

更新 /lib/rumbl/counter.ex

defmodule Rumbl.Counter do
  use GenServer

  def inc(pid), do: GenServer.cast(pid, :inc)

  def dec(pid), do: GenServer.cast(pid, :dec)

  def val(pid) do
    GenServer.call(pid, :val)
  end

  def start_link(initial_val) do
    GenServer.start_link(__MODULE__, initial_val)
  end

  def init(initial_val) do
    {:ok, initial_val}
  end

  def handle_cast(:inc, val) do
    {:noreply, val + 1}
  end

  def handle_cast(:dec, val) do
    {:noreply, val - 1}
  end

  def handle_call(:val, _from, val) do
    {:reply, val, val}
  end
end

GenServer.cast 是非同步呼叫,server 以 handle_cast 處理,最後會 return {:noreply, val + 1} ,因為呼叫者不需要這個 reply message

GenServer.call 是同步呼叫,server 以 handle_call 處理

測試

$ iex -S mix
iex(1)> alias Rumbl.Counter
Rumbl.Counter
iex(2)> {:ok, counter} = Counter.start_link(0)
{:ok, #PID<0.269.0>}
iex(3)> Counter.inc(counter)
:ok
iex(4)> Counter.val(counter)
1
iex(5)> Counter.dec(counter)
:ok
iex(6)> Counter.val(counter)
0

Adding Failover

利用 Supervisor 監控 counter

Phoenix 並沒有很多處理 fail exception 的 code,而是以 error reporting 的方式處理,同時加上自動 restart service。

修改 /lib/rumbl.ex

defmodule Rumbl do
  use Application

  # See http://elixir-lang.org/docs/stable/elixir/Application.html
  # for more information on OTP Applications
  def start(_type, _args) do
    import Supervisor.Spec, warn: false

    children = [
      supervisor(Rumbl.Endpoint, []),
      supervisor(Rumbl.Repo, []),
      worker(Rumbl.Counter, [5]), # new counter worker
    ]

    opts = [strategy: :one_for_one, name: Rumbl.Supervisor]
    Supervisor.start_link(children, opts)
  end

  # Tell Phoenix to update the endpoint configuration
  # whenever the application is updated.
  def config_change(changed, _new, removed) do
    Rumbl.Endpoint.config_change(changed, removed)
    :ok
  end
end

child spec 定義 Elixir application 要啟動的 children,將 counter 這個 worker 加入 children list。

opts 是 supervision policy,這裡是使用 :oneforone

:oneforall 是 kill and restart all child processes


在 /lib/rumbl/counter.ex 加上 :tick 及處理 :tick 的 handle_info,每 1000ms 就倒數一次

  def init(initial_val) do
    Process.send_after(self, :tick, 1000)
    {:ok, initial_val}
  end

  def handle_info(:tick, val) do
    IO.puts "tick #{val}"
    Process.send_after(self, :tick, 1000)
    {:noreply, val - 1}
  end

再加上一點檢查,只倒數到 0 ,就會 raise exception,OTP process 會 crash

def init(initial_val) do
    Process.send_after(self, :tick, 1000)
    {:ok, initial_val}
  end

  def handle_info(:tick, val) when val <= 0, do: raise "boom!"
  def handle_info(:tick, val) do
    IO.puts "tick #{val}"
    Process.send_after(self, :tick, 1000)
    {:noreply, val - 1}
  end

但可發現它會自動重新啟動

$ iex -S mix
iex(1)> tick 5
tick 4
tick 3
tick 2
tick 1
[error] GenServer #PID<0.365.0> terminating
** (RuntimeError) boom!
    (rumbl) lib/rumbl/counter.ex:21: Rumbl.Counter.handle_info/2
    (stdlib) gen_server.erl:616: :gen_server.try_dispatch/4
    (stdlib) gen_server.erl:686: :gen_server.handle_msg/6
    (stdlib) proc_lib.erl:247: :proc_lib.init_p_do_apply/3
Last message: :tick
State: 0
tick 5
tick 4
tick 3
tick 2
tick 1

Restart Strategies

預設 child processes 的 restart strategy 是 :permanent,也可以用 restart 指定

worker(Rumbl.Counter, [5], restart: :permanent),
  1. :permanent

    child is always restarted (default)

  2. :temporary

    child is never restarted

  3. :transient

    只在異常終止時 restart,也就是 :normal, :shutdown, {:shutdown, term} 以外的 exit reason

另外還有 maxrestarts 及 maxseconds 參數,在 maxsecodns 時間內可以 restart maxrestarts 次

預設是 3 restarts in 5 seconds

Supervision Strategies

  1. :oneforone

    a child terminates -> supervisor restars only that process

  2. :oneforall

    a child tetminates -> supervisor terminates all children and restarts them

  3. :restforone

    a child terminates -> supervisor terminates all child processes defiend after the one that dies,並 restart all terminated processes

  4. :simpleonefor_one

    類似 :oneforone,用在 supervisor 需要動態 supervise processes 的情況,例如 web server 需要 supervise web requests,通常有 10 ~ 100,000 個 concurrent running processes

把 strategy 換成 :oneforall 測試

def start(_type, _args) do
    import Supervisor.Spec, warn: false

    children = [
      supervisor(Rumbl.Endpoint, []),
      supervisor(Rumbl.Repo, []),
      worker(Rumbl.Counter, [5]), # new counter worker
    ]

    opts = [strategy: :one_for_all, name: Rumbl.Supervisor]
    Supervisor.start_link(children, opts)
  end

啟動 Phoenix server 會發現 Cowboy 也 restart

tick 2
tick 1
[error] GenServer #PID<0.348.0> terminating
** (RuntimeError) boom!
    (rumbl) lib/rumbl/counter.ex:21: Rumbl.Counter.handle_info/2
    (stdlib) gen_server.erl:616: :gen_server.try_dispatch/4
    (stdlib) gen_server.erl:686: :gen_server.handle_msg/6
    (stdlib) proc_lib.erl:247: :proc_lib.init_p_do_apply/3
Last message: :tick
State: 0
[info] Running Rumbl.Endpoint with Cowboy using http://localhost:4000
tick 5

現在把 worker 拿掉,strategy 改回 :oneforone

children = [
    supervisor(Rumbl.Endpoint, []),
    supervisor(Rumbl.Repo, [])
]

opts = [strategy: :one_for_one, name: Rumbl.Supervisor]

Using Agents

agent 類似 GenServer,但只有 5 個 main functions

start_link 啟動 agent, stop 停止, update 更新 agent 狀態

$ iex -S mix

iex(1)> import Agent
Agent
iex(2)> {:ok, agent} = start_link fn -> 5 end
{:ok, #PID<0.258.0>}
iex(3)> update agent, &(&1 + 1)
:ok
iex(4)> get agent, &(&1)
6
iex(5)> stop agent
:ok

加上 :name option

iex(7)> {:ok, agent} = start_link fn -> 5 end, name: MyAgent
{:ok, #PID<0.265.0>}
iex(8)> update MyAgent, &(&1 + 1)
:ok
iex(9)> get MyAgent, &(&1)
6
iex(10)> stop MyAgent
:ok

重複名稱會發生 error

iex(11)> {:ok, agent} = start_link fn -> 5 end, name: MyAgent
{:ok, #PID<0.271.0>}
iex(12)> {:ok, agent} = start_link fn -> 5 end, name: MyAgent
** (MatchError) no match of right hand side value: {:error, {:already_started, #PID<0.271.0>}}

Phoenix.Channel 就是用 Agent 實作

Design an Information System with OTP

新增 /lib/rumbl/infosyssupervisor.ex


defmodule Rumbl.InfoSys.Supervisor do
  use Supervisor

  def start_link() do
    Supervisor.start_link(__MODULE__, [], name: __MODULE__)
  end

  def init(_opts) do
    children = [
      worker(Rumbl.InfoSys, [], restart: :temporary)
    ]

    supervise children, strategy: :simple_one_for_one
  end
end

修改 /lib/rumbl.ex

    children = [
      supervisor(Rumbl.Endpoint, []),
      supervisor(Rumbl.InfoSys.Supervisor, []), # new supervisor
      supervisor(Rumbl.Repo, []),
#      worker(Rumbl.Counter, [5]), # new counter worker
    ]

Building a start_link Proxy

啟動時動態決定要幾個 service,要製作一個 worker,可啟動多個 backends

proxy function 是 client, server 之間的 lightweight function interface

新增 /lib/rumbl/info_sys.ex


defmodule Rumbl.InfoSys do
  @backends [Rumbl.InfoSys.Wolfram]

  defmodule Result do
    defstruct score: 0, text: nil, url: nil, backend: nil
  end

  def start_link(backend, query, query_ref, owner, limit) do 
    backend.start_link(query, query_ref, owner, limit)
  end

  def compute(query, opts \\ []) do 
    limit = opts[:limit] || 10
    backends = opts[:backends] || @backends

    backends
    |> Enum.map(&spawn_query(&1, query, limit))
  end

  defp spawn_query(backend, query, limit) do 
    query_ref = make_ref()
    opts = [backend, query, query_ref, self(), limit]
    {:ok, pid} = Supervisor.start_child(Rumbl.InfoSys.Supervisor, opts)
    {pid, query_ref}
  end
end

InfoSys 跟一般 GenServer 有一點不同,裡面存放到 results 到 module attribute -> 所有支援的 backends 為 list

Result struct 會儲存每個 search result 的結果,還有 score 及 relevnace, text to describe the result, url

startlink 就是 proxy,會再呼叫其他 backend 的 startlink

compute 會 maps over all backends,呼叫每個 backend 的 spawn_query

Building the Wolfram into System

修改 /rumbl/mix.exs

{:sweet_xml, "~> 0.5.0"},
mix deps.get

wolfram 申請帳號,及 APP ID

APP NAME: wolfram
APPID: LP93J3-XXXXXXXXXX

新增設定 /config/dev.secret.exs

use Mix.Config
config :rumbl, :wolfram, app_id: "LP93J3-XXXXXXXXXX"

記得將 /config/dev.secret.exs 放到 .gitignore

修改 /config/dev.exs,最後面加上

import_config "dev.secret.exs"

新增 /lib/rumbl/info_sys/wolfram.ex

defmodule Rumbl.InfoSys.Wolfram do
  import SweetXml
  alias Rumbl.InfoSys.Result

  def start_link(query, query_ref, owner, limit) do 
    Task.start_link(__MODULE__, :fetch, [query, query_ref, owner, limit])
  end

  def fetch(query_str, query_ref, owner, _limit) do 
    query_str
    |> fetch_xml()
    |> xpath(~x"/queryresult/pod[contains(@title, 'Result') or
                                 contains(@title, 'Definitions')]
                            /subpod/plaintext/text()")
    |> send_results(query_ref, owner)
  end

  defp send_results(nil, query_ref, owner) do 
    send(owner, {:results, query_ref, []})
  end
  defp send_results(answer, query_ref, owner) do
    results = [%Result{backend: "wolfram", score: 95, text: to_string(answer)}]
    send(owner, {:results, query_ref, results})
  end

  defp fetch_xml(query_str) do 
    {:ok, {_, _, body}} = :httpc.request(
      String.to_char_list("http://api.wolframalpha.com/v2/query" <>
        "?appid=#{app_id()}" <>
        "&input=#{URI.encode(query_str)}&format=plaintext"))
    body
  end

  defp app_id, do: Application.get_env(:rumbl, :wolfram)[:app_id]
end

這個 module 並沒有 GenServer 的 callbacks,因為這個 process 是一個 task,GenServer 是一個 generic server 可計算並儲存 state,但有時我們只需要 store state 或是 只需要執行某個 function。

Agent 是簡化的 GenServer 可儲存 state task 是個簡單的 process 可執行某個 function

SweetXml 用來 parse XML,Result 是 the struct for the results

Task.start_link 是啟動 Task 的方式

fetch_xml 裡面試用 :httpc,這是 erlang 的 standard library,可處理 HTTP request

sendresults(queryref, owner) 將結果回傳給 requester

有兩種 send_results,分為有 answer 或沒有

先用 iex -S mix 測試

iex(4)> Rumbl.InfoSys.compute("what is elixir?")
[{#PID<0.592.0>, #Reference<0.3775272462.1982070785.26789>}]
iex(5)> flush()
:ok
iex(6)> flush()
:ok
iex(7)> flush()
{:results, #Reference<0.3775272462.1982070785.26789>,
 [%Rumbl.InfoSys.Result{backend: "wolfram", score: 95,
   text: "1 | noun | a sweet flavored liquid (usually containing a small amount of alcohol) used in compounding medicines to be taken by mouth in order to mask an unpleasant taste\n2 | noun | hypothetical substance that the alchemists believed to be capable of changing base metals into gold\n3 | noun | a substance believed to cure all ills",
   url: nil}]}
:ok

要讓 service 更堅固,必須做以下工作

  1. 偵測 backend crash,這樣就不要等 results

  2. 由 backend 取得結果要根據 score 排序

  3. 需要 timeout 機制


Monitoring Processes

使用 Process.monitor 在 waiting results 時偵測 backend crashes,一但設定了 monitor,會在該 process dies 時,收到 message。

測試

iex(1)> pid = spawn(fn -> :ok end)
#PID<0.261.0>
iex(2)> Process.monitor(pid)
#Reference<0.777943872.2254438401.3282>
iex(3)> flush()
{:DOWN, #Reference<0.777943872.2254438401.3282>, :process, #PID<0.261.0>,
 :noproc}
:ok

修改 /lib/rumbl/info_sys.ex

defmodule Rumbl.InfoSys do
  @backends [Rumbl.InfoSys.Wolfram]

  defmodule Result do
    defstruct score: 0, text: nil, url: nil, backend: nil
  end

  def start_link(backend, query, query_ref, owner, limit) do
    backend.start_link(query, query_ref, owner, limit)
  end

  def compute(query, opts \\ []) do
    limit = opts[:limit] || 10
    backends = opts[:backends] || @backends

    backends
    |> Enum.map(&spawn_query(&1, query, limit))
    |> await_results(opts)
    |> Enum.sort(&(&1.score >= &2.score))
    |> Enum.take(limit)
  end

  defp spawn_query(backend, query, limit) do
    query_ref = make_ref()
    opts = [backend, query, query_ref, self(), limit]
    {:ok, pid} = Supervisor.start_child(Rumbl.InfoSys.Supervisor, opts)
    monitor_ref = Process.monitor(pid)
    {pid, monitor_ref, query_ref}
  end

  defp await_results(children, _opts) do
    await_result(children, [], :infinity)
  end

  defp await_result([head|tail], acc, timeout) do
    {pid, monitor_ref, query_ref} = head

    receive do
      {:results, ^query_ref, results} ->
        Process.demonitor(monitor_ref, [:flush])
        await_result(tail, results ++ acc, timeout)
      {:DOWN, ^monitor_ref, :process, ^pid, _reason} ->
        await_result(tail, acc, timeout)
    end
  end

  defp await_result([], acc, _) do
    acc
  end
end

compute 會自動等待 results,收到時,會 sorting by score,並回報 top ones

spawn_query 裡面增加了 Process.monitor(pid)

awaitresults 是 recursive function,在每次呼叫 awaitresults 就會新增一個 result 到 list

正確的 result 會 match {:results, ^query_ref, result}

Process.demonitor(monitor_ref, [:flush]) 是將 monitor process 移除

現在 compute 會自動處理結果

iex(1)> Rumbl.InfoSys.compute("what is the meaning of life?")
[%Rumbl.InfoSys.Result{backend: "wolfram", score: 95,
  text: "42\n(according to the book The Hitchhiker's Guide to the Galaxy, by Douglas Adams)",
  url: nil}]

Timeout

receive 可設定 after 這個 timeout 機制

receive do
  :this_will_never_arrive -> :ok
after
  1_000 -> :timedout
end

修改 /lib/rumbl/infosys.ex 等待 backend 5000 ms


defmodule Rumbl.InfoSys do
  @backends [Rumbl.InfoSys.Wolfram]

  defmodule Result do
    defstruct score: 0, text: nil, url: nil, backend: nil
  end

  def start_link(backend, query, query_ref, owner, limit) do
    backend.start_link(query, query_ref, owner, limit)
  end

  def compute(query, opts \\ []) do
    limit = opts[:limit] || 10
    backends = opts[:backends] || @backends

    backends
    |> Enum.map(&spawn_query(&1, query, limit))
    |> await_results(opts)
    |> Enum.sort(&(&1.score >= &2.score))
    |> Enum.take(limit)
  end

  defp spawn_query(backend, query, limit) do
    query_ref = make_ref()
    opts = [backend, query, query_ref, self(), limit]
    {:ok, pid} = Supervisor.start_child(Rumbl.InfoSys.Supervisor, opts)
    monitor_ref = Process.monitor(pid)
    {pid, monitor_ref, query_ref}
  end

  defp await_results(children, opts) do
    timeout = opts[:timeout] || 5000
    timer = Process.send_after(self(), :timedout, timeout) 
    results = await_result(children, [], :infinity)
    cleanup(timer)
    results
  end

  defp await_result([head|tail], acc, timeout) do
    {pid, monitor_ref, query_ref} = head

    receive do
      {:results, ^query_ref, results} ->
        Process.demonitor(monitor_ref, [:flush])
        await_result(tail, results ++ acc, timeout)
      {:DOWN, ^monitor_ref, :process, ^pid, _reason} ->
        await_result(tail, acc, timeout)
      :timedout -> 
        kill(pid, monitor_ref)
        await_result(tail, acc, 0)
    after
      timeout ->
        kill(pid, monitor_ref)
        await_result(tail, acc, 0)
    end
  end

  defp await_result([], acc, _) do
    acc
  end

  defp kill(pid, ref) do 
    Process.demonitor(ref, [:flush])
    Process.exit(pid, :kill)
  end

  defp cleanup(timer) do  
    :erlang.cancel_timer(timer)
    receive do
      :timedout -> :ok
    after
      0 -> :ok
    end
  end
end

Integrating OTP Services with Channels

將剛剛的服務放到 VideoChannel 中

修改 /web/channels/video_channel.ex


defmodule Rumbl.VideoChannel do
  use Rumbl.Web, :channel
  alias Rumbl.AnnotationView

  def join("videos:" <> video_id, params, socket) do
    last_seen_id = params["last_seen_id"] || 0
    video_id = String.to_integer(video_id)
    video = Repo.get!(Rumbl.Video, video_id)

    annotations = Repo.all(
      from a in assoc(video, :annotations),
      where: a.id > ^last_seen_id,
      order_by: [asc: a.at, asc: a.id],
      limit: 200,
      preload: [:user]
    )
    resp = %{annotations: Phoenix.View.render_many(annotations, AnnotationView,
      "annotation.json")}

    {:ok, resp, assign(socket, :video_id, video.id)}
  end

  def handle_in(event, params, socket) do
    user = Repo.get(Rumbl.User, socket.assigns.user_id)
    handle_in(event, params, user, socket)
  end

  def handle_in("new_annotation", params, user, socket) do
    changeset =
      user
      |> build_assoc(:annotations, video_id: socket.assigns.video_id)
      |> Rumbl.Annotation.changeset(params)

    case Repo.insert(changeset) do
      {:ok, ann} ->
        broadcast_annotation(socket, ann)
        Task.start_link(fn -> compute_additional_info(ann, socket) end)
        {:reply, :ok, socket}

      {:error, changeset} ->
        {:reply, {:error, %{errors: changeset}}, socket}
    end
  end

  defp broadcast_annotation(socket, annotation) do
    annotation = Repo.preload(annotation, :user)
    rendered_ann = Phoenix.View.render(AnnotationView, "annotation.json", %{
      annotation: annotation
    })
    broadcast! socket, "new_annotation", rendered_ann
  end

  defp compute_additional_info(ann, socket) do
    for result <- Rumbl.InfoSys.compute(ann.body, limit: 1, timeout: 10_000) do
      attrs = %{url: result.url, body: result.text, at: ann.at}
      info_changeset =
        Repo.get_by!(Rumbl.User, username: result.backend)
        |> build_assoc(:annotations, video_id: ann.video_id)
        |> Rumbl.Annotation.changeset(attrs)

      case Repo.insert(info_changeset) do
        {:ok, info_ann} -> broadcast_annotation(socket, info_ann)
        {:error, _changeset} -> :ignore
      end
    end
  end
end

新增 /rumbl/priv/repo/backend_seeds.exs

alias Rumbl.Repo
alias Rumbl.User

Repo.insert!(%User{name: "Wolfram", username: "wolfram"})
$ mix run priv/repo/backend_seeds.exs
Compiling 19 files (.ex)
warning: String.to_char_list/1 is deprecated, use String.to_charlist/1
  lib/rumbl/info_sys/wolfram.ex:29

warning: function authenticate/2 is unused
  web/controllers/user_controller.ex:40

[debug] QUERY OK db=0.2ms queue=12.1ms
begin []
[debug] QUERY OK db=0.9ms
INSERT INTO `users` (`name`,`username`,`inserted_at`,`updated_at`) VALUES (?,?,?,?) ["Wolfram", "wolfram", {{2017, 9, 7}, {3, 55, 52, 165910}}, {{2017, 9, 7}, {3, 55, 52, 168676}}]
[debug] QUERY OK db=1.8ms
commit []

Inpsecting with Observer

erlang 有個 Observer 工具,可用這個方式啟動

:observer.start
iex -S mix

# 用這個方式啟動,可查看 Phoenix 部分的狀況
iex -S mix phoenix.server

supervision tree 是一個好的工具,可以觀察要怎麼將 application 分開。現在我們要將 application 分成兩個部分 :rumbl 及 :info_sys

利用 umbrella project 來處理

Using Umbrellas

每個 umbrella project 目錄包含以下的部分

  1. shared configuration of the project
  2. dependencies for the project
  3. apps 目錄 with child applications

要重新產生一個 umbrella project

$ mix new rumbrella --umbrella
* creating .gitignore
* creating README.md
* creating mix.exs
* creating apps
* creating config
* creating config/config.exs

Your umbrella project was created successfully.
Inside your project, you will find an apps/ directory
where you can create and host many apps:

    cd rumbrella
    cd apps
    mix new my_app

Commands like "mix compile" and "mix test" when executed
in the umbrella project root will automatically run
for each application in the apps/ directory.

將 InfoSys 移動到 rumbrella 下面

$ cd rumbrella/apps

$ mix new info_sys --sup
* creating README.md
* creating .gitignore
* creating mix.exs
* creating config
* creating config/config.exs
* creating lib
* creating lib/info_sys.ex
* creating lib/info_sys/application.ex
* creating test
* creating test/test_helper.exs
* creating test/info_sys_test.exs

Your Mix project was created successfully.
You can use "mix" to compile it, test it, and more:

    cd info_sys
    mix test

Run "mix help" for more commands.

將 InfoSys 由 rumbl 移到 info_sys

  1. 將 rumbl/lib/rumbl/infosys.ex Rumbl.InfoSys 改為 InfoSys 並複製到 infosys/lib/info_sys.ex ,前面加上

    use Application
    
    def start(_type, _args) do
    InfoSys.Supervisor.start_link()
    end
    
    @backends [InfoSys.Wolfram]
  2. 將 /rumbl/lib/rumbl/infosys/supervisor.ex Rumbl.InfoSys.Supervisor 改為 InfoSys.Supervisor 並複製到 infosys/lib/info_sys/supervisor.ex

  3. 將 /rumbl/lib/rumbl/infosys/wolfram.ex 的 Rumbl.InfoSys.Wolfram module 改為 InfoSys.Wolfram,並複製到 infosys/lib/info_sys/wolfram.ex

  4. 把所有 "Rumbl.InfoSys" 都改為 "InfoSys"

  5. 修改 lib/infosys/wolfram.ex,由 :rumbl 改為 :infosys

    defp app_id, do: Application.get_env(:info_sys, :wolfram)[:app_id]
  6. 修改 apps/info_sys/mix.exs

    def deps do
        [
        {:sweet_xml, "~> 0.5.0"}
        ]
    end
  7. 在 rumbrella 目錄執行 $ mix deps.get


將 rumbl 改為 umbrella child

  1. 將整個 rumbl 移動到 apps 目錄

  2. 修改 rumbrella/apps/rumbl/mix.exs,增加

      build_path: "../../_build",
      config_path: "../../config/config.exs",
      deps_path: "../../deps",
      lockfile: "../../mix.lock",
  3. 修改 applicaton ,增加 :info_sys

    def application do
        [mod: {Rumbl, []},
         applications: [:phoenix, :phoenix_pubsub, :phoenix_html, :cowboy, :logger, :gettext,
                        :phoenix_ecto, :mariaex, :comeonin, :info_sys]]
      end
  4. 更新 dependencies,移除 :sweetxml ,增加 :infosys

    {:info_sys, in_umbrella: true},
  5. 修改 rumbl/lib/rumbl.ex 移除 Rumbl.InfoSys

        children = [
          supervisor(Rumbl.Endpoint, []),
          supervisor(Rumbl.Repo, []),
        ]
  6. 修改 /web/channlel/videochannel.ex 的 computeadditional_info 將 Rumbl.InfoSys 改為 InfoSys

  7. 修改 cofnig/dev.secrets.exs 改為 :info_sys

    use Mix.Config
    config :info_sys, :wolfram, app_id: "LP93J3-XXXXXX"
  8. 修改 rumbl/package.json

      "dependencies": {
        "phoenix": "file:../../deps/phoenix",
        "phoenix_html": "file:../../deps/phoenix_html"
      },

在 rumbrella 目錄中

$ mix deps.get
$ mix test

References

Programming Phoenix

2018年10月29日

Phoenix_8_Channels

Phoenix channel 是一個 conversation,channel 會發送, 接收 messages,並保存 state,這些 messages 稱為 events,state 會存放在稱為 socket 的 struct 內。

conversation 就是 topic,就像是 chat room, local map , a game, or a video。超過一個人在同一時間對相同的主題有興趣,Channels 可用在多個 users 之間互通訊息。因為這是用 erlang isolated, dedicated process 實作的。

網頁的 request/response 是 stateless,但 conversation 是長時間運作的的 process,是 stateful 的。

Phoenix Clients with ES6

利用 ECMAScript 6 JavaScript 功能實作 client,ES6的code 可以 transpile 為 ES5,實作 client 對 video 新增 annotations,並發送給所有 users。

每一個 Phonenix conversation 是一個 Topic,因此要先確認,要以什麼為 Topic,目前是以 video 為 topic。

新增 /web/static/js/video.js

import Player from "./player"

let Video = {

  init(socket, element){ if(!element){ return }
    let playerId = element.getAttribute("data-player-id")
    let videoId  = element.getAttribute("data-id")
    socket.connect()
    Player.init(element.id, playerId, () => { 
      this.onReady(videoId, socket)
    })
  },

  onReady(videoId, socket){
    let msgContainer = document.getElementById("msg-container")
    let msgInput     = document.getElementById("msg-input")
    let postButton   = document.getElementById("msg-submit")
    let vidChannel   = socket.channel("videos:" + videoId)
    // TODO join the vidChannel
  }
}
export default Video

socket.connect() 可產生一個 websocket

注意 let vidChannel = socket.channel("videos:" + videoId) ,這是 ES6 Client 連接 Phoenix VideoChannel 的 channel。

Topic 需要一個 identifier,我們選用 "videos:" + videoId 這個格式,我們需要在 topic 內對其他相同 topic 的 users 發送 events。


修改 /web/static/js/app.js

import "phoenix_html"

import socket from "./socket"
import Video from "./video"

Video.init(socket, document.getElementById("video"))

如果瀏覽 http://localhost:4000/watch/2-elixir 時,js console 會出現這樣的錯誤訊息。

Unable to join – {reason: "unmatched topic"}

Preparing Our Server for the Channel

傳統 web request/response 每一次都會產生一個 connection,也就是 Plug.Conn,每個新的 request 都會有新的 conn,接下來用 pipeline 處理,最後 die。

channel 的流程跟上面的不同,client 會用 socket 建立 connection,在建立連線後,socket 會在整個 connection 的過程中持續被 transformed。socket 就是 client/server 之間持續運作的 conversation。

首先要決定是否能建立 connection,然後要產生 initial socket,包含所有 custom application setup。

修改 /web/static/js/socket.js

import {Socket} from "phoenix"

let socket = new Socket("/socket", {
  params: {token: window.userToken},
  logger: (kind, msg, data) => { console.log(`${kind}: ${msg}`, data) }
})

export default socket

let socket = new Socket("/socket".... 會讓 Phoenix 建立新的 socket。

查看 /lib/rumbl/endpoint.ex,有一個 /socket 的定義,UserSocket 就是處理 socket connection 的 module。

socket "/socket", Rumbl.UserSocket

先看一下 /web/channels/user_socket.ex 的內容

defmodule Rumbl.UserSocket do
  use Phoenix.Socket

  ## Transports
  transport :websocket, Phoenix.Transports.WebSocket
  # transport :longpoll, Phoenix.Transports.LongPoll

  def connect(_params, socket) do
    {:ok, socket}
  end

  def id(_socket), do: nil
end

UserSocket 使用 connection 處理所有 channel processes。 Phoenix 支援兩種 Transport protocols: websocket 或是 longpoll,也可以自訂一個。除了 transport 不同之外,其他的部分都是一樣的。

使用 shared socket abstraction,然後讓 Phoenix 處理其他的工作。

UserSocket 有兩個 functions: connect 及 id。id 是用來識別 socket,以便儲存不同的 state。目前 id 為 nil,connect 基本上就是接受所有連線。

接下來要利用 rumbl.Auth 增加 socket authentication。

如果再一次瀏覽網址 http://localhost:4000/watch/2-elixir,在 js console 就可看到此 debug message,表示已經連上 server。

transport: connected to ws://localhost:4000/socket/websocket?token=undefined&vsn=1.0.0

Creating the Channel

channel 就是 a conversation on a topic,topic 的 id 為 videos:video_id,我們希望 user 能取得某個 topic 的所有 events,也就是 video 的所有 annotations。

topic id 的一般形式就是 topic:subtopic,topic 為 resource name,subtopic 為 ID

因為 URL 就有參數,可以識別 conversation,也就是 :id


Joining a Channel

在 /web/channels/user_socket.ex 增加一行

channel "videos:*", Rumbl.VideoChannel

videos:* conversation 以 resource name 及 ID 作為 topic 的分類方式


Building the Channel Module

新增 /web/channels/video_channel.ex

defmodule Rumbl.VideoChannel do
  use Rumbl.Web, :channel

  def join("videos:" <> video_id, _params, socket) do
    {:ok, assign(socket, :video_id, String.to_integer(video_id))}
  end
end

channel 的第一個 callback 就是 join,clients 可 join topics on a channel,如果成功就回傳 {:ok, socket},拒絕連線就回傳 {:error, socket}

現在先讓所有 socket 可任意 join video topics,並新增 video ID (由 topic 取得) 到 socket.assigns。socket 會在 socket.assigns 儲存某個 conversation 的所有狀態 state。

socket 會被 transformed 為 loop,而不是一連串的 pipelines。當 events 進出 channel 時,可同時存取 socket state。

修改 /web/static/js/video.js

import Player from "./player"

let Video = {

    init(socket, element){ if(!element){ return }
        let playerId = element.getAttribute("data-player-id")
        let videoId  = element.getAttribute("data-id")
        socket.connect()
        Player.init(element.id, playerId, () => {
            this.onReady(videoId, socket)
    })
    },

    onReady(videoId, socket){
        let msgContainer = document.getElementById("msg-container")
        let msgInput     = document.getElementById("msg-input")
        let postButton   = document.getElementById("msg-submit")
        
        // 以 "videos:" videoId 產生新的 channel object
        let vidChannel   = socket.channel("videos:" + videoId)

        vidChannel.join()
            .receive("ok", resp => console.log("joined the video channel", resp) )
    .receive("error", reason => console.log("join failed", reason) )
    }
}
export default Video

如果再一次瀏覽網址 http://localhost:4000/watch/2-elixir,在 js console 就可看到

[Log] push: videos:2 phx_join (1) – {} (app.js, line 1586)
[Log] receive: ok videos:2 phx_reply (1) – {status: "ok", response: {}} (app.js, line 1586)
[Log] joined the video channel – {}

server 的 console log 為

[info] JOIN videos:2 to Rumbl.VideoChannel
  Transport:  Phoenix.Transports.WebSocket
  Parameters: %{}
[info] Replied videos:2 :ok

Sending and Receiving Event

在 channel 收到的訊息為 event name + payload + arbitrary data

channel 有三種接收訊息的方式:

  1. handle_in

    receives direct channel events

  2. handle_out

    intercepts broadcast events

  3. handle_info

    receives OTP messages


Taking Channels for a Trial run

目前是讓 join function 每 5 seconds 就發送一次 :ping message 到 channel

修改 /web/channels/video_channel.ex

defmodule Rumbl.VideoChannel do
  use Rumbl.Web, :channel

  def join("videos:" <> video_id, _params, socket) do
    :timer.send_interval(5_000, :ping)
    {:ok, socket}
  end

  # 當 elixir message 到達 channel 就會呼叫 handle_info
  # 目前每收到一次就將 :count +1
  def handle_info(:ping, socket) do
    count = socket.assigns[:count] || 1
    push socket, "ping", %{count: count}

     # :noreply 代表不發送 reply,並將 transformed 後的 socket 回傳回去
    {:noreply, assign(socket, :count, count + 1)}
  end
end

client 要對應修改 video.js,增加 vidChannel.on("ping", ({count}) => console.log("PING", count) )

    let vidChannel   = socket.channel("videos:" + videoId)

    vidChannel.on("ping", ({count}) => console.log("PING", count) )

    vidChannel.join()
      .receive("ok", resp => console.log("joined the video channel", resp) )
      .receive("error", reason => console.log("join failed", reason) )

js console 會持續看到

[Log] receive:  videos:2 ping  – {count: 1} (app.js, line 1586)
[Log] receive:  videos:2 ping  – {count: 2} (app.js, line 1586)
[Log] receive:  videos:2 ping  – {count: 3} (app.js, line 1586)
[Log] receive:  videos:2 ping  – {count: 4} (app.js, line 1586)

handle_info 就是 loop

client js 的部分是以 vidChannel.on(event, callback) 處理訊息

後面會看到怎麼用 handle_in 處理 synchronous messaging

controller 處理 request 而 channels hold a conversation


Annotating Videos

需要一個 Annotation model 儲存 user annotations

修改 /web/static/js/video.js

import Player from "./player"

let Video = {

  init(socket, element){ if(!element){ return }
    let playerId = element.getAttribute("data-player-id")
    let videoId  = element.getAttribute("data-id")
    socket.connect()
    Player.init(element.id, playerId, () => {
      this.onReady(videoId, socket)
    })
  },

  onReady(videoId, socket){
    let msgContainer = document.getElementById("msg-container")
    let msgInput     = document.getElementById("msg-input")
    let postButton   = document.getElementById("msg-submit")
    let vidChannel   = socket.channel("videos:" + videoId)

     // 處理 post 按鈕的 click event
     // 利用 vidChannel.push 發送 new_annotation
    postButton.addEventListener("click", e => {
      let payload = {body: msgInput.value, at: Player.getCurrentTime()}
      vidChannel.push("new_annotation", payload)        
                .receive("error", e => console.log(e) ) 
      msgInput.value = ""
    })

    // 接收 server 發送的 new_annotation,把 annotation 顯示在畫面 msgContainer 上
    vidChannel.on("new_annotation", (resp) => {         
      this.renderAnnotation(msgContainer, resp)
    })

    vidChannel.join()
      .receive("ok", resp => console.log("joined the video channel", resp) )
      .receive("error", reason => console.log("join failed", reason) )
  },

  // safely escape user input,可避免發生 XSS attack
  esc(str){ 
    let div = document.createElement("div")
    div.appendChild(document.createTextNode(str))
    return div.innerHTML
  },

  renderAnnotation(msgContainer, {user, body, at}){ 
    let template = document.createElement("div")

    template.innerHTML = `
    <a href="#" data-seek="${this.esc(at)}">
      <b>${this.esc(user.username)}</b>: ${this.esc(body)}
    </a>
    `
    msgContainer.appendChild(template)
    msgContainer.scrollTop = msgContainer.scrollHeight
  }
}
export default Video

Adding Annotation on the Server

修改 /web/channels/video_channel.ex

defmodule Rumbl.VideoChannel do
  use Rumbl.Web, :channel

  def join("videos:" <> video_id, _params, socket) do
    {:ok, socket}
  end

  # 處理 new_annotation,並 broadcast! 給目前 topic 的所有 users
  # broadcast! 有三個參數  socket, name of the event, payload (任意的 map)
  def handle_in("new_annotation", params, socket) do
    broadcast! socket, "new_annotation", %{
      user: %{username: "anon"},
      body: params["body"],
      at: params["at"]
    }

    # :reply 有兩種 :ok 或是 :error
    # 不然就是用 :noreply
    {:reply, :ok, socket}
  end
end

注意: 將原始的 message payload 直接轉送給其他人,而沒有 inspection,可能會有 security 問題

如果這樣寫,就有可能有資安問題

broadcast! socket, "new_annotation", Map.put(params, "user", %{
username: "anon"
})

現在打開兩個網頁,就可以互傳訊息

Socket Authentication

因為 channel 是 long-duration connection,利用 Phoenix.Token token authentication,可為每個 user 指定一個 unique token

不使用 session cookie 的原因是,可能會有 cross-domain attack。

因為已經有利用 Rumbl.Auth plug 增加的 current_user,現在要做的是利用 authenticated user 產生 token 並傳給 socket 前端。

首先修改 /web/templates/layout/app.html.eex,將 userToken 由 layout assigns 取出並放在 browser window 中

    </div> <!-- /container -->
    <script>window.userToken = "<%= assigns[:user_token] %>"</script>
    <script src="<%= static_path(@conn, "/js/app.js") %>"></script>

修改 /web/controllers/auth.ex

  def call(conn, repo) do
    user_id = get_session(conn, :user_id)

    cond do
      user = conn.assigns[:current_user] ->
        put_current_user(conn, user) 
      user = user_id && repo.get(Rumbl.User, user_id) ->
        put_current_user(conn, user)
      true ->
        assign(conn, :current_user, nil)
    end
  end

  def login(conn, user) do
    conn
    |> put_current_user(user) 
    |> put_session(:user_id, user.id)
    |> configure_session(renew: true)
  end

  # 將 current_user 及 user_token 放到 conn.assigns
  defp put_current_user(conn, user) do
    token = Phoenix.Token.sign(conn, "user socket", user.id)

    conn
    |> assign(:current_user, user)
    |> assign(:user_token, token)
  end

修改 /web/static/js/socket.js ,將 user token 傳入 socket.connect,並在 UserSocket.connect callback 中驗證 token。

import {Socket} from "phoenix"

let socket = new Socket("/socket", {
  // :params 會出現在 UserSocket.connect 的第一個參數
  params: {token: window.userToken},
  logger: (kind, msg, data) => { console.log(`${kind}: ${msg}`, data) }
})

export default socket

修改 /web/channles/user_socket.ex

defmodule Rumbl.UserSocket do
  use Phoenix.Socket

  ## Channels
  channel "videos:*", Rumbl.VideoChannel

  ## Transports
  transport :websocket, Phoenix.Transports.WebSocket
  # transport :longpoll, Phoenix.Transports.LongPoll

  # 2 weeks
  @max_age 2 * 7 * 24 * 60 * 60

  def connect(%{"token" => token}, socket) do
    case Phoenix.Token.verify(socket, "user socket", token, max_age: @max_age) do
      {:ok, user_id} ->
        {:ok, assign(socket, :user_id, user_id)}
      {:error, _reason} ->
        :error
    end
  end
  def connect(_params, _socket), do: :error

  def id(socket), do: "users_socket:#{socket.assigns.user_id}"
end

利用 Phoenix.Token.verify 檢查 token,可設定 max_age

如果 token 正確,就會收到 user_id 並存在 socket.assigns,回傳 {:ok, socket} 用以建立 connection。token 錯誤,就 return :error

refresh your page,application 還是能正常運作,但已經有了 user authentication

Persisting Annotations

建立 Annotation model,create annotations on videos,每個 annotation 會 belong to a user and a video

$ mix phoenix.gen.model Annotation annotations body:text at:integer user_id:references:users video_id:references:videos
* creating web/models/annotation.ex
* creating test/models/annotation_test.exs
* creating priv/repo/migrations/20170906163920_create_annotation.exs

Remember to update your repository by running migrations:

    $ mix ecto.migrate


$ mix ecto.migrate
Compiling 1 file (.ex)
Generated rumbl app
[info] == Running Rumbl.Repo.Migrations.CreateAnnotation.change/0 forward
[info] create table annotations
[info] create index annotations_user_id_index
[info] create index annotations_video_id_index
[info] == Migrated in 0.0s

還要處理 User, Video 的 relationships

修改 /web/models/user.ex 及 /web/models/video.ex,增加 has_many

has_many :annotations, Rumbl.Annotation

回到 /web/channels/video_channel.ex

defmodule Rumbl.VideoChannel do
  use Rumbl.Web, :channel

  def join("videos:" <> video_id, _params, socket) do
    {:ok, assign(socket, :video_id, String.to_integer(video_id))}
  end

  # 確保所有 events 都會有 current_user,然後再呼叫其他 handle_in
  def handle_in(event, params, socket) do 
    user = Repo.get(Rumbl.User, socket.assigns.user_id)
    handle_in(event, params, user, socket)
  end

  def handle_in("new_annotation", params, user, socket) do 
    # 以 new_annotation 產生 changeset 並透過 Repo 存到 DB
    changeset =
      user
      |> build_assoc(:annotations, video_id: socket.assigns.video_id)
      |> Rumbl.Annotation.changeset(params)

    case Repo.insert(changeset) do
      # insert 成功,才 broadcast 給所有 subscribers
      # 也可以用  {:noreply, socket} 不送 reply
      {:ok, annotation} ->
        broadcast! socket, "new_annotation", %{
          id: annotation.id,
          user: Rumbl.UserView.render("user.json", %{user: user}), 
          body: annotation.body,
          at: annotation.at
        }
        {:reply, :ok, socket}

      {:error, changeset} ->
        {:reply, {:error, %{errors: changeset}}, socket}
    end
  end
end

因為也需要 notify subscribers 該 user 的資訊,在 UserView 新增 user.json template

defmodule Rumbl.UserView do
  use Rumbl.Web, :view
  alias Rumbl.User

  def first_name(%User{name: name}) do
    name
    |> String.split(" ")
    |> Enum.at(0)
  end

  def render("user.json", %{user: user}) do
    %{id: user.id, username: user.username}
  end
end

現在新增 annotation, server log 就會出現

INSERT INTO `annotations` (`at`,`body`,`user_id`,`video_id`,`inserted_at`,`updated_at`) VALUES (?,?,?,?,?,?) [0, "test", 1, 2, {{2017, 9, 6}, {16, 49, 10, 444088}}, {{2017, 9, 6}, {16, 49, 10, 447017}}]
[debug] QUERY OK db=0.6ms

在 refresh page 後,annotations 就會消失,所以在 user join channel 時,要把 messages 送給 client

修改 /web/channels/video_channel.ex,改寫 join,取得 video's annotations

defmodule Rumbl.VideoChannel do
  use Rumbl.Web, :channel
  alias Rumbl.AnnotationView

  def join("videos:" <> video_id, _params, socket) do
    video_id = String.to_integer(video_id)
    video = Repo.get!(Rumbl.Video, video_id)

    # 要 preload user
    annotations = Repo.all(
      from a in assoc(video, :annotations),
        order_by: [asc: a.at, asc: a.id],
        limit: 200,
        preload: [:user]
    )

    resp = %{annotations: Phoenix.View.render_many(annotations, AnnotationView,
                                                   "annotation.json")}
    {:ok, resp, assign(socket, :video_id, video_id)}
  end

  def handle_in("new_annotation", params, socket) do
    user = Rumbl.Repo.get(Rumbl.User, socket.assigns.user_id)

    changeset =
      user
      |> build_assoc(:annotations, video_id: socket.assigns.video_id)
      |> Rumbl.Annotation.changeset(params)

    case Repo.insert(changeset) do
      {:ok, annotation} ->
        broadcast! socket, "new_annotation", %{
          id: annotation.id,
          user: Rumbl.UserView.render("user.json", %{user: user}),
          body: annotation.body,
          at: annotation.at
        }
        {:reply, :ok, socket}

      {:error, changeset} ->
        {:reply, {:error, %{errors: changeset}}, socket}
    end
  end
end

Phoenix.View.render_many 能夠 collects the render results for all elements in the enumerable passed to it

新增 /web/views/annotation_view.ex

defmodule Rumbl.AnnotationView do
  use Rumbl.Web, :view

  def render("annotation.json", %{annotation: ann}) do
    %{
      id: ann.id,
      body: ann.body,
      at: ann.at,
      user: render_one(ann.user, Rumbl.UserView, "user.json")
    }
  end
end

注意 annotaion's user 的 render_one,可處理 nil results

更新 vidChannle.join() 以便 render list of annotations on join

    vidChannel.join()
      .receive("ok", ({annotations}) => {
        annotations.forEach( ann => this.renderAnnotation(msgContainer, ann) )
      })
      .receive("error", reason => console.log("join failed", reason) )

現在 reload 頁面就能看到所有 annotations


現在我們需要 schedule the annotations to appear synced up with the video playback

更新 /web/static/js/video.js


import Player from "./player"

let Video = {

    init(socket, element){ if(!element){ return }
        let playerId = element.getAttribute("data-player-id")
        let videoId  = element.getAttribute("data-id")
        socket.connect()
        Player.init(element.id, playerId, () => {
            this.onReady(videoId, socket)
    })
    },

    onReady(videoId, socket){
        let msgContainer = document.getElementById("msg-container")
        let msgInput     = document.getElementById("msg-input")
        let postButton   = document.getElementById("msg-submit")
        let vidChannel   = socket.channel("videos:" + videoId)

        postButton.addEventListener("click", e => {
            let payload = {body: msgInput.value, at: Player.getCurrentTime()}
            vidChannel.push("new_annotation", payload)
            .receive("error", e => console.log(e) )
        msgInput.value = ""
    })

        msgContainer.addEventListener("click", e => {
            e.preventDefault()
        let seconds = e.target.getAttribute("data-seek") ||
            e.target.parentNode.getAttribute("data-seek")
        if(!seconds){ return }

        Player.seekTo(seconds)
    })

        vidChannel.on("new_annotation", (resp) => {
            this.renderAnnotation(msgContainer, resp)
    })

        vidChannel.join()
            .receive("ok", resp => {
            this.scheduleMessages(msgContainer, resp.annotations)
    })
    .receive("error", reason => console.log("join failed", reason) )
    },

    renderAnnotation(msgContainer, {user, body, at}){
        let template = document.createElement("div")
        template.innerHTML = `
    <a href="#" data-seek="${this.esc(at)}">
      [${this.formatTime(at)}]
      <b>${this.esc(user.username)}</b>: ${this.esc(body)}
    </a>
    `
        msgContainer.appendChild(template)
        msgContainer.scrollTop = msgContainer.scrollHeight
    },

    scheduleMessages(msgContainer, annotations){
        setTimeout(() => {
            let ctime = Player.getCurrentTime()
            let remaining = this.renderAtTime(annotations, ctime, msgContainer)
            this.scheduleMessages(msgContainer, remaining)
    }, 1000)
    },

    renderAtTime(annotations, seconds, msgContainer){
        return annotations.filter( ann => {
                if(ann.at > seconds){
            return true
        } else {
            this.renderAnnotation(msgContainer, ann)
            return false
        }
    })
    },

    formatTime(at){
        let date = new Date(null)
        date.setSeconds(at / 1000)
        return date.toISOString().substr(14, 5)
    },

    esc(str){
        let div = document.createElement("div")
        div.appendChild(document.createTextNode(str))
        return div.innerHTML
    }
}
export default Video

根據 current player time 去 render annotations

scheduleMessages 每秒都會執行一次,每次都呼叫 renderAtTime

renderAtTime 會 filter 要 render 的 messages

現在再次 reload 頁面,就可看到 annotation 的時間


增加讓 annotation 可以點擊的功能,就可以直接跳躍影片到該 annotaion 產生的時間

修改 /web/static/js/video.js

    msgContainer.addEventListener("click", e => {
      e.preventDefault()
      let seconds = e.target.getAttribute("data-seek") ||
                    e.target.parentNode.getAttribute("data-seek")
      if(!seconds){ return }

      Player.seekTo(seconds)
    })

Handling Disconnects

JS client 可斷線再 reconnect,Server 可能會 restart,或是網路可能發生問題,這些問題都會造成斷線。

如果發送一個 annotation,然後馬上把 server 關掉,client 會以 exponential back-off 的方式進行 reconnect。重新啟動 server,會發現 server 會認為是新的連線,然後發送所有的 annotations,client 會出現重複的 annotations。client 必須偵測 duplicate annotations 並忽略處理。

我們可以在 client 追蹤 lastseenid,並在每次收到新的 annotation 時更新這個值。

當 client 重連時,可將 lastseenid 發送給 server,server 就只需要發送未收到的訊息。

修改 /web/static/js/video.js,增加 vidChannel.params.lastseenid

        vidChannel.on("new_annotation", (resp) => {
            vidChannel.params.last_seen_id = resp.id
        this.renderAnnotation(msgContainer, resp)
    })

        vidChannel.join()
            .receive("ok", resp => {
            let ids = resp.annotations.map(ann => ann.id)
        if(ids.length > 0){ vidChannel.params.last_seen_id = Math.max(...ids) }
        this.scheduleMessages(msgContainer, resp.annotations)
    })
    .receive("error", reason => console.log("join failed", reason) )

client 的 channel 會儲存 params 物件,並在每次 join 時,發送給 server。在 join 也要更新這個參數。

修改 /web/channels/video_channel.ex 的 join

  def join("videos:" <> video_id, params, socket) do
    last_seen_id = params["last_seen_id"] || 0
    video_id = String.to_integer(video_id)
    video = Repo.get!(Rumbl.Video, video_id)

    annotations = Repo.all(
      from a in assoc(video, :annotations),
      where: a.id > ^last_seen_id,
      order_by: [asc: a.at, asc: a.id],
      limit: 200,
      preload: [:user]
    )

    resp = %{annotations: Phoenix.View.render_many(annotations, AnnotationView,
      "annotation.json")}
    {:ok, resp, assign(socket, :video_id, video_id)}
  end

References

Programming Phoenix