Skip to content

Commit cdae88a

Browse files
committed
fix: Allow for batched init responses + protocol version handling
1 parent 3d2c15a commit cdae88a

1 file changed

Lines changed: 90 additions & 6 deletions

File tree

‎lib/octocon_web/user_socket.ex‎

Lines changed: 90 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,17 @@ defmodule OctoconWeb.UserSocket do
33

44
channel "system:*", OctoconWeb.UserChannel
55

6+
@base_version Version.parse("1.0.0")
7+
@batched_init_version Version.parse("2.0.0")
8+
9+
@batched_dummy_response %{
10+
"batched" => true,
11+
"system" => nil,
12+
"alters" => nil,
13+
"fronts" => nil,
14+
"tags" => nil
15+
}
16+
617
def connect(%{"token" => token}, socket, _connect_info) do
718
case Octocon.Auth.Guardian.resource_from_token(token) do
819
{:ok, system_id, _claims} ->
@@ -52,12 +63,38 @@ defmodule OctoconWeb.UserChannel do
5263
{:allow, _count} ->
5364
is_reconnect = Map.get(params, "isReconnect", false)
5465

55-
if is_reconnect do
56-
{:ok, socket}
57-
else
58-
Process.send_after(socket.transport_pid, :garbage_collect, :timer.seconds(1))
59-
60-
{:ok, generate_init_data(system_id), socket}
66+
protocol_version =
67+
case Map.get(params, "protocolVersion") do
68+
nil -> {:ok, @base_version}
69+
version_str -> Version.parse(version_str)
70+
end
71+
72+
case protocol_version do
73+
{:ok, version} ->
74+
if is_reconnect do
75+
{:ok, socket}
76+
else
77+
init_data = generate_init_data(system_id)
78+
79+
# What the fuck
80+
exceeds_ios_limit = :erlang.external_size(init_data) * 1.1 > 1_048_576
81+
82+
if exceeds_ios_limit && version >= @batched_init_version do
83+
spawn(fn ->
84+
# TODO: Necessary delay?
85+
Process.sleep(50)
86+
send_batched_init(socket, init_data)
87+
end)
88+
89+
{:ok, %{@batched_dummy_response | system: init_data["system"]}, socket}
90+
else
91+
Process.send_after(socket.transport_pid, :garbage_collect, :timer.seconds(1))
92+
{:ok, init_data, socket}
93+
end
94+
end
95+
96+
_ ->
97+
{:error, %{reason: "unsupported_protocol_version"}}
6198
end
6299

63100
{:deny, _limit} ->
@@ -101,6 +138,53 @@ defmodule OctoconWeb.UserChannel do
101138
}
102139
end
103140

141+
defp send_batched_init(socket, %{"alters" => alters, "tags" => tags, "fronts" => fronts}) do
142+
batched_alters = Enum.chunk_every(alters, 3_000)
143+
batched_tags = Enum.chunk_every(tags, 1_000)
144+
batched_fronts = Enum.chunk_every(fronts, 50)
145+
146+
alters_batch_count = length(batched_alters)
147+
tags_batch_count = length(batched_tags)
148+
fronts_batch_count = length(batched_fronts)
149+
150+
Enum.with_index(batched_alters)
151+
|> Enum.each(fn {alter_batch, index} ->
152+
Process.sleep(50)
153+
154+
push(socket, "batched_init_alters", %{
155+
"batch_index" => index + 1,
156+
"total_batches" => alters_batch_count,
157+
"alters" => alter_batch
158+
})
159+
end)
160+
161+
Enum.with_index(batched_tags)
162+
|> Enum.each(fn {tag_batch, index} ->
163+
Process.sleep(50)
164+
165+
push(socket, "batched_init_tags", %{
166+
"batch_index" => index + 1,
167+
"total_batches" => tags_batch_count,
168+
"tags" => tag_batch
169+
})
170+
end)
171+
172+
Enum.with_index(batched_fronts)
173+
|> Enum.each(fn {front_batch, index} ->
174+
Process.sleep(50)
175+
176+
push(socket, "batched_init_fronts", %{
177+
"batch_index" => index + 1,
178+
"total_batches" => fronts_batch_count,
179+
"fronts" => front_batch
180+
})
181+
end)
182+
183+
Process.send_after(socket.transport_pid, :garbage_collect, :timer.seconds(1))
184+
185+
push(socket, "batched_init_complete", %{})
186+
end
187+
104188
@impl true
105189
def handle_in(
106190
"endpoint",

0 commit comments

Comments
 (0)