KAL
Menu

PubSub

Broadcast pesan lintas process dan node.

Daftar isi

Phoenix.PubSub menyebarkan satu pesan kepada sekumpulan process yang berlangganan kanal yang sama, tanpa pengirim perlu mengetahui pid process mana pun. Topik ini membahas kanal dan langganan pada PubSub yang berdiri sendiri, integrasinya dengan LiveView melalui handle_info/2, siaran lintas node, serta satu error langganan yang sering muncul. Hasil akhirnya adalah kemampuan menghubungkan halaman-halaman LiveView yang terpisah, misalnya memperbarui papan kunjungan di semua browser ketika satu kunjungan baru dicatat.

Kanal dan langganan#

Model PubSub berbentuk publish-subscribe, yaitu pengirim menyiarkan pesan (broadcast) ke sebuah kanal, sedangkan process yang berlangganan (subscribe) kanal itu menerima pesannya di mailbox masing-masing. Kanal hanya berupa string, misalnya "perpustakaan", dan satu process dapat berlangganan banyak kanal sekaligus. Tidak ada daftar penerima yang ditulis pengirim, karena pengiriman dilakukan oleh server PubSub kepada seluruh pelanggan kanal.

Server PubSub adalah process yang diberi nama, misalnya :demo, dan semua fungsi Phoenix.PubSub selalu menyebut nama itu sebagai argumen pertama. Pada aplikasi Phoenix, server PubSub sudah dimulai oleh pohon supervision aplikasi, sehingga kode halaman tinggal memanggil subscribe/2 dan broadcast!/3. Pesan siaran dapat berupa term Elixir apa pun, dan tuple dengan atom sebagai elemen pertamanya tetap menjadi konvensi yang baik, sama seperti pesan process biasa pada modul Process.

Keluarga fungsi siaran menawarkan beberapa varian yang wajib dibedakan. broadcast/3 mengembalikan :ok atau {:error, alasan}, sedangkan broadcast!/3 adalah versi yang melempar error ketika siaran gagal, sehingga kegagalan langsung terlihat alih-alih tersimpan sebagai nilai balikan yang tidak dibaca. local_broadcast/3 membatasi siaran pada node tempat pemanggilnya berjalan, sedangkan direct_broadcast!/4 menyasar satu node tertentu, dan keduanya berguna ketika cakupan pesan memang ingin diatur secara eksplisit.

Skrip utuh: kanal perpustakaan#

Skrip berikut memulai PubSub bernama :demo, membuat process utama berlangganan kanal "perpustakaan", lalu menyiarkan satu pesan dari process lain. Penerimaan pesan dilakukan dengan receive yang diberi batas after, sehingga skrip tetap berakhir apabila pesan tidak pernah tiba:

elixir kanal.exs 
# kanal.exs — langganan dan broadcast pada Phoenix.PubSub yang berdiri sendiri
Mix.install([{:phoenix_pubsub, "~> 2.1"}])
Logger.configure(level: :warning)

# pada aplikasi Phoenix, baris ini dijalankan otomatis oleh application.ex
{:ok, _} = Supervisor.start_link([{Phoenix.PubSub, name: :demo}], strategy: :one_for_one)

# process ini berlangganan kanal "perpustakaan"
:ok = Phoenix.PubSub.subscribe(:demo, "perpustakaan")

# pesan dikirim dari process lain, seperti LiveView lain yang menerima kunjungan
spawn(fn ->
  Phoenix.PubSub.broadcast!(:demo, "perpustakaan", {:kunjungan, "Budi"})
end)

receive do
  {:kunjungan, nama} -> IO.inspect({:kunjungan, nama}, label: "pesan tiba")
after
  5000 -> IO.puts("waktu habis")
end
text Keluaran 
pesan tiba: {:kunjungan, "Budi"}

Keluaran menunjukkan rantai lengkap satu siaran. Process hasil spawn memanggil broadcast!/3 tanpa mengenal process penerima, server PubSub mencari seluruh pelanggan kanal "perpustakaan", dan process utama menerima pesannya di mailbox lalu membacanya dengan receive. Jika kanal itu memiliki sepuluh pelanggan, sepuluh process menerima salinan yang sama, dan tugas siaran itu selesai tanpa satu pun perulangan send yang ditulis manual. Skripnya dapat dijalankan ulang dari terminal:

bash 
# menjalankan skrip kanal dari folder topik
elixir kanal.exs
text Keluaran 
pesan tiba: {:kunjungan, "Budi"}

LiveView menerima siaran lewat handle_info#

Koneksi antara PubSub dan LiveView adalah handle_info/2. Halaman berlangganan satu kanal pada saat mount, dan setiap siaran yang tiba diproses sebagai pesan info, lalu perubahan assigns yang dibuat callback itu dikirim ke browser sebagai patch. Halaman yang berlangganan kanal yang sama dengan halaman lain otomatis diperbarui tanpa perlu mengetahui keberadaan satu sama lain.

Skrip berikut memuat LiveView papan kunjungan yang berlangganan kanal "kunjungan", menerima satu siaran, lalu merender HTML sebelum dan sesudah siaran itu diproses:

elixir papan.exs 
# papan.exs — LiveView menerima broadcast PubSub melalui handle_info
Mix.install([{:phoenix_live_view, "~> 1.1"}, {:jason, "~> 1.4"}])
Logger.configure(level: :warning)

defmodule PapanLive do
  use Phoenix.LiveView

  def mount(_params, _session, socket) do
    if connected?(socket), do: Phoenix.PubSub.subscribe(:papan, "kunjungan")
    {:ok, assign(socket, :terakhir, "belum ada")}
  end

  def handle_info({:kunjungan, nama}, socket) do
    {:noreply, assign(socket, :terakhir, nama)}
  end

  def render(assigns) do
    ~H"""
    <p>Kunjungan terakhir: {@terakhir}</p>
    """
  end
end

{:ok, _} = Supervisor.start_link([{Phoenix.PubSub, name: :papan}], strategy: :one_for_one)
Code.ensure_loaded(Phoenix.LiveViewTest)

{:ok, socket} = PapanLive.mount(%{}, %{}, struct(Phoenix.LiveView.Socket))

# connected? bernilai false pada socket tiruan sehingga langganan di mount terlewati,
# maka langganan dilakukan manual satu kali di sini
Phoenix.PubSub.subscribe(:papan, "kunjungan")
IO.puts(Phoenix.LiveViewTest.rendered_to_string(PapanLive.render(socket.assigns)))

# process lain (misalnya LiveView ruangan lain) menyiarkan kunjungan baru
Phoenix.PubSub.broadcast!(:papan, "kunjungan", {:kunjungan, "Dewi"})

# process utama berperan sebagai process LiveView: membaca pesan lalu meneruskannya
pesan =
  receive do
    msg -> msg
  after
    5000 -> :waktu_habis
  end

IO.inspect(pesan, label: "pesan di mailbox")
{:noreply, socket2} = PapanLive.handle_info(pesan, socket)
IO.puts(Phoenix.LiveViewTest.rendered_to_string(PapanLive.render(socket2.assigns)))
text Keluaran 
<p>Kunjungan terakhir: belum ada</p>
pesan di mailbox: {:kunjungan, "Dewi"}
<p>Kunjungan terakhir: Dewi</p>

Baris pertama berasal dari assigns awal, sedangkan baris terakhir berasal dari assigns yang disetel oleh handle_info/2 setelah siaran dibaca dari mailbox. Baris tengah memperlihatkan isi pesannya apa adanya, yaitu tuple {:kunjungan, "Dewi"}. Pada proyek nyata, pemisahan peran itu tidak berubah, dan satu-satunya perbedaan adalah nama server PubSub yang berasal dari aplikasi:

elixir 
# tidak-dijalankan
# potongan proyek nyata: langganan di mount kedua dan siaran dari sebuah context
defmodule RuanganLive do
  use MyAppWeb, :live_view

  def mount(_params, _session, socket) do
    if connected?(socket), do: Phoenix.PubSub.subscribe(MyApp.PubSub, "ruangan")
    {:ok, assign(socket, :terakhir, "belum ada")}
  end

  def handle_info({:kunjungan, nama}, socket) do
    {:noreply, assign(socket, :terakhir, nama)}
  end
end

# di dalam modul context, misalnya PenerimaanTamu.catat/2
# Phoenix.PubSub.broadcast(MyApp.PubSub, "ruangan", {:kunjungan, nama})

Skrip papan ini juga dapat dijalankan ulang dari terminal:

bash 
# menjalankan skrip papan dari folder topik
elixir papan.exs
text Keluaran 
<p>Kunjungan terakhir: belum ada</p>
pesan di mailbox: {:kunjungan, "Dewi"}
<p>Kunjungan terakhir: Dewi</p>

Siaran lintas node#

Adapter bawaan bernama Phoenix.PubSub.PG2 menggunakan registri process Erlang, sehingga node-node yang terhubung dalam satu klaster saling berbagi langganan. Siaran biasa menyebar ke seluruh node yang tergabung, sedangkan direct_broadcast!/4 menyasar node tertentu secara eksplisit dengan namanya. Konsekuensinya, dua halaman LiveView yang dijalankan dua server berbeda tetap dapat saling memperbarui, sebab langganan dan siaran tidak lagi terbatas pada satu mesin.

Penggabungan node adalah pengaturan infrastruktur, bukan perubahan kode. Apabila klaster tidak dibentuk, siaran tetap bekerja dengan benar pada satu node, dan perilakunya sama seperti pada contoh skrip di atas.

Kesalahan umum#

Kesalahan pertama adalah berlangganan ke nama PubSub yang belum dimulai. Fungsi subscribe/2 mencari process server dengan nama yang diberikan, dan nama yang tidak terdaftar memicu error berikut:

elixir 
# langganan ke PubSub yang belum berjalan
Mix.install([{:phoenix_pubsub, "~> 2.1"}])
Logger.configure(level: :warning)

try do
  Phoenix.PubSub.subscribe(:tidak_ada, "perpustakaan")
rescue
  e in ArgumentError -> IO.puts("ArgumentError: #{Exception.message(e)}")
end
text Keluaran 
ArgumentError: unknown registry: :tidak_ada

Perbaikannya ada di sisi pohon supervision, yaitu memastikan entri {Phoenix.PubSub, name: ...} benar-benar ada pada application.ex dan namanya sama dengan yang dipanggil oleh kode halaman.

Kesalahan kedua adalah mengirim siaran yang tidak memiliki klausa penerima. Pesan yang tidak cocok dengan handle_info/2 mana pun membuat process halaman berhenti, dan halaman pada browser ikut berhenti tanpa pesan error apa pun. Kanal yang dilanggani sebaiknya dipasangkan dengan pola pesan yang jelas dan satu klausa penangkap sisa pesan, sehingga pesan asing tercatat alih-alih menghentikan halaman.

Kesalahan ketiga adalah mengira siaran sampai ke browser tanpa perubahan assigns. broadcast!/3 hanya mengirim pesan ke process, sedangkan perubahan pada browser baru terjadi apabila handle_info/2 mengembalikan socket dengan assigns baru. Callback yang menerima pesan tanpa mengubah assigns tidak menghasilkan patch apa pun, dan halaman tampak tidak bergerak.

Ringkasan#

  • Phoenix.PubSub menyebarkan pesan dari satu pengirim ke seluruh process yang berlangganan kanal yang sama, dengan server PubSub yang dikenali melalui namanya.
  • Halaman LiveView berlangganan pada mount yang terhubung, menerima siaran lewat handle_info/2, dan memperbarui browser hanya lewat assigns yang dikembalikan callback itu.
  • Registri PG2 membuat langganan menyebar lintas node dalam satu klaster, sedangkan direct_broadcast!/4 menyasar satu node tertentu.
  • Langganan ke nama PubSub yang belum dimulai memicu unknown registry, dan pesan tanpa klausa handle_info/2 yang cocok menghentikan process halaman.

Referensi#