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
|
#define AL_LOG_SECTION "multiplex"
//#define AL_LOG_ENABLE_TRACE
#include <al/log.h>
#include "multiplex.h"
bool nn_multiplex_socket_init(struct nn_multiplex_socket *multi, u8 type,
bool (*connection_callback)(void *, u8, struct nn_packet_stream *), void *userdata)
{
multi->sock.type = type;
s32 flags = NNWT_SOCKET_NONBLOCKING;
if (type == NNWT_SOCKET_TCP) flags |= NNWT_SOCKET_REUSE_ADDR;
if (!nn_socket_init(&multi->sock, flags)) {
return false;
}
multi->connection_callback = connection_callback;
multi->userdata = userdata;
return true;
}
static void socket_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
struct nn_multiplex_connection *conn = (struct nn_multiplex_connection *)w->data;
struct nn_multiplex_socket *multi = conn->multi;
(void)revents;
u8 id;
ssize_t ret = nn_socket_read(&conn->sock, &id, 1);
if (ret <= 0) {
ev_io_stop(loop, &conn->event);
nn_socket_close(&conn->sock);
al_free(conn);
if (ret == 0) log_warn("read() returned 0 on POLLIN.");
return;
}
log_trace("ID byte: 0x%hhx.", id);
switch (id) {
case 0x48: // '(H)EAD'
break;
case 0x47: // '(G)ET'
// Attempt WebSocket upgrade.
break;
case 0x50: // '(P)OST'
break;
}
ev_io_stop(loop, &conn->event);
struct nn_packet_stream *stream = al_alloc_object(struct nn_packet_stream);
if (multi->connection_callback(multi->userdata, id, stream)) {
nn_packet_stream_from_socket(stream, multi->loop, &conn->sock);
} else {
nn_socket_close(&conn->sock);
}
al_free(conn);
}
static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
struct nn_multiplex_socket *multi = (struct nn_multiplex_socket *)w->data;
(void)revents;
struct nn_multiplex_connection *conn = al_alloc_object(struct nn_multiplex_connection);
if (!nn_socket_accept(&multi->sock, &conn->sock, NNWT_SOCKET_NONBLOCKING)) {
al_free(conn);
return;
}
conn->multi = multi;
conn->event.data = conn;
ev_io_init_n(&conn->event, socket_read_callback, nn_socket_get_fd(&conn->sock), EV_READ);
ev_io_start(loop, &conn->event);
}
bool nn_multiplex_socket_listen(struct nn_multiplex_socket *multi, struct nn_event_loop *loop, str *addr, u16 port)
{
if (!nn_socket_bind(&multi->sock, addr, port) || !nn_socket_listen(&multi->sock)) {
return false;
}
multi->loop = loop;
multi->event.data = multi;
ev_io_init_n(&multi->event, socket_connection_callback, nn_socket_get_fd(&multi->sock), EV_READ);
ev_io_start(multi->loop->ev, &multi->event);
return true;
}
void nn_multiplex_socket_close(struct nn_multiplex_socket *multi)
{
ev_io_stop(multi->loop->ev, &multi->event);
nn_socket_shutdown(&multi->sock);
nn_socket_close(&multi->sock);
nn_socket_cleanup(&multi->sock);
}
static struct nn_multiplex_direct multiplex_direct_global = { 0 };
void nn_multiplex_direct_init(bool (*connection_callback)(void *, u8, struct nn_packet_stream *), void *userdata)
{
multiplex_direct_global.closing_bridge = al_alloc_object(struct nn_packet_stream);
multiplex_direct_global.connection_callback = connection_callback;
multiplex_direct_global.userdata = userdata;
}
static void direct_connect(struct nn_packet_stream *client, u8 id)
{
struct nn_packet_stream *server = al_alloc_object(struct nn_packet_stream);
nn_packet_stream_init(server, NULL, NULL, NULL);
multiplex_direct_global.connection_callback(multiplex_direct_global.userdata, id, server);
server->direct = client;
client->direct = server;
// Explicitly signal connected on the server first.
nn_packet_stream_set_connected(server);
nn_packet_stream_set_connected(client);
}
void nn_multiplex_direct_connect(struct nn_packet_stream *client, u8 id)
{
client->id = id;
direct_connect(client, id);
}
void nn_multiplex_direct_reconnect(struct nn_packet_stream *client)
{
direct_connect(client, client->id);
}
struct nn_packet_stream *nn_multiplex_direct_get_bridge(void)
{
return multiplex_direct_global.closing_bridge;
}
void nn_multiplex_direct_close(void)
{
al_free(multiplex_direct_global.closing_bridge);
}
|