1/* Order, growth, bounds, timeouts, disconnects and drops, with threads. */
2#include "mesh/chan.h"
3#include "mesh/check.h"
4
5#include <pthread.h>
6
7static int dropped;
8static void count_drop(void *item) { dropped += *(int *)item; }
9
10static void *producer(void *arg) {
11    mesh_chan *c = arg;
12    for (int i = 0; i < 10000; i++) mesh_chan_send(c, &i);
13    mesh_chan_drop_sender(c);
14    mesh_chan_release(c);
15    return NULL;
16}
17
18int main(void) {
19    mesh_chan *c = mesh_chan_new(sizeof(int), 0, NULL);
20    for (int i = 0; i < 100; i++) CHECK_I(mesh_chan_send(c, &i), MESH_CHAN_OK);
21    CHECK_U(mesh_chan_len(c), 100);
22    int v = -1, ok = 1;
23    for (int i = 0; i < 100; i++) ok &= mesh_chan_try_recv(c, &v) == MESH_CHAN_OK && v == i;
24    CHECK(ok);
25    CHECK_I(mesh_chan_try_recv(c, &v), MESH_CHAN_EMPTY);
26    int64_t t0 = mesh_mono_ms();
27    CHECK_I(mesh_chan_recv(c, &v, 50), MESH_CHAN_TIMEOUT);
28    int64_t dt = mesh_mono_ms() - t0;
29    CHECK(dt >= 45 && dt < 1000);
30    /* A producer thread: every message, in order, then disconnected. */
31    mesh_chan_retain(c);
32    mesh_chan_add_sender(c); /* the thread's */
33    pthread_t th;
34    pthread_create(&th, NULL, producer, c);
35    mesh_chan_drop_sender(c); /* ours: only the thread's remains */
36    long sum = 0;
37    int n = 0, last = -1, ordered = 1;
38    while (mesh_chan_recv(c, &v, -1) == MESH_CHAN_OK) { ordered &= v == last + 1; last = v; sum += v; n++; }
39    pthread_join(th, NULL);
40    CHECK(ordered);
41    CHECK_I(n, 10000);
42    CHECK_I(sum, 49995000L);
43    CHECK_I(mesh_chan_try_recv(c, &v), MESH_CHAN_CLOSED);
44    mesh_chan_release(c);
45
46    /* Bounded: full; the receiver gone: closed, and queued messages dropped. */
47    mesh_chan *b = mesh_chan_new(sizeof(int), 2, count_drop);
48    int one = 1, two = 2, three = 3;
49    CHECK_I(mesh_chan_try_send(b, &one), MESH_CHAN_OK);
50    CHECK_I(mesh_chan_try_send(b, &two), MESH_CHAN_OK);
51    CHECK_I(mesh_chan_try_send(b, &three), MESH_CHAN_FULL);
52    mesh_chan_close_rx(b);
53    CHECK_I(dropped, 3);
54    CHECK_I(mesh_chan_try_send(b, &three), MESH_CHAN_CLOSED);
55    mesh_chan_release(b);
56
57    /* Released with messages queued: they are dropped. */
58    dropped = 0;
59    mesh_chan *d = mesh_chan_new(sizeof(int), 0, count_drop);
60    mesh_chan_send(d, &two);
61    mesh_chan_send(d, &three);
62    mesh_chan_release(d);
63    CHECK_I(dropped, 5);
64    return mesh_check_done();
65}