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
160
161
162
163
164
|
#include "line_processor.h"
// Read from a socket/fd and callback for each "line" determined by a delimiter string.
// This isn't meant to be used seriously and is currently just for testing.
#define LINE_PROCESSOR_CHUNK_SIZE 128
static str LINE_PROCESSOR_END = al_str_c("X");
static void remove_client(struct nn_line_processor *pro, struct nn_line_processor_client *client)
{
ev_io_stop(pro->loop->ev, &client->event);
if (pro->type == NNWT_LINE_PROCESSOR_SOCKET) {
nn_socket_close(&client->sock);
}
nn_buffer_free(&client->buf);
al_array_remove(pro->clients, client);
al_free(client);
}
static void read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
(void)loop;
struct nn_line_processor_client *cl = (struct nn_line_processor_client *)w->data;
(void)revents;
struct nn_line_processor *pro = cl->pro;
u8 type = pro->type;
nn_buffer_ensure_space(&cl->buf, cl->index + LINE_PROCESSOR_CHUNK_SIZE);
u8 *ptr = nn_buffer_get_ptr(&cl->buf, cl->index);
ssize_t ret;
switch (type) {
case NNWT_LINE_PROCESSOR_SOCKET:
ret = nn_socket_read(&cl->sock, ptr, LINE_PROCESSOR_CHUNK_SIZE);
break;
case NNWT_LINE_PROCESSOR_FD:
ret = nn_read(pro->fd, ptr, LINE_PROCESSOR_CHUNK_SIZE);
break;
default:
al_assert_and_return();
}
if (ret <= 0) {
return;
}
cl->index += ret;
ptr = nn_buffer_get_ptr(&cl->buf, 0);
str s = al_str_w((char *)ptr, 0, cl->index);
str *dl = &pro->delim;
u32 offset = dl->length - 1;
for (size_t i = cl->mark; i < cl->index - offset; i++) {
if (al_str_cmp(&s, dl, i, dl->length) == 0) {
s.length = i;
if (al_str_eq(&s, &LINE_PROCESSOR_END)) {
remove_client(pro, cl);
return;
}
switch (pro->callback(pro->userdata, &s)) {
case NNWT_LINE_PROCESSOR_STOP:
nn_line_processor_stop(pro);
return;
case NNWT_LINE_PROCESSOR_CONTINUE:
break;
}
size_t skip = i + offset + 1;
cl->index = cl->index - skip;
al_memmove(ptr, &ptr[skip], cl->index);
s = al_str_w((char *)ptr, 0, cl->index);
cl->mark = i = 0;
}
}
cl->mark = cl->index;
}
static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
(void)loop;
struct nn_line_processor *pro = (struct nn_line_processor *)w->data;
(void)revents;
struct nn_line_processor_client *cl = al_alloc_object(struct nn_line_processor_client);
if (!nn_socket_accept(pro->sock, &cl->sock, NNWT_SOCKET_NONBLOCKING)) {
al_free(cl);
return;
}
cl->index = 0;
nn_buffer_init(&cl->buf);
cl->mark = 0;
cl->pro = pro;
cl->event.data = cl;
al_array_push(pro->clients, cl);
ev_io_init_n(&cl->event, read_callback, nn_socket_get_fd(&cl->sock), EV_READ);
ev_io_start(pro->loop->ev, &cl->event);
}
void nn_line_processor_init(struct nn_line_processor *pro, str *delim)
{
pro->loop = NULL;
al_str_clone(&pro->delim, delim);
al_array_init(pro->clients);
}
void nn_line_processor_open_socket(struct nn_line_processor *pro, struct nn_socket *sock)
{
pro->type = NNWT_LINE_PROCESSOR_SOCKET;
pro->sock = sock;
}
void nn_line_processor_open_fd(struct nn_line_processor *pro, s32 fd)
{
pro->type = NNWT_LINE_PROCESSOR_FD;
pro->fd = fd;
}
void nn_line_processor_run(struct nn_line_processor *pro, struct nn_event_loop *loop)
{
pro->loop = loop;
switch (pro->type) {
case NNWT_LINE_PROCESSOR_SOCKET:
pro->event.data = pro;
ev_io_init_n(&pro->event, socket_connection_callback, nn_socket_get_fd(pro->sock), EV_READ);
ev_io_start(pro->loop->ev, &pro->event);
break;
case NNWT_LINE_PROCESSOR_FD: { // Not supported on windows.
struct nn_line_processor_client *cl = al_alloc_object(struct nn_line_processor_client);
cl->index = 0;
nn_buffer_init(&cl->buf);
cl->mark = 0;
cl->pro = pro;
cl->event.data = cl;
al_array_push(pro->clients, cl);
ev_io_init_n(&cl->event, read_callback, pro->fd, EV_READ);
ev_io_start(pro->loop->ev, &cl->event);
break;
}
}
}
void nn_line_processor_stop(struct nn_line_processor *pro)
{
if (pro->loop) {
switch (pro->type) {
case NNWT_LINE_PROCESSOR_SOCKET:
ev_io_stop(pro->loop->ev, &pro->event);
break;
case NNWT_LINE_PROCESSOR_FD:
break;
}
struct nn_line_processor_client *cl;
al_array_foreach(pro->clients, i, cl) {
remove_client(cl->pro, cl);
}
}
al_array_free(pro->clients);
al_str_free(&pro->delim);
}
|