example: initial

README.md

@@ -40,7 +40,12 @@ static uv_link_methods_t methods = {
   .read_stop = read_stop_impl,
   .write = write_impl,
   .try_write = try_write_impl,
-  .shutdown = shutdown_impl
+  .shutdown = shutdown_impl,
+
+  /* These will be used only when chaining two links together */
+
+  .alloc_cb_override = alloc_cb_impl,
+  .read_cb_override = read_cb_impl
 };
 
 uv_link_init(&link, &methods);

example/example.gyp

@@ -0,0 +1,20 @@
+{
+  "targets": [{
+    "target_name": "uv_link_t-example",
+    "type": "executable",
+
+    "include_dirs": [
+      "src"
+    ],
+
+    "dependencies": [
+      "../test/deps/libuv/uv.gyp:libuv",
+      "../uv_link_t.gyp:uv_link_t"
+    ],
+
+    "sources": [
+      "src/main.c",
+      "src/middle.c",
+    ],
+  }],
+}

example/src/main.c

@@ -0,0 +1,99 @@
+#include <assert.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "uv.h"
+#include "uv_link_t.h"
+
+/* Declaration of `middle_methods` */
+#include "middle.h"
+
+typedef struct client_s client_t;
+
+static uv_tcp_t server;
+
+struct client_s {
+  uv_tcp_t tcp;
+  uv_link_source_t source;
+  uv_link_t middle;
+  uv_link_observer_t observer;
+};
+
+static void read_cb(uv_link_observer_t* observer,
+                    ssize_t nread,
+                    const uv_buf_t* buf) {
+  client_t* client;
+
+  client = observer->link.data;
+
+  if (nread < 0) {
+    fprintf(stderr, "error or close\n");
+    uv_link_read_stop(&observer->link);
+    uv_close((uv_handle_t*) &client->tcp, NULL);
+    free(client);
+    return;
+  }
+
+  fprintf(stderr, "read \"%.*s\"\n", (int) nread, buf->base);
+}
+
+
+static void connection_cb(uv_stream_t* s, int status) {
+  int err;
+  client_t* client;
+
+  client = malloc(sizeof(*client));
+  assert(client != NULL);
+
+  err = uv_tcp_init(uv_default_loop(), &client->tcp);
+  assert(err == 0);
+
+  err = uv_accept(s, (uv_stream_t*) &client->tcp);
+  assert(err == 0);
+
+  err = uv_link_source_init(&client->source, (uv_stream_t*) &client->tcp);
+  assert(err == 0);
+
+  err = uv_link_init(&client->middle, &middle_methods);
+  assert(err == 0);
+
+  err = uv_link_chain(&client->source.link, &client->middle);
+  assert(err == 0);
+
+  err = uv_link_observer_init(&client->observer, &client->middle);
+  assert(err == 0);
+
+  client->observer.read_cb = read_cb;
+  client->observer.link.data = client;
+
+  err = uv_link_read_start(&client->observer.link);
+  assert(err == 0);
+}
+
+
+int main() {
+  static const int kBacklog = 128;
+
+  int err;
+  uv_loop_t* loop;
+  struct sockaddr_in addr;
+
+  loop = uv_default_loop();
+
+  err = uv_tcp_init(loop, &server);
+  assert(err == 0);
+
+  err = uv_ip4_addr("0.0.0.0", 9000, &addr);
+  assert(err == 0);
+
+  err = uv_tcp_bind(&server, (struct sockaddr*) &addr, 0);
+  assert(err == 0);
+
+  err = uv_listen((uv_stream_t*) &server, kBacklog, connection_cb);
+
+  err = uv_run(loop, UV_RUN_DEFAULT);
+  assert(err == 0);
+
+  return 0;
+}

example/src/middle.c

@@ -0,0 +1,58 @@
+#include <assert.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "uv_link_t.h"
+
+static int read_start_impl(uv_link_t* link) {
+  return uv_link_read_start(link->parent);
+}
+
+
+static int read_stop_impl(uv_link_t* link) {
+  return uv_link_read_stop(link->parent);
+}
+
+
+static void alloc_cb_override(uv_link_t* link,
+                              size_t suggested_size,
+                              uv_buf_t* buf) {
+  buf->base = malloc(suggested_size);
+  assert(buf->base != NULL);
+  buf->len = suggested_size;
+}
+
+
+static void read_cb_override(uv_link_t* link,
+                             ssize_t nread,
+                             const uv_buf_t* buf) {
+  const char* res;
+  uv_buf_t tmp;
+
+  if (nread >= 0) {
+    if (nread == 9 && strncmp(buf->base, "password\n", 9) == 0)
+      res = "welcome";
+    else
+      res = "go away";
+  } else {
+    res = "error";
+  }
+
+  free(buf->base);
+
+  uv_link_invoke_alloc_cb(link, strlen(res), &tmp);
+  assert(tmp.len >= strlen(res));
+
+  memcpy(tmp.base, res, strlen(res));
+  uv_link_invoke_read_cb(link, strlen(res), &tmp);
+}
+
+
+uv_link_methods_t middle_methods = {
+  .read_start = read_start_impl,
+  .read_stop = read_stop_impl,
+
+  /* Other doesn't matter in this example */
+  .alloc_cb_override = alloc_cb_override,
+  .read_cb_override = read_cb_override
+};

example/src/middle.h

@@ -0,0 +1,6 @@
+#ifndef EXAMPLE_SRC_MIDDLE_H_
+#define EXAMPLE_SRC_MIDDLE_H_
+
+uv_link_methods_t middle_methods;
+
+#endif  /* EXAMPLE_SRC_MIDDLE_H_ */

gyp_uv_link

@@ -52,11 +52,14 @@ def run_gyp(args):
 if __name__ == '__main__':
   args = sys.argv[1:]
 
-  if 'test' not in args:
-    args.append(os.path.join(os.path.abspath(root), 'uv_link_t.gyp'))
-  else:
+  if 'test' in args:
     args.append(os.path.join(os.path.abspath(root), 'test/test.gyp'))
     args = filter(lambda arg: arg != 'test', args)
+  elif 'example' in args:
+    args.append(os.path.join(os.path.abspath(root), 'example/example.gyp'))
+    args = filter(lambda arg: arg != 'example', args)
+  else:
+    args.append(os.path.join(os.path.abspath(root), 'uv_link_t.gyp'))
 
   common_fn  = os.path.join(os.path.abspath(root), 'common.gypi')
   options_fn = os.path.join(os.path.abspath(root), 'options.gypi')

include/uv_link_t.h

@@ -31,6 +31,10 @@ struct uv_link_methods_s {
                    unsigned int nbufs);
 
   int (*shutdown)(uv_link_t* link, uv_link_shutdown_cb cb);
+
+  /* Overriding callbacks */
+  uv_link_alloc_cb alloc_cb_override;
+  uv_link_read_cb read_cb_override;
 };
 
 struct uv_link_s {
@@ -40,6 +44,8 @@ struct uv_link_s {
   uv_link_alloc_cb alloc_cb;
   uv_link_read_cb read_cb;
 
+  void* data;
+
   /* Read-only after assigning initial values */
 
   /* Sort of virtual table */
@@ -53,10 +59,7 @@ struct uv_link_s {
 UV_EXTERN int uv_link_init(uv_link_t* link, uv_link_methods_t const* methods);
 UV_EXTERN void uv_link_close(uv_link_t* link);
 
-UV_EXTERN int uv_link_chain(uv_link_t* from,
-                            uv_link_t* to,
-                            uv_link_alloc_cb alloc_cb,
-                            uv_link_read_cb read_cb);
+UV_EXTERN int uv_link_chain(uv_link_t* from, uv_link_t* to);
 UV_EXTERN int uv_link_unchain(uv_link_t* from, uv_link_t* to);
 
 /* Use this to invoke methods */

src/uv_link_observer_t.c

@@ -81,7 +81,10 @@ static uv_link_methods_t uv_link_observer_methods = {
   .read_stop = uv_link_observer_read_stop,
   .write = uv_link_observer_write,
   .try_write = uv_link_observer_try_write,
-  .shutdown = uv_link_observer_shutdown
+  .shutdown = uv_link_observer_shutdown,
+
+  .alloc_cb_override = uv_link_observer_alloc_cb,
+  .read_cb_override = uv_link_observer_read_cb
 };
 
 
@@ -97,8 +100,7 @@ int uv_link_observer_init(uv_link_observer_t* observer,
 
   observer->target = target;
 
-  err = uv_link_chain(target, &observer->link, uv_link_observer_alloc_cb,
-                      uv_link_observer_read_cb);
+  err = uv_link_chain(target, &observer->link);
   if (err != 0) {
     uv_link_close(&observer->link);
     return err;

src/uv_link_t.c

@@ -41,10 +41,7 @@ void uv_link_close(uv_link_t* link) {
 }
 
 
-int uv_link_chain(uv_link_t* from,
-                  uv_link_t* to,
-                  uv_link_alloc_cb alloc_cb,
-                  uv_link_read_cb read_cb) {
+int uv_link_chain(uv_link_t* from, uv_link_t* to) {
   if (from->child != NULL || to->parent != NULL)
     return UV_EINVAL;
 
@@ -53,8 +50,8 @@ int uv_link_chain(uv_link_t* from,
 
   from->saved_alloc_cb = from->alloc_cb;
   from->saved_read_cb = from->read_cb;
-  from->alloc_cb = alloc_cb;
-  from->read_cb = read_cb;
+  from->alloc_cb = to->methods->alloc_cb_override;
+  from->read_cb = to->methods->read_cb_override;
 
   return 0;
 }