Added un.pair

dimkr · Nov 23, 2017 · d9c69ab · d9c69ab
1 parent 95a9e24
commit d9c69ab
Show file tree

Hide file tree

Showing 4 changed files with 179 additions and 1 deletion.
diff --git a/docs/streams.md b/docs/streams.md
@@ -35,3 +35,7 @@
     {$un.server dgram /tmp/server 5}
 
 *un.server* creates a Unix socket server stream.
+
+    {$un.pair stream}
+
+*un.pair* creates a pair of connected Unix socket streams.
diff --git a/src/b6b_socket.c b/src/b6b_socket.c
@@ -651,6 +651,65 @@ static enum b6b_res b6b_socket_proc_un_server(struct b6b_interp *interp,
 	return b6b_return(interp, o);
 }
 
+static enum b6b_res b6b_socket_proc_un_pair(struct b6b_interp *interp,
+                                            struct b6b_obj *args)
+{
+	struct b6b_obj *t, *a, *b, *l;
+	const struct b6b_strm_ops *ops = &b6b_un_stream_client_ops;
+	int fds[2], type = SOCK_STREAM;
+
+	if (!b6b_proc_get_args(interp, args, "os", NULL, &t))
+		return B6B_ERR;
+
+	if (strcmp(t->s, "dgram") == 0) {
+		type = SOCK_DGRAM;
+		ops = &b6b_un_dgram_client_ops;
+	} else if (strcmp(t->s, "stream") != 0)
+		return B6B_ERR;
+
+	if (socketpair(AF_UNIX, type, 0, fds) < 0)
+		return b6b_return_strerror(interp, errno);
+
+	a = b6b_socket_new(interp,
+	                   fds[0],
+	                   NULL,
+	                   0,
+	                   NULL,
+	                   0,
+	                   "un.client",
+	                   ops);
+	if (!a) {
+		close(fds[1]);
+		close(fds[0]);
+		return B6B_ERR;
+	}
+
+	b = b6b_socket_new(interp,
+	                   fds[1],
+	                   NULL,
+	                   0,
+	                   NULL,
+	                   0,
+	                   "un.client",
+	                   ops);
+	if (!b) {
+		close(fds[1]);
+		b6b_destroy(a);
+		return B6B_ERR;
+	}
+
+	l = b6b_list_build(a, b, NULL);
+	if (b6b_unlikely(!l)) {
+		b6b_destroy(b);
+		b6b_destroy(a);
+		return B6B_ERR;
+	}
+
+	b6b_unref(b);
+	b6b_unref(a);
+	return b6b_return(interp, l);
+}
+
 static enum b6b_res b6b_socket_proc_bswap16(struct b6b_interp *interp,
                                             struct b6b_obj *args)
 {
@@ -710,6 +769,12 @@ static const struct b6b_ext_obj b6b_socket[] = {
 		.val.s = "un.server",
 		.proc = b6b_socket_proc_un_server
 	},
+	{
+		.name = "un.pair",
+		.type = B6B_TYPE_STR,
+		.val.s = "un.pair",
+		.proc = b6b_socket_proc_un_pair
+	},
 	{
 		.name = "htonl",
 		.type = B6B_TYPE_STR,

diff --git a/tests/b6b_test_socket.c b/tests/b6b_test_socket.c
@@ -0,0 +1,108 @@
+/*
+ * This file is part of b6b.
+ *
+ * Copyright 2017 Dima Krasner
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+#include <elf.h>
+
+#include <b6b.h>
+
+int main()
+{
+	struct b6b_interp interp;
+
+	/* bi-directional reading and writing should succeed */
+	assert(b6b_interp_new_argv(&interp, 0, NULL, B6B_OPT_TRACE));
+	assert(b6b_call_copy(&interp, "{$map {a b} [$un.pair dgram] {{$a write abcd} {$b write efgh} {$return [$list.new [$a read] [$b read]]}}}", 105) == B6B_RET);
+	assert(b6b_as_list(interp.fg->_));
+	assert(!b6b_list_empty(interp.fg->_));
+	assert(b6b_as_str(b6b_list_first(interp.fg->_)->o));
+	assert(b6b_list_first(interp.fg->_)->o->slen == 4);
+	assert(strcmp(b6b_list_first(interp.fg->_)->o->s, "efgh") == 0);
+	assert(b6b_list_next(b6b_list_first(interp.fg->_)));
+	assert(b6b_as_str(b6b_list_next(b6b_list_first(interp.fg->_))->o));
+	assert(b6b_list_next(b6b_list_first(interp.fg->_))->o->slen == 4);
+	assert(strcmp(b6b_list_next(b6b_list_first(interp.fg->_))->o->s, "abcd") == 0);
+	b6b_interp_destroy(&interp);
+
+	/* partial bi-directional reading and writing should succeed */
+	assert(b6b_interp_new_argv(&interp, 0, NULL, B6B_OPT_TRACE));
+	assert(b6b_call_copy(&interp, "{$map {a b} [$un.pair dgram] {{$a write abcd} {$b write efgh} {$return [$list.new [$a read 2] [$b read 3]]}}}", 109) == B6B_RET);
+	assert(b6b_as_list(interp.fg->_));
+	assert(!b6b_list_empty(interp.fg->_));
+	assert(b6b_as_str(b6b_list_first(interp.fg->_)->o));
+	assert(b6b_list_first(interp.fg->_)->o->slen == 2);
+	assert(strcmp(b6b_list_first(interp.fg->_)->o->s, "ef") == 0);
+	assert(b6b_list_next(b6b_list_first(interp.fg->_)));
+	assert(b6b_as_str(b6b_list_next(b6b_list_first(interp.fg->_))->o));
+	assert(b6b_list_next(b6b_list_first(interp.fg->_))->o->slen == 3);
+	assert(strcmp(b6b_list_next(b6b_list_first(interp.fg->_))->o->s, "abc") == 0);
+	b6b_interp_destroy(&interp);
+
+	/* multiple reads should succeed */
+	assert(b6b_interp_new_argv(&interp, 0, NULL, B6B_OPT_TRACE));
+	assert(b6b_call_copy(&interp, "{$map {a b} [$un.pair dgram] {{$a write abcd} {$a write efgh} {$return [$list.new [$b read] [$b read]]}}}", 105) == B6B_RET);
+	assert(b6b_as_list(interp.fg->_));
+	assert(!b6b_list_empty(interp.fg->_));
+	assert(b6b_as_str(b6b_list_first(interp.fg->_)->o));
+	assert(b6b_list_first(interp.fg->_)->o->slen == 4);
+	assert(strcmp(b6b_list_first(interp.fg->_)->o->s, "abcd") == 0);
+	assert(b6b_list_next(b6b_list_first(interp.fg->_)));
+	assert(b6b_as_str(b6b_list_next(b6b_list_first(interp.fg->_))->o));
+	assert(b6b_list_next(b6b_list_first(interp.fg->_))->o->slen == 4);
+	assert(strcmp(b6b_list_next(b6b_list_first(interp.fg->_))->o->s, "efgh") == 0);
+	b6b_interp_destroy(&interp);
+
+	/* writing to a closed stream should fail */
+	assert(b6b_interp_new_argv(&interp, 0, NULL, B6B_OPT_TRACE));
+	assert(b6b_call_copy(&interp, "{$map {a b} [$un.pair dgram] {{$b close} {$a write efgh}}}", 58) == B6B_ERR);
+	b6b_interp_destroy(&interp);
+
+	/* two buffers written to a stream socket should be received in one piece */
+	assert(b6b_interp_new_argv(&interp, 0, NULL, B6B_OPT_TRACE));
+	assert(b6b_call_copy(&interp, "{$map {a b} [$un.pair stream] {{$a write abcd} {$a write efgh} {$return [$b read]}}}", 84) == B6B_RET);
+	assert(b6b_as_str(interp.fg->_));
+	assert(interp.fg->_->slen == 8);
+	assert(strcmp(interp.fg->_->s, "abcdefgh") == 0);
+	b6b_interp_destroy(&interp);
+
+	/* reading from a stream socket should succeed after the peer shuts down the connection */
+	assert(b6b_interp_new_argv(&interp, 0, NULL, B6B_OPT_TRACE));
+	assert(b6b_call_copy(&interp, "{$map {a b} [$un.pair stream] {{$a write abcd} {$a close} {$return [$b read]}}}", 79) == B6B_RET);
+	assert(b6b_as_str(interp.fg->_));
+	assert(interp.fg->_->slen == 4);
+	assert(strcmp(interp.fg->_->s, "abcd") == 0);
+	b6b_interp_destroy(&interp);
+
+	/* after a partial read, remaining data sent to a stream socket should be still queued for reading */
+	assert(b6b_interp_new_argv(&interp, 0, NULL, B6B_OPT_TRACE));
+	assert(b6b_call_copy(&interp, "{$map {a b} [$un.pair stream] {{$a write abcd} {$return [$list.new [$b read 2] [$b read]]}}}", 92) == B6B_RET);
+	assert(b6b_as_list(interp.fg->_));
+	assert(!b6b_list_empty(interp.fg->_));
+	assert(b6b_as_str(b6b_list_first(interp.fg->_)->o));
+	assert(b6b_list_first(interp.fg->_)->o->slen == 2);
+	assert(strcmp(b6b_list_first(interp.fg->_)->o->s, "ab") == 0);
+	assert(b6b_list_next(b6b_list_first(interp.fg->_)));
+	assert(b6b_as_str(b6b_list_next(b6b_list_first(interp.fg->_))->o));
+	assert(b6b_list_next(b6b_list_first(interp.fg->_))->o->slen == 2);
+	assert(strcmp(b6b_list_next(b6b_list_first(interp.fg->_))->o->s, "cd") == 0);
+	b6b_interp_destroy(&interp);
+
+	return EXIT_SUCCESS;
+}
diff --git a/tests/meson.build b/tests/meson.build
@@ -29,7 +29,8 @@ tests = [
 	['sh', 'slow', 10],
 	['math', 'quick', 5],
 	['bswap', 'quick', 5],
-	['rand', 'quick', 5]
+	['rand', 'quick', 5],
+	['socket', 'quick', 5]
 ]
 
 if get_option('with_threads')