1 files changed, 38 insertions, 14 deletions

diff --git a/test/core/util/passthru_endpoint.c b/test/core/util/passthru_endpoint.c
index 7ed9e97bd6..b3405f02e9 100644
--- a/test/core/util/passthru_endpoint.c
+++ b/test/core/util/passthru_endpoint.c
@@ -33,17 +33,21 @@
 
 #include "test/core/util/passthru_endpoint.h"
 
+#include <inttypes.h>
+
 #include <grpc/support/alloc.h>
 #include <grpc/support/string_util.h>
+#include "src/core/lib/iomgr/sockaddr.h"
 
 typedef struct passthru_endpoint passthru_endpoint;
 
 typedef struct {
   grpc_endpoint base;
   passthru_endpoint *parent;
-  gpr_slice_buffer read_buffer;
-  gpr_slice_buffer *on_read_out;
+  grpc_slice_buffer read_buffer;
+  grpc_slice_buffer *on_read_out;
   grpc_closure *on_read;
+  grpc_resource_user *resource_user;
 } half;
 
 struct passthru_endpoint {
@@ -55,14 +59,14 @@ struct passthru_endpoint {
 };
 
 static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
-                    gpr_slice_buffer *slices, grpc_closure *cb) {
+                    grpc_slice_buffer *slices, grpc_closure *cb) {
   half *m = (half *)ep;
   gpr_mu_lock(&m->parent->mu);
   if (m->parent->shutdown) {
     grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_CREATE("Already shutdown"),
                         NULL);
   } else if (m->read_buffer.count > 0) {
-    gpr_slice_buffer_swap(&m->read_buffer, slices);
+    grpc_slice_buffer_swap(&m->read_buffer, slices);
     grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
   } else {
     m->on_read = cb;
@@ -77,7 +81,7 @@ static half *other_half(half *h) {
 }
 
 static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
-                     gpr_slice_buffer *slices, grpc_closure *cb) {
+                     grpc_slice_buffer *slices, grpc_closure *cb) {
   half *m = other_half((half *)ep);
   gpr_mu_lock(&m->parent->mu);
   grpc_error *error = GRPC_ERROR_NONE;
@@ -85,13 +89,13 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
     error = GRPC_ERROR_CREATE("Endpoint already shutdown");
   } else if (m->on_read != NULL) {
     for (size_t i = 0; i < slices->count; i++) {
-      gpr_slice_buffer_add(m->on_read_out, gpr_slice_ref(slices->slices[i]));
+      grpc_slice_buffer_add(m->on_read_out, grpc_slice_ref(slices->slices[i]));
     }
     grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_NONE, NULL);
     m->on_read = NULL;
   } else {
     for (size_t i = 0; i < slices->count; i++) {
-      gpr_slice_buffer_add(&m->read_buffer, gpr_slice_ref(slices->slices[i]));
+      grpc_slice_buffer_add(&m->read_buffer, grpc_slice_ref(slices->slices[i]));
     }
   }
   gpr_mu_unlock(&m->parent->mu);
@@ -120,6 +124,7 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
     m->on_read = NULL;
   }
   gpr_mu_unlock(&m->parent->mu);
+  grpc_resource_user_shutdown(exec_ctx, m->resource_user);
 }
 
 static void me_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
@@ -128,8 +133,10 @@ static void me_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   if (0 == --p->halves) {
     gpr_mu_unlock(&p->mu);
     gpr_mu_destroy(&p->mu);
-    gpr_slice_buffer_destroy(&p->client.read_buffer);
-    gpr_slice_buffer_destroy(&p->server.read_buffer);
+    grpc_slice_buffer_destroy(&p->client.read_buffer);
+    grpc_slice_buffer_destroy(&p->server.read_buffer);
+    grpc_resource_user_unref(exec_ctx, p->client.resource_user);
+    grpc_resource_user_unref(exec_ctx, p->server.resource_user);
     gpr_free(p);
   } else {
     gpr_mu_unlock(&p->mu);
@@ -140,8 +147,15 @@ static char *me_get_peer(grpc_endpoint *ep) {
   return gpr_strdup("fake:mock_endpoint");
 }
 
+static int me_get_fd(grpc_endpoint *ep) { return -1; }
+
 static grpc_workqueue *me_get_workqueue(grpc_endpoint *ep) { return NULL; }
 
+static grpc_resource_user *me_get_resource_user(grpc_endpoint *ep) {
+  half *m = (half *)ep;
+  return m->resource_user;
+}
+
 static const grpc_endpoint_vtable vtable = {
     me_read,
     me_write,
@@ -150,23 +164,33 @@ static const grpc_endpoint_vtable vtable = {
     me_add_to_pollset_set,
     me_shutdown,
     me_destroy,
+    me_get_resource_user,
     me_get_peer,
+    me_get_fd,
 };
 
-static void half_init(half *m, passthru_endpoint *parent) {
+static void half_init(half *m, passthru_endpoint *parent,
+                      grpc_resource_quota *resource_quota,
+                      const char *half_name) {
   m->base.vtable = &vtable;
   m->parent = parent;
-  gpr_slice_buffer_init(&m->read_buffer);
+  grpc_slice_buffer_init(&m->read_buffer);
   m->on_read = NULL;
+  char *name;
+  gpr_asprintf(&name, "passthru_endpoint_%s_%" PRIxPTR, half_name,
+               (intptr_t)parent);
+  m->resource_user = grpc_resource_user_create(resource_quota, name);
+  gpr_free(name);
 }
 
 void grpc_passthru_endpoint_create(grpc_endpoint **client,
-                                   grpc_endpoint **server) {
+                                   grpc_endpoint **server,
+                                   grpc_resource_quota *resource_quota) {
   passthru_endpoint *m = gpr_malloc(sizeof(*m));
   m->halves = 2;
   m->shutdown = 0;
-  half_init(&m->client, m);
-  half_init(&m->server, m);
+  half_init(&m->client, m, resource_quota, "client");
+  half_init(&m->server, m, resource_quota, "server");
   gpr_mu_init(&m->mu);
   *client = &m->client.base;
   *server = &m->server.base;