I’m trying to stream a file to all connected websockets so all of them will receive the same line from the text file “test.txt”
in init/3 I create a process that reads a file and send a line each second to all subscribers
in the client side I just console.log all received messages
defmodule WebsocketHandler do
@behaviour :cowboy_websocket_handler
@fpath "./test.txt"
def init({_tcp, _http}, _req, _opts) do
IO.puts "init({tcp, http}, _req, _opts)"
{:ok, _pid} = PubSub.start_link()
spawn fn ->
stream_from_file(@fpath, 10)
end
{:upgrade, :protocol, :cowboy_websocket}
end
def websocket_init(_TransportName, req, _opts) do
IO.puts "init. Starting timer. PID is #{inspect(self())}"
:erlang.start_timer(1000, self(), [])
:ok = PubSub.subscribe(spawn(fn -> loop(req) end), :radio)
{:ok, req, :undefined_state }
end
def websocket_handle({:text, content}, req, state) do
{ :ok, %{ "message" => message} } = JSEX.decode(content)
{ :ok, reply } = JSEX.encode(%{ reply: String.reverse(message)})
IO.puts("Message: #{message} [ #{state} ]")
{:reply, {:text, reply}, req, state}
end
def websocket_handle(_data, req, state) do
IO.puts "websocket_handle(_data, req, state)"
{:ok, req, state}
end
def websocket_info({:info, info}, req, state) do
IO.puts "websocket_info({:info, info}, req, state) #{info}"
{ :ok, message } = JSEX.encode(%{ radio: info})
{ :reply, {:text, message}, req, state}
end
def websocket_info(_info, req, state) do
IO.puts "websocket_info(_info, req, state)"
:erlang.start_timer(1000, self(), [])
{ :ok, message } = JSEX.encode(%{ radio: "yo"}) # TODO send lines here
{ :reply, {:text, message}, req, state}
end
def websocket_terminate(_reason, _req, _state) do
:ok
end
defp loop(req) do
receive do
{:radio_data, data} ->
IO.puts "received [ #{data} ]"
websocket_info({:info, data}, req, :undefined_state)
loop(req)
end
end
defp stream_from_file(fpath, bytes) do
File.stream!(fpath, [], bytes)
|> Enum.each(fn chunk ->
PubSub.publish(:radio, {:radio_data, chunk})
:timer.sleep(1_000)
end)
stream_from_file(fpath, bytes)
end
end
explicit invoking websocket_info/3 in loop/1 doesn’t work although the log written
IO.puts "received [ #{data} ]" # this line runs
websocket_info({:info, data}, req, :undefined_state) # doesn't send to socket
deps : [{:cowboy, “~> 1.0.0” }, {:jsex, “~> 2.0.0” }, {:plug, “~> 1.0”}, {:pubsub, “~> 0.0.2”}]
