Deployment and lifecycle management for Nix
agent: move core channel client into a separate module
+172
-80
+168
apps/sower_agent/lib/sower_agent/channel_client.ex
+168
apps/sower_agent/lib/sower_agent/channel_client.ex
···
1
+
defmodule SowerAgent.ChannelClient do
2
+
@moduledoc """
3
+
Shared Slipstream client helpers.
4
+
5
+
Use this module from a consumer-specific socket client to get the default
6
+
client interface along with generic Slipstream callbacks. Override the
7
+
callbacks you need in the consumer module to customize behavior.
8
+
"""
9
+
10
+
defmacro __using__(opts) do
11
+
quote bind_quoted: [opts: opts] do
12
+
use Slipstream
13
+
require Logger
14
+
15
+
@lobby_topic Keyword.fetch!(opts, :lobby_topic)
16
+
17
+
#
18
+
# client
19
+
#
20
+
21
+
def send(message) do
22
+
GenServer.call(__MODULE__, message)
23
+
end
24
+
25
+
def send(event, params) do
26
+
GenServer.call(__MODULE__, {event, params})
27
+
end
28
+
29
+
def cast(event) when is_atom(event) do
30
+
GenServer.cast(__MODULE__, event)
31
+
end
32
+
33
+
def cast(event, params) do
34
+
GenServer.cast(__MODULE__, {event, params})
35
+
end
36
+
37
+
def restart() do
38
+
GenServer.stop(__MODULE__, :shutdown)
39
+
end
40
+
41
+
def lobby_topic, do: @lobby_topic
42
+
43
+
#
44
+
# Slipstream callbacks
45
+
#
46
+
47
+
@impl Slipstream
48
+
def handle_call(:ping, _from, socket) do
49
+
{:ok, ref} = push(socket, lobby_topic(), "ping", %{})
50
+
{:ok, "pong"} = await_reply(ref)
51
+
{:reply, {:ok, :pong}, socket}
52
+
end
53
+
54
+
def handle_call({event, params}, _from, socket) do
55
+
{:ok, ref} = push(socket, private_channel(socket), event, params)
56
+
{:reply, await_reply(ref), socket}
57
+
end
58
+
59
+
def handle_call(request, from, socket) do
60
+
Logger.error(msg: "Unsupported call", request: request, from: from)
61
+
{:reply, {:error, :unsupported_request}, socket}
62
+
end
63
+
64
+
@impl Slipstream
65
+
def handle_cast({event, params}, socket) do
66
+
{:ok, _} = push(socket, private_channel(socket), event, params)
67
+
{:noreply, socket}
68
+
end
69
+
70
+
def handle_cast(message, socket) do
71
+
Logger.error(msg: "Unsupported cast", message: message)
72
+
{:noreply, socket}
73
+
end
74
+
75
+
def start_link(args) do
76
+
Slipstream.start_link(__MODULE__, args, name: __MODULE__)
77
+
end
78
+
79
+
@impl Slipstream
80
+
def init(_args) do
81
+
config = Application.get_all_env(__MODULE__)
82
+
83
+
case connect(config) do
84
+
{:ok, socket} ->
85
+
Logger.debug(msg: "Connecting")
86
+
{:ok, socket}
87
+
88
+
{:error, reason} ->
89
+
Logger.error(
90
+
"Could not start #{__MODULE__} because of " <>
91
+
"validation failure: #{inspect(reason)}"
92
+
)
93
+
94
+
:ignore
95
+
end
96
+
end
97
+
98
+
@impl Slipstream
99
+
def handle_connect(socket) do
100
+
Logger.info(
101
+
msg: "Connected to websocket",
102
+
authority: socket.channel_config.uri.authority,
103
+
path: socket.channel_config.uri.path
104
+
)
105
+
106
+
{:ok, join(socket, lobby_topic())}
107
+
end
108
+
109
+
@impl Slipstream
110
+
def handle_join(topic, _params, socket) do
111
+
Logger.info(msg: "Joined channel topic", topic: topic)
112
+
{:ok, socket}
113
+
end
114
+
115
+
@impl Slipstream
116
+
def handle_message(topic, message, params, socket) do
117
+
Logger.debug(
118
+
msg: "Received unknown message",
119
+
topic: topic,
120
+
message: message,
121
+
params: params
122
+
)
123
+
124
+
{:noreply, socket}
125
+
end
126
+
127
+
@impl Slipstream
128
+
def handle_reply(_ref, :ok, socket) do
129
+
{:noreply, socket}
130
+
end
131
+
132
+
def handle_reply(ref, payload, socket) do
133
+
Logger.debug(msg: "Received unknown reply", ref: ref, payload: payload)
134
+
{:noreply, socket}
135
+
end
136
+
137
+
def private_channel(%{assigns: %{private_topic: topic}}) when is_binary(topic) do
138
+
topic
139
+
end
140
+
141
+
def private_channel(_socket) do
142
+
raise ArgumentError, "Assign :private_topic or override private_channel/1"
143
+
end
144
+
145
+
def push_message(socket, %module{} = struct) do
146
+
event = module.event()
147
+
148
+
topic =
149
+
case module.topic_type() do
150
+
:private -> private_channel(socket)
151
+
:lobby -> lobby_topic()
152
+
end
153
+
154
+
push(socket, topic, event, struct)
155
+
end
156
+
157
+
defoverridable handle_call: 3,
158
+
handle_cast: 2,
159
+
init: 1,
160
+
handle_connect: 1,
161
+
handle_join: 3,
162
+
handle_message: 4,
163
+
handle_reply: 3,
164
+
private_channel: 1,
165
+
push_message: 2
166
+
end
167
+
end
168
+
end
+2
-79
apps/sower_agent/lib/sower_agent/socket_client.ex
+2
-79
apps/sower_agent/lib/sower_agent/socket_client.ex
···
1
1
defmodule SowerAgent.SocketClient do
2
-
use Slipstream
2
+
use SowerAgent.ChannelClient, lobby_topic: "agent:lobby"
3
3
4
4
require Logger
5
5
6
-
@lobby_topic "agent:lobby"
7
-
8
6
alias SowerAgent.Storage
9
7
10
-
#
11
-
# client
12
-
#
13
-
14
-
def send(message) do
15
-
GenServer.call(__MODULE__, message)
16
-
end
17
-
18
-
def send(event, params) do
19
-
GenServer.call(__MODULE__, {event, params})
20
-
end
21
-
22
-
def cast(event) when is_atom(event) do
23
-
GenServer.cast(__MODULE__, event)
24
-
end
25
-
26
-
def cast(event, params) do
27
-
GenServer.cast(__MODULE__, {event, params})
28
-
end
29
-
30
-
def restart() do
31
-
GenServer.stop(__MODULE__, :shutdown)
32
-
end
33
-
34
-
@impl Slipstream
35
-
def handle_call(:ping, _, socket) do
36
-
{:ok, ref} = push(socket, "agent:lobby", "ping", %{})
37
-
{:ok, "pong"} = await_reply(ref)
38
-
{:reply, {:ok, :pong}, socket}
39
-
end
40
-
41
8
def handle_call(
42
9
{:deployment_request, subscription = %SowerClient.Schemas.Orchestration.Subscription{}},
43
10
_from,
···
53
20
{:reply, :ok, Map.put(socket, :upgrade_ref, ref)}
54
21
end
55
22
56
-
def handle_call({event, params}, _from, socket) do
57
-
{:ok, ref} = push(socket, private_channel(), event, params)
58
-
{:reply, await_reply(ref), socket}
59
-
end
60
-
61
-
def handle_call(request, from, socket) do
62
-
Logger.error(msg: "Unsupported call", request: request, from: from)
63
-
{:reply, {:error, :unsupported_request}, socket}
64
-
end
65
-
66
23
@impl Slipstream
67
-
def handle_cast({event, params}, socket) do
68
-
{:ok, _} = push(socket, private_channel(), event, params)
69
-
{:noreply, socket}
70
-
end
71
-
72
24
def handle_cast(:register_subscriptions, socket) do
73
25
subscriptions =
74
26
SowerAgent.Config.get().subscriptions
···
104
56
# server
105
57
#
106
58
107
-
def start_link(args) do
108
-
Slipstream.start_link(__MODULE__, args, name: __MODULE__)
109
-
end
110
-
111
59
@impl Slipstream
112
60
def init(_args) do
113
61
config = Application.get_all_env(__MODULE__)
···
136
84
137
85
:ignore
138
86
end
139
-
end
140
-
141
-
@impl Slipstream
142
-
def handle_connect(socket) do
143
-
Logger.info(
144
-
msg: "Connected to websocket",
145
-
authority: socket.channel_config.uri.authority,
146
-
path: socket.channel_config.uri.path
147
-
)
148
-
149
-
{:ok, join(socket, @lobby_topic)}
150
87
end
151
88
152
89
@impl Slipstream
···
257
194
{:noreply, socket}
258
195
end
259
196
260
-
defp private_channel() do
197
+
def private_channel(_socket) do
261
198
"agent:#{Storage.read().agent_sid}"
262
-
end
263
-
264
-
defp push_message(socket, %module{} = struct) do
265
-
event = module.event()
266
-
267
-
topic =
268
-
case module.topic_type() do
269
-
:private -> private_channel()
270
-
:lobby -> @lobby_topic
271
-
end
272
-
273
-
with {:ok, ref} <- push(socket, topic, event, struct) do
274
-
{:ok, ref}
275
-
end
276
199
end
277
200
end