feat(server): extend connection with handle_info and pump
This commit is contained in:
1 parent
d29a606c88
commit
fd5356e789
1 file changed
+23
-8
@@ -11,7 +11,7 @@ defmodule Excache.Server.Connection do
|
|||||||
|
|
||||||
@impl true
|
@impl true
|
||||||
def init(socket) do
|
def init(socket) do
|
||||||
base = %{socket: socket, buffer: "", subscriptions: MapSet.new()}
|
base = %{socket: socket, buffer: "", subscriptions: MapSet.new(), blocked: nil}
|
||||||
{:ok, Map.merge(base, Transaction.initial_state())}
|
{:ok, Map.merge(base, Transaction.initial_state())}
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -22,16 +22,16 @@ defmodule Excache.Server.Connection do
|
|||||||
end
|
end
|
||||||
|
|
||||||
def handle_info({:tcp, socket, data}, state) do
|
def handle_info({:tcp, socket, data}, state) do
|
||||||
{output, state} = consume(state.buffer <> data, %{state | buffer: ""})
|
state = pump(%{state | buffer: state.buffer <> data})
|
||||||
|
|
||||||
unless output == [] do
|
|
||||||
:gen_tcp.send(socket, output)
|
|
||||||
end
|
|
||||||
|
|
||||||
:inet.setopts(socket, active: :once)
|
:inet.setopts(socket, active: :once)
|
||||||
{:noreply, state}
|
{:noreply, state}
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def handle_info({:excacheUnblocked, ref, reply}, %{blocked: ref} = state) do
|
||||||
|
:gen_tcp.send(state.socket, Encoder.encode(reply))
|
||||||
|
{:noreply, pump(%{state | blocked: nil})}
|
||||||
|
end
|
||||||
|
|
||||||
def handle_info({:tcp_closed, _socket}, state), do: {:stop, :normal, state}
|
def handle_info({:tcp_closed, _socket}, state), do: {:stop, :normal, state}
|
||||||
def handle_info({:tcp_error, _socket, _reason}, state), do: {:stop, :normal, state}
|
def handle_info({:tcp_error, _socket, _reason}, state), do: {:stop, :normal, state}
|
||||||
|
|
||||||
@@ -41,6 +41,14 @@ defmodule Excache.Server.Connection do
|
|||||||
{:noreply, state}
|
{:noreply, state}
|
||||||
end
|
end
|
||||||
|
|
||||||
|
defp pump(%{blocked: nil} = state) do
|
||||||
|
{output, state} = consume(state.buffer, %{state | buffer: ""})
|
||||||
|
unless output == [], do: :gen_tcp.send(state.socket, output)
|
||||||
|
state
|
||||||
|
end
|
||||||
|
|
||||||
|
defp pump(state), do: state
|
||||||
|
|
||||||
defp consume(buffer, state) do
|
defp consume(buffer, state) do
|
||||||
case Parser.parse_command(buffer) do
|
case Parser.parse_command(buffer) do
|
||||||
:incomplete ->
|
:incomplete ->
|
||||||
@@ -51,9 +59,16 @@ defmodule Excache.Server.Connection do
|
|||||||
|
|
||||||
{:ok, args, rest} ->
|
{:ok, args, rest} ->
|
||||||
{replies, state} = Command.execute(args, state)
|
{replies, state} = Command.execute(args, state)
|
||||||
|
continue(replies, rest, state)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp continue([{:blocked, ref}], rest, state),
|
||||||
|
do: {[], %{state | buffer: rest, blocked: ref}}
|
||||||
|
|
||||||
|
defp continue(replies, rest, state) do
|
||||||
encoded = Enum.map(replies, &Encoder.encode/1)
|
encoded = Enum.map(replies, &Encoder.encode/1)
|
||||||
{more, state} = consume(rest, state)
|
{more, state} = consume(rest, state)
|
||||||
{[encoded | more], state}
|
{[encoded | more], state}
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
|
||||||
Reference in new issue
Block a user