git.meshlink.io Git - meshlink/blob - test/echo-fork.c

   1 #define _GNU_SOURCE
   2 #include <stdio.h>
   3 #include <unistd.h>
   4 #include <stdlib.h>
   5 #include <string.h>
   6
   7 #include "../src/meshlink.h"
   8
   9 /*
  10  * To run this test case, direct a large file to strd
  11  */
  12
  13 volatile bool bar_reachable = false;
  14 volatile bool bar_responded = false;
  15 volatile bool foo_closed = false;
  16 int debug_level;
  17
  18 void log_cb(meshlink_handle_t *mesh, meshlink_log_level_t level, const char *text) {
  19         if(mesh)
  20                 fprintf(stderr, "(%s) ", mesh->name);
  21         fprintf(stderr, "[%d] %s\n", level, text);
  22 }
  23
  24 void status_cb(meshlink_handle_t *mesh, meshlink_node_t *node, bool reachable) {
  25         if(!strcmp(node->name, "bar"))
  26                 bar_reachable = reachable;
  27         else if(!strcmp(node->name, "foo"))
  28                 if(!reachable)
  29                         foo_closed = true;
  30 }
  31
  32 void foo_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) {
  33         // One way only.
  34 }
  35
  36 void bar_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) {
  37         if(!len) {
  38                 fprintf(stderr, "Connection closed by foo\n");
  39                 foo_closed = true;
  40                 return;
  41         }
  42         // Write data to stdout.
  43         write(1, data, len);
  44 }
  45
  46 bool reject_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) {
  47         return false;
  48 }
  49
  50 bool accept_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) {
  51         if(port != 7)
  52                 return false;
  53         meshlink_set_channel_receive_cb(mesh, channel, bar_receive_cb);
  54         if(data)
  55                 bar_receive_cb(mesh, channel, data, len);
  56         return true;
  57 }
  58
  59 void poll_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t len) {
  60         meshlink_set_channel_poll_cb(mesh, channel, NULL);
  61         bar_responded = true;
  62 }
  63
  64 int main1(void) {
  65         close(1);
  66
  67         meshlink_set_log_cb(NULL, debug_level, log_cb);
  68
  69         meshlink_handle_t *mesh1 = meshlink_open("echo-fork_conf.1", "foo", "echo-fork", DEV_CLASS_BACKBONE);
  70         if(!mesh1) {
  71                 fprintf(stderr, "Could not initialize configuration for foo\n");
  72                 return 1;
  73         }
  74
  75         meshlink_set_log_cb(mesh1, debug_level, log_cb);
  76         meshlink_set_channel_accept_cb(mesh1, reject_cb);
  77         meshlink_set_node_status_cb(mesh1, status_cb);
  78
  79         if(!meshlink_start(mesh1)) {
  80                 fprintf(stderr, "Foo could not start\n");
  81                 return 1;
  82         }
  83
  84         for(int i = 0; i < 20; i++) {
  85                 sleep(1);
  86                 if(bar_reachable)
  87                         break;
  88         }
  89
  90         if(!bar_reachable) {
  91                 fprintf(stderr, "Bar not reachable for foo after 20 seconds\n");
  92                 return 1;
  93         }
  94
  95         // Open a channel from foo to bar.
  96
  97         meshlink_node_t *bar = meshlink_get_node(mesh1, "bar");
  98         if(!bar) {
  99                 fprintf(stderr, "Foo could not find bar\n");
 100                 return 1;
 101         }
 102
 103         meshlink_channel_t *channel = meshlink_channel_open(mesh1, bar, 7, foo_receive_cb, NULL, 0);
 104         meshlink_set_channel_poll_cb(mesh1, channel, poll_cb);
 105
 106         // read and buffer stdin
 107         int BUF_SIZE = 1024 * 1024;
 108         char buffer[BUF_SIZE];
 109
 110         for(int i = 0; i < 5; i++) {
 111                 sleep(1);
 112                 if(bar_responded)
 113                         break;
 114         }
 115
 116         if(!bar_responded) {
 117                 fprintf(stderr, "Bar did not respond to foo's channel message\n");
 118                 return 1;
 119         }
 120
 121         do {
 122                 //fprintf(stderr, ":");
 123                 ssize_t len = read(0, buffer, BUF_SIZE);
 124                 if(len <= 0)
 125                         break;
 126                 char *p = buffer;
 127                 while(len > 0) {
 128                         ssize_t sent = meshlink_channel_send(mesh1, channel, p, len);
 129                         if(sent < 0) {
 130                                 fprintf(stderr, "Sending message failed\n");
 131                                 return 1;
 132                         }
 133                         if(!sent)
 134                                 usleep(100000);
 135                         else {
 136                                 len -= sent;
 137                                 p += sent;
 138                         }
 139                 }
 140         } while(true);
 141
 142         fprintf(stderr, "Foo finished sending\n");
 143
 144         meshlink_channel_close(mesh1, channel);
 145         sleep(1);
 146
 147         // Clean up.
 148
 149         meshlink_close(mesh1);
 150
 151         return 0;
 152 }
 153
 154
 155 int main2(void) {
 156         close(0);
 157
 158         sleep(1);
 159
 160         meshlink_set_log_cb(NULL, debug_level, log_cb);
 161
 162         meshlink_handle_t *mesh2 = meshlink_open("echo-fork_conf.2", "bar", "echo-fork", DEV_CLASS_BACKBONE);
 163         if(!mesh2) {
 164                 fprintf(stderr, "Could not initialize configuration for bar\n");
 165                 return 1;
 166         }
 167
 168         meshlink_set_log_cb(mesh2, debug_level, log_cb);
 169         meshlink_set_channel_accept_cb(mesh2, accept_cb);
 170         meshlink_set_node_status_cb(mesh2, status_cb);
 171
 172         if(!meshlink_start(mesh2)) {
 173                 fprintf(stderr, "Bar could not start\n");
 174                 return 1;
 175         }
 176
 177         while(!foo_closed)
 178                 sleep(1);
 179
 180         // Clean up.
 181
 182         meshlink_close(mesh2);
 183
 184         return 0;
 185 }
 186
 187
 188 int main(int argc, char *argv[]) {
 189         debug_level = getenv("DEBUG") ? MESHLINK_DEBUG : MESHLINK_ERROR;
 190
 191         // Initialize and exchange configuration.
 192
 193         meshlink_handle_t *foo = meshlink_open("echo-fork_conf.1", "foo", "echo-fork", DEV_CLASS_BACKBONE);
 194         meshlink_handle_t *bar = meshlink_open("echo-fork_conf.2", "bar", "echo-fork", DEV_CLASS_BACKBONE);
 195         meshlink_add_address(foo, "localhost");
 196         meshlink_import(bar, meshlink_export(foo));
 197         meshlink_import(foo, meshlink_export(bar));
 198         meshlink_close(foo);
 199         meshlink_close(bar);
 200
 201         if(fork())
 202                 return main1();
 203         else
 204                 return main2();
 205 }