X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=test%2Fecho-fork.c;h=e6aefaab8a4fdcdaa741eb8cce8ae07aa08ebc44;hb=refs%2Fheads%2Fencrypted-storage;hp=b1b2bfe3fe3905b8dd23e80831c3b6640b9df5f2;hpb=6db78ef7a4b7190b566a87f615a6e7e7ff2004e9;p=meshlink diff --git a/test/echo-fork.c b/test/echo-fork.c old mode 100755 new mode 100644 index b1b2bfe3..e6aefaab --- a/test/echo-fork.c +++ b/test/echo-fork.c @@ -12,86 +12,95 @@ volatile bool bar_reachable = false; volatile bool bar_responded = false; +volatile bool foo_closed = false; +int debug_level; void log_cb(meshlink_handle_t *mesh, meshlink_log_level_t level, const char *text) { - if(mesh) + if(mesh) { fprintf(stderr, "(%s) ", mesh->name); + } + fprintf(stderr, "[%d] %s\n", level, text); } void status_cb(meshlink_handle_t *mesh, meshlink_node_t *node, bool reachable) { - if(!strcmp(node->name, "bar")) + (void)mesh; + + if(!strcmp(node->name, "bar")) { bar_reachable = reachable; + } else if(!strcmp(node->name, "foo")) + if(!reachable) { + foo_closed = true; + } } void foo_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) { - char tmp[len+1]; - memset( tmp, 0, sizeof tmp ); - snprintf( tmp, len+1, "%s", (char*)data ); - fprintf(stderr, "Foo received from Bar:\n%s\n", tmp); - fprintf(stderr, "==============================\n"); - fprintf(stdout, "%s", tmp ); + (void)mesh; + (void)channel; + (void)data; + (void)len; + + // One way only. } void bar_receive_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len) { - // Echo the data back. - //char tmp[len+1]; - //memset( tmp, 0, sizeof tmp ); - //snprintf( tmp, len+1, "%s", (char*)data ); - //fprintf(stderr, "Bar received:\n%s", tmp); - //fprintf(stderr, "==============================\n"); - meshlink_channel_send(mesh, channel, data, len); + (void)mesh; + (void)channel; + + if(!len) { + fprintf(stderr, "Connection closed by foo\n"); + foo_closed = true; + return; + } + + // Write data to stdout. + write(1, data, len); } bool reject_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) { + (void)mesh; + (void)channel; + (void)port; + (void)data; + (void)len; + return false; } bool accept_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, uint16_t port, const void *data, size_t len) { - if(port != 7) + if(port != 7) { return false; + } + meshlink_set_channel_receive_cb(mesh, channel, bar_receive_cb); - if(data) + + if(data) { bar_receive_cb(mesh, channel, data, len); + } + return true; } void poll_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t len) { + (void)len; + meshlink_set_channel_poll_cb(mesh, channel, NULL); - bar_responded=true; + bar_responded = true; } -int main1(int rfd, int wfd) { - meshlink_set_log_cb(NULL, MESHLINK_DEBUG, log_cb); +int main1(void) { + close(1); - meshlink_handle_t *mesh1 = meshlink_open("channels_conf.1", "foo", "channels", DEV_CLASS_BACKBONE); - if(!mesh1) { - fprintf(stderr, "Could not initialize configuration for foo\n"); - return 1; - } + meshlink_set_log_cb(NULL, debug_level, log_cb); - meshlink_add_address(mesh1, "localhost"); + meshlink_handle_t *mesh1 = meshlink_open("echo-fork_conf.1", "foo", "echo-fork", DEV_CLASS_BACKBONE); - char *data = meshlink_export(mesh1); - if(!data) { - fprintf(stderr, "Foo could not export its configuration\n"); + if(!mesh1) { + fprintf(stderr, "Could not initialize configuration for foo\n"); return 1; } - size_t len = strlen(data); - write(wfd, &len, sizeof len); - write(wfd, data, len); - free(data); - - read(rfd, &len, sizeof len); - char indata[len + 1]; - read(rfd, indata, len); - indata[len] = 0; - - fprintf(stderr, "Foo exchanged data\n"); - - meshlink_import(mesh1, indata); - + meshlink_set_log_cb(mesh1, debug_level, log_cb); meshlink_set_channel_accept_cb(mesh1, reject_cb); meshlink_set_node_status_cb(mesh1, status_cb); @@ -102,8 +111,10 @@ int main1(int rfd, int wfd) { for(int i = 0; i < 20; i++) { sleep(1); - if(bar_reachable) + + if(bar_reachable) { break; + } } if(!bar_reachable) { @@ -112,8 +123,9 @@ int main1(int rfd, int wfd) { } // Open a channel from foo to bar. - + meshlink_node_t *bar = meshlink_get_node(mesh1, "bar"); + if(!bar) { fprintf(stderr, "Foo could not find bar\n"); return 1; @@ -123,27 +135,15 @@ int main1(int rfd, int wfd) { meshlink_set_channel_poll_cb(mesh1, channel, poll_cb); // read and buffer stdin - int BUF_SIZE = 1024; + int BUF_SIZE = 1024 * 1024; char buffer[BUF_SIZE]; - size_t contentSize = 1; - char *content = malloc( sizeof(char) * BUF_SIZE ); - if (!content) { - fprintf(stderr, "Could not allocate buffer\n"); - } - - fprintf(stderr, "Foo reading from stdin...\n"); - content[0] = '\0'; - while(fgets(buffer,BUF_SIZE,stdin)) { - char *old = content; - contentSize += strlen(buffer); - content = realloc(content, contentSize); - strcat(content,buffer); - } for(int i = 0; i < 5; i++) { sleep(1); - if(bar_responded) + + if(bar_responded) { break; + } } if(!bar_responded) { @@ -151,22 +151,37 @@ int main1(int rfd, int wfd) { return 1; } - //fprintf(stderr, "Foo sending:\n%s", content); - //fprintf(stderr, "==============================\n"); + do { + //fprintf(stderr, ":"); + ssize_t len = read(0, buffer, BUF_SIZE); - size_t total = 0; - while ( total != contentSize ) - { - ssize_t tmp = meshlink_channel_send(mesh1, channel, content + total, contentSize - total); - if (tmp >= 0) - total += tmp; - } + if(len <= 0) { + break; + } + + char *p = buffer; + + while(len > 0) { + ssize_t sent = meshlink_channel_send(mesh1, channel, p, len); - sleep(60); + if(sent < 0) { + fprintf(stderr, "Sending message failed\n"); + return 1; + } - free(content); + if(!sent) { + usleep(100000); + } else { + len -= sent; + p += sent; + } + } + } while(true); + + fprintf(stderr, "Foo finished sending\n"); meshlink_channel_close(mesh1, channel); + sleep(1); // Clean up. @@ -176,45 +191,32 @@ int main1(int rfd, int wfd) { } -int main2(int rfd, int wfd) { +int main2(void) { + close(0); + sleep(1); - meshlink_set_log_cb(NULL, MESHLINK_DEBUG, log_cb); + meshlink_set_log_cb(NULL, debug_level, log_cb); + + meshlink_handle_t *mesh2 = meshlink_open("echo-fork_conf.2", "bar", "echo-fork", DEV_CLASS_BACKBONE); - meshlink_handle_t *mesh2 = meshlink_open("channels_conf.2", "bar", "channels", DEV_CLASS_BACKBONE); if(!mesh2) { fprintf(stderr, "Could not initialize configuration for bar\n"); return 1; } - char *data = meshlink_export(mesh2); - if(!data) { - fprintf(stderr, "Bar could not export its configuration\n"); - return 1; - } - - size_t len = strlen(data); - if(write(wfd, &len, sizeof len) <= 0) abort(); - if(write(wfd, data, len) <= 0) abort(); - free(data); - - read(rfd, &len, sizeof len); - char indata[len + 1]; - read(rfd, indata, len); - indata[len] = 0; - - fprintf(stderr, "Bar exchanged data\n"); - - meshlink_import(mesh2, indata); - + meshlink_set_log_cb(mesh2, debug_level, log_cb); meshlink_set_channel_accept_cb(mesh2, accept_cb); + meshlink_set_node_status_cb(mesh2, status_cb); if(!meshlink_start(mesh2)) { fprintf(stderr, "Bar could not start\n"); return 1; } - sleep(60); + while(!foo_closed) { + sleep(1); + } // Clean up. @@ -224,14 +226,22 @@ int main2(int rfd, int wfd) { } -int main(int argc, char *argv[]) { - int fda[2], fdb[2], result; +int main() { + debug_level = getenv("DEBUG") ? MESHLINK_DEBUG : MESHLINK_ERROR; + + // Initialize and exchange configuration. - pipe2(fda, 0); - pipe2(fdb, 0); + meshlink_handle_t *foo = meshlink_open("echo-fork_conf.1", "foo", "echo-fork", DEV_CLASS_BACKBONE); + meshlink_handle_t *bar = meshlink_open("echo-fork_conf.2", "bar", "echo-fork", DEV_CLASS_BACKBONE); + meshlink_add_address(foo, "localhost"); + meshlink_import(bar, meshlink_export(foo)); + meshlink_import(foo, meshlink_export(bar)); + meshlink_close(foo); + meshlink_close(bar); - if(fork()) - return main1(fda[0], fdb[1]); - else - return main2(fdb[0], fda[1]); + if(fork()) { + return main1(); + } else { + return main2(); + } }