1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
var rand: ?Random = null;
pub fn init() !void {
var prng = Random.DefaultPrng.init(blk: {
var seed: u64 = undefined;
try posix.getrandom(mem.asBytes(&seed));
break :blk seed;
});
rand = prng.random();
try network.init();
}
pub fn deinit() void {
network.deinit();
}
fn broadcastSaprusMessage(msg: SaprusMessage, udp_port: u16, allocator: Allocator) !void {
const msg_bytes = try msg.toBytes(allocator);
defer allocator.free(msg_bytes);
var sock = try network.Socket.create(.ipv4, .udp);
defer sock.close();
try sock.setBroadcast(true);
// Bind to 0.0.0.0:0
const bind_addr = network.EndPoint{
.address = network.Address{ .ipv4 = network.Address.IPv4.any },
.port = 0,
};
const dest_addr = network.EndPoint{
.address = network.Address{ .ipv4 = network.Address.IPv4.broadcast },
.port = udp_port,
};
try sock.bind(bind_addr);
_ = try sock.sendTo(dest_addr, msg_bytes);
}
pub fn sendRelay(payload: []const u8, allocator: Allocator) !void {
const msg = SaprusMessage{
.relay = .{
.header = .{ .dest = .{ 255, 255, 255, 255 } },
.payload = payload,
},
};
try broadcastSaprusMessage(msg, 8888, allocator);
}
fn randomPort() u16 {
var p: u16 = 0;
if (rand) |r| {
p = r.intRangeAtMost(u16, 1024, 65000);
} else unreachable;
return p;
}
pub fn sendInitialConnection(payload: []const u8, initial_port: u16, allocator: Allocator) !SaprusMessage {
const dest_port = randomPort();
const msg = SaprusMessage{
.connection = .{
.header = .{
.src_port = initial_port,
.dest_port = dest_port,
.seq_num = 1,
.msg_id = 2,
.reserved = 5,
},
.payload = payload,
},
};
try broadcastSaprusMessage(msg, 8888, allocator);
return msg;
}
var setting_up_socket: std.Thread.Semaphore = std.Thread.Semaphore{};
fn awaitSentinelConnectionResponse(
res: *?SaprusMessage,
err: *?anyerror,
allocator: Allocator,
) !void {
var sock = try network.Socket.create(.ipv4, .udp);
defer sock.close();
// Bind to 255.255.255.255:8888
const bind_addr = network.EndPoint{
.address = network.Address{ .ipv4 = network.Address.IPv4.broadcast },
.port = 8888,
};
// timeout 1s
try sock.setReadTimeout(1_000_000);
try sock.bind(bind_addr);
// Signal that the socket is ready to receive data.
setting_up_socket.post();
var response_buf: [4096]u8 = undefined;
_ = try sock.receive(&response_buf);
const len = sock.receive(&response_buf) catch |e| {
err.* = e;
return;
};
res.* = try SaprusMessage.fromBytes(response_buf[0..len], allocator);
}
pub fn connect(payload: []const u8, allocator: Allocator) !?SaprusMessage {
var initial_port: u16 = 0;
if (rand) |r| {
initial_port = r.intRangeAtMost(u16, 1024, 65000);
} else unreachable;
var err: ?anyerror = null;
var initial_conn_res: ?SaprusMessage = null;
errdefer if (initial_conn_res) |c| c.deinit(allocator);
const response_thread = try std.Thread.spawn(
.{},
awaitSentinelConnectionResponse,
.{
&initial_conn_res,
&err,
allocator,
},
);
// Block until the socket is set up.
try setting_up_socket.timedWait(5 * 1000 * 1000 * 1000 * 1000);
const msg = try sendInitialConnection(payload, initial_port, allocator);
// Await response from the sentinel
response_thread.join();
// Complete handshake after awaiting response
try broadcastSaprusMessage(msg, randomPort(), allocator);
return initial_conn_res;
}
const SaprusMessage = @import("./saprus_message.zig").SaprusMessage;
const std = @import("std");
const Random = std.Random;
const posix = std.posix;
const mem = std.mem;
const network = @import("network");
const Allocator = mem.Allocator;
|
