utils: ensure that byte-order conversion functions evaluate the argument only once
[project/libubox.git] / examples / ustream-example.c
1 #include <sys/socket.h>
2 #include <netinet/in.h>
3
4 #include <stdio.h>
5 #include <getopt.h>
6 #include <stdlib.h>
7 #include <string.h>
8 #include <unistd.h>
9
10 #include "ustream.h"
11 #include "uloop.h"
12 #include "usock.h"
13
14 static struct uloop_fd server;
15 static const char *port = "10000";
16 struct client *next_client = NULL;
17
18 struct client {
19         struct sockaddr_in sin;
20
21         struct ustream_fd s;
22         int ctr;
23 };
24
25 static void client_read_cb(struct ustream *s, int bytes)
26 {
27         struct client *cl = container_of(s, struct client, s.stream);
28         struct ustream_buf *buf = s->r.head;
29         char *newline, *str;
30
31         do {
32                 str = ustream_get_read_buf(s, NULL);
33                 if (!str)
34                         break;
35
36                 newline = strchr(buf->data, '\n');
37                 if (!newline)
38                         break;
39
40                 *newline = 0;
41                 ustream_printf(s, "%s\n", str);
42                 ustream_consume(s, newline + 1 - str);
43                 cl->ctr += newline + 1 - str;
44         } while(1);
45
46         if (s->w.data_bytes > 256 && !ustream_read_blocked(s)) {
47                 fprintf(stderr, "Block read, bytes: %d\n", s->w.data_bytes);
48                 ustream_set_read_blocked(s, true);
49         }
50 }
51
52 static void client_close(struct ustream *s)
53 {
54         struct client *cl = container_of(s, struct client, s.stream);
55
56         fprintf(stderr, "Connection closed\n");
57         ustream_free(s);
58         close(cl->s.fd.fd);
59         free(cl);
60 }
61
62 static void client_notify_write(struct ustream *s, int bytes)
63 {
64         fprintf(stderr, "Wrote %d bytes, pending: %d\n", bytes, s->w.data_bytes);
65
66         if (s->w.data_bytes < 128 && ustream_read_blocked(s)) {
67                 fprintf(stderr, "Unblock read\n");
68                 ustream_set_read_blocked(s, false);
69         }
70 }
71
72 static void client_notify_state(struct ustream *s)
73 {
74         struct client *cl = container_of(s, struct client, s.stream);
75
76         if (!s->eof)
77                 return;
78
79         fprintf(stderr, "eof!, pending: %d, total: %d\n", s->w.data_bytes, cl->ctr);
80         if (!s->w.data_bytes)
81                 return client_close(s);
82
83 }
84
85 static void server_cb(struct uloop_fd *fd, unsigned int events)
86 {
87         struct client *cl;
88         unsigned int sl = sizeof(struct sockaddr_in);
89         int sfd;
90
91         if (!next_client)
92                 next_client = calloc(1, sizeof(*next_client));
93
94         cl = next_client;
95         sfd = accept(server.fd, (struct sockaddr *) &cl->sin, &sl);
96         if (sfd < 0) {
97                 fprintf(stderr, "Accept failed\n");
98                 return;
99         }
100
101         cl->s.stream.string_data = true;
102         cl->s.stream.notify_read = client_read_cb;
103         cl->s.stream.notify_state = client_notify_state;
104         cl->s.stream.notify_write = client_notify_write;
105         ustream_fd_init(&cl->s, sfd);
106         next_client = NULL;
107         fprintf(stderr, "New connection\n");
108 }
109
110 static int run_server(void)
111 {
112
113         server.cb = server_cb;
114         server.fd = usock(USOCK_TCP | USOCK_SERVER | USOCK_IPV4ONLY | USOCK_NUMERIC, "127.0.0.1", port);
115         if (server.fd < 0) {
116                 perror("usock");
117                 return 1;
118         }
119
120         uloop_init();
121         uloop_fd_add(&server, ULOOP_READ);
122         uloop_run();
123
124         return 0;
125 }
126
127 static int usage(const char *name)
128 {
129         fprintf(stderr, "Usage: %s -p <port>\n", name);
130         return 1;
131 }
132
133 int main(int argc, char **argv)
134 {
135         int ch;
136
137         while ((ch = getopt(argc, argv, "p:")) != -1) {
138                 switch(ch) {
139                 case 'p':
140                         port = optarg;
141                         break;
142                 default:
143                         return usage(argv[0]);
144                 }
145         }
146
147         return run_server();
148 }