1/* Asking the driver (answered, dropped, timed out), the http relay's counting, link queues, fail-stop. */
2#define _XOPEN_SOURCE 700 /* sexc compiles tests without the package's cflags (reported) */
3#include "mesh/check.h"
4#include "mesh/log.h"
5#include "mesh/shared.h"
6
7#include <pthread.h>
8#include <stdlib.h>
9#include <string.h>
10
11static void *driver(void *arg) {
12    mesh_shared *sh = arg;
13    mesh_cmd c;
14    while (mesh_chan_recv(sh->cmd, &c, 2000) == MESH_CHAN_OK) {
15        if (c.k == MESH_CMD_ROLLBACK) mesh_reply_errf(c.reply, "no program \"%s\" (`programs` lists them)", c.name);
16        else if (c.k == MESH_CMD_REQ) mesh_reply_bytes(c.reply, "HTTP/1.0 200 OK\r\n\r\nhi", 21);
17        else if (c.k == MESH_CMD_RESET) { mesh_reply_drop(c.reply); }
18        else if (c.k == MESH_CMD_UNLOAD) { mesh_cmd_free(&c); return NULL; }
19        else mesh_reply_ok(c.reply, "blink 412");
20        c.reply = NULL;
21        mesh_cmd_free(&c);
22    }
23    return NULL;
24}
25
26int main(void) {
27    static _Atomic bool quit;
28    mesh_mailbox mb = mesh_mailbox_test();
29    mesh_shared *sh = mesh_shared_new(mb, "out/programs", false, 0, 1, &quit);
30    pthread_t th;
31    pthread_create(&th, NULL, driver, sh);
32    char *t;
33    mesh_cmd c = mesh_cmd_new(MESH_CMD_LOAD);
34    c.has_src = true;
35    c.src = mesh_src_catalog("hello");
36    CHECK(mesh_shared_ask(sh, c, 1000, &t));
37    CHECK_S(t, "blink 412");
38    free(t);
39    c = mesh_cmd_new(MESH_CMD_ROLLBACK);
40    c.name = mesh_xstrdup("x");
41    CHECK(!mesh_shared_ask(sh, c, 1000, &t));
42    CHECK_S(t, "no program \"x\" (`programs` lists them)");
43    free(t);
44    /* Dropped unanswered: at once, with the Rust's words. */
45    c = mesh_cmd_new(MESH_CMD_RESET);
46    int64_t t0 = mesh_mono_ms();
47    CHECK(!mesh_shared_ask(sh, c, 60000, &t));
48    CHECK_S(t, "no answer from the driver in 60s");
49    CHECK(mesh_mono_ms() - t0 < 1000);
50    free(t);
51    uint8_t *resp;
52    size_t rn;
53    char *err = NULL;
54    CHECK(mesh_shared_http(sh, (const uint8_t *)"GET /status HTTP/1.0\r\n\r\n", 24, &resp, &rn, &err));
55    CHECK(rn == 21 && memcmp(resp, "HTTP/1.0 200 OK", 15) == 0);
56    free(resp);
57    CHECK_U(mesh_host_block_get_u32(&sh->host, MESH_S_HTTP_REQUESTS), 1);
58    uint8_t snap[MESH_SNAP_HOST];
59    mesh_host_block_bytes(&sh->host, snap);
60    CHECK(memcmp(snap + MESH_S_LAST_HTTP, "GET /status HTTP/1.0\0", 21) == 0);
61    c = mesh_cmd_new(MESH_CMD_UNLOAD);
62    mesh_chan_send(sh->cmd, &c);
63    pthread_join(th, NULL);
64    /* Nobody listens: a timeout. */
65    c = mesh_cmd_new(MESH_CMD_PROG_RUN);
66    CHECK(!mesh_shared_ask(sh, c, 30, &t));
67    CHECK_S(t, "no answer from the driver in 30ms");
68    free(t);
69    /* Not in a fleet. */
70    CHECK(!mesh_shared_change(sh, "load x", 100, &t));
71    CHECK_S(t, "not in a fleet");
72    free(t);
73
74    /* Link queues: copies, bounded, broadcast to the linked ones only. */
75    const uint8_t f[3] = {MESH_F_BEAT, 1, 2};
76    CHECK(!mesh_shared_send_peer(sh, 1, f, 3));
77    sh->link_out[1] = mesh_link_queue_new(2);
78    sh->link_out[3] = mesh_link_queue_new(2);
79    CHECK(mesh_shared_send_peer(sh, 1, f, 3));
80    CHECK(mesh_shared_broadcast(sh, f, 3));
81    CHECK(!mesh_shared_send_peer(sh, 1, f, 3)); /* full */
82    CHECK(!mesh_shared_send_peer_wait(sh, 1, f, 3, 20));
83    mesh_link_frame x;
84    CHECK(mesh_chan_try_recv(sh->link_out[1], &x) == MESH_CHAN_OK && x.n == 3 && x.b[0] == MESH_F_BEAT);
85    free(x.b);
86    CHECK(mesh_shared_send_peer_wait(sh, 1, f, 3, 20));
87    CHECK_U(mesh_chan_len(sh->link_out[3]), 1);
88    mesh_log_set_fd(-1);
89    mesh_shared_fail_stop(sh, "disk full");
90    CHECK(atomic_load(&quit) && !mesh_shared_send_peer(sh, 3, f, 3));
91    mesh_shared_free(sh);
92    free(mb.base);
93    return mesh_check_done();
94}