void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
                                 const char *reason) {
  gpr_log(GPR_DEBUG, "CHANNEL: unref %p %d -> %d [%s]", channel,
          channel->refs.count, channel->refs.count - 1, reason);
#else
void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx,
                                 grpc_channel *channel) {
#endif
  if (gpr_unref(&channel->refs)) {
    destroy_channel(exec_ctx, channel);
  }
}

void grpc_channel_destroy(grpc_channel *channel) {
  grpc_transport_op op;
  grpc_channel_element *elem;
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
  GRPC_API_TRACE("grpc_channel_destroy(channel=%p)", 1, (channel));
  memset(&op, 0, sizeof(op));
  op.disconnect = 1;
  elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CHANNEL(channel), 0);
  elem->filter->start_transport_op(&exec_ctx, elem, &op);

  GRPC_CHANNEL_INTERNAL_UNREF(&exec_ctx, channel, "channel");

  grpc_exec_ctx_finish(&exec_ctx);
}

grpc_auth_property_iterator grpc_auth_context_peer_identity(
    const grpc_auth_context *ctx) {
  GRPC_API_TRACE("grpc_auth_context_peer_identity(ctx=%p)", 1, (ctx));
  if (ctx == NULL) return empty_iterator;
  return grpc_auth_context_find_properties_by_name(
      ctx, ctx->peer_identity_property_name);
}

grpc_call_error grpc_server_request_call(
    grpc_server *server, grpc_call **call, grpc_call_details *details,
    grpc_metadata_array *initial_metadata,
    grpc_completion_queue *cq_bound_to_call,
    grpc_completion_queue *cq_for_notification, void *tag) {
  grpc_call_error error;
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
  requested_call *rc = gpr_malloc(sizeof(*rc));
  GRPC_API_TRACE(
      "grpc_server_request_call("
      "server=%p, call=%p, details=%p, initial_metadata=%p, "
      "cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)",
      7, (server, call, details, initial_metadata, cq_bound_to_call,
          cq_for_notification, tag));
  if (!grpc_cq_is_server_cq(cq_for_notification)) {
    gpr_free(rc);
    error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
    goto done;
  }
  grpc_cq_begin_op(cq_for_notification, tag);
  details->reserved = NULL;
  rc->type = BATCH_CALL;
  rc->server = server;
  rc->tag = tag;
  rc->cq_bound_to_call = cq_bound_to_call;
  rc->cq_for_notification = cq_for_notification;
  rc->call = call;
  rc->data.batch.details = details;
  rc->initial_metadata = initial_metadata;
  error = queue_call_request(&exec_ctx, server, rc);
done:
  grpc_exec_ctx_finish(&exec_ctx);
  return error;
}

void grpc_call_details_destroy(grpc_call_details* cd) {
  GRPC_API_TRACE("grpc_call_details_destroy(cd=%p)", 1, (cd));
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
  grpc_slice_unref_internal(&exec_ctx, cd->method);
  grpc_slice_unref_internal(&exec_ctx, cd->host);
  grpc_exec_ctx_finish(&exec_ctx);
}

grpc_channel *grpc_insecure_channel_create_from_fd(
    const char *target, int fd, const grpc_channel_args *args) {
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
  GRPC_API_TRACE("grpc_insecure_channel_create(target=%p, fd=%d, args=%p)", 3,
                 (target, fd, args));

  grpc_arg default_authority_arg;
  default_authority_arg.type = GRPC_ARG_STRING;
  default_authority_arg.key = GRPC_ARG_DEFAULT_AUTHORITY;
  default_authority_arg.value.string = "test.authority";
  grpc_channel_args *final_args =
      grpc_channel_args_copy_and_add(args, &default_authority_arg, 1);

  int flags = fcntl(fd, F_GETFL, 0);
  GPR_ASSERT(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);

  grpc_endpoint *client = grpc_tcp_client_create_from_fd(
      &exec_ctx, grpc_fd_create(fd, "client"), args, "fd-client");

  grpc_transport *transport =
      grpc_create_chttp2_transport(&exec_ctx, final_args, client, 1);
  GPR_ASSERT(transport);
  grpc_channel *channel = grpc_channel_create(
      &exec_ctx, target, final_args, GRPC_CLIENT_DIRECT_CHANNEL, transport);
  grpc_channel_args_destroy(final_args);
  grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);

  grpc_exec_ctx_finish(&exec_ctx);

  return channel != NULL ? channel : grpc_lame_client_channel_create(
                                         target, GRPC_STATUS_INTERNAL,
                                         "Failed to create client channel");
}

void grpc_server_start(grpc_server *server) {
  listener *l;
  size_t i;
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;

  GRPC_API_TRACE("grpc_server_start(server=%p)", 1, (server));

  server->started = true;
  size_t pollset_count = 0;
  server->pollsets = gpr_malloc(sizeof(grpc_pollset *) * server->cq_count);
  for (i = 0; i < server->cq_count; i++) {
    if (!grpc_cq_is_non_listening_server_cq(server->cqs[i])) {
      server->pollsets[pollset_count++] = grpc_cq_pollset(server->cqs[i]);
    }
  }
  request_matcher_init(&server->unregistered_request_matcher,
                       server->max_requested_calls, server);
  for (registered_method *rm = server->registered_methods; rm; rm = rm->next) {
    request_matcher_init(&rm->request_matcher, server->max_requested_calls,
                         server);
  }

  for (l = server->listeners; l; l = l->next) {
    l->start(&exec_ctx, server, l->arg, server->pollsets, pollset_count);
  }

  grpc_exec_ctx_finish(&exec_ctx);
}

int grpc_compression_algorithm_parse(const char *name, size_t name_length,
                                     grpc_compression_algorithm *algorithm) {
  /* we use strncmp not only because it's safer (even though in this case it
   * doesn't matter, given that we are comparing against string literals, but
   * because this way we needn't have "name" nil-terminated (useful for slice
   * data, for example) */
  GRPC_API_TRACE(
      "grpc_compression_algorithm_parse("
      "name=%*.*s, name_length=%lu, algorithm=%p)",
      5, ((int)name_length, (int)name_length, name, (unsigned long)name_length,
          algorithm));
  if (name_length == 0) {
    return 0;
  }
  if (strncmp(name, "identity", name_length) == 0) {
    *algorithm = GRPC_COMPRESS_NONE;
  } else if (strncmp(name, "gzip", name_length) == 0) {
    *algorithm = GRPC_COMPRESS_GZIP;
  } else if (strncmp(name, "deflate", name_length) == 0) {
    *algorithm = GRPC_COMPRESS_DEFLATE;
  } else {
    return 0;
  }
  return 1;
}

void *grpc_server_register_method(grpc_server *server, const char *method,
                                  const char *host) {
  registered_method *m;
  GRPC_API_TRACE("grpc_server_register_method(server=%p, method=%s, host=%s)",
                 3, (server, method, host));
  if (!method) {
    gpr_log(GPR_ERROR,
            "grpc_server_register_method method string cannot be NULL");
    return NULL;
  }
  for (m = server->registered_methods; m; m = m->next) {
    if (streq(m->method, method) && streq(m->host, host)) {
      gpr_log(GPR_ERROR, "duplicate registration for %[email protected]%s", method,
              host ? host : "*");
      return NULL;
    }
  }
  m = gpr_malloc(sizeof(registered_method));
  memset(m, 0, sizeof(*m));
  request_matcher_init(&m->request_matcher, server->max_requested_calls);
  m->method = gpr_strdup(method);
  m->host = gpr_strdup(host);
  m->next = server->registered_methods;
  server->registered_methods = m;
  return m;
}

GRPCAPI grpc_stream_compression_algorithm
grpc_stream_compression_algorithm_for_level(
    grpc_stream_compression_level level, uint32_t accepted_stream_encodings) {
  GRPC_API_TRACE("grpc_stream_compression_algorithm_for_level(level=%d)", 1,
                 ((int)level));
  if (level > GRPC_STREAM_COMPRESS_LEVEL_HIGH) {
    gpr_log(GPR_ERROR, "Unknown compression level %d.", (int)level);
    abort();
  }

  switch (level) {
    case GRPC_STREAM_COMPRESS_LEVEL_NONE:
      return GRPC_STREAM_COMPRESS_NONE;
    case GRPC_STREAM_COMPRESS_LEVEL_LOW:
    case GRPC_STREAM_COMPRESS_LEVEL_MED:
    case GRPC_STREAM_COMPRESS_LEVEL_HIGH:
      if (GPR_BITGET(accepted_stream_encodings, GRPC_STREAM_COMPRESS_GZIP) ==
          1) {
        return GRPC_STREAM_COMPRESS_GZIP;
      } else {
        return GRPC_STREAM_COMPRESS_NONE;
      }
    default:
      abort();
  }
}

grpc_call *grpc_channel_create_call(grpc_channel *channel,
                                    grpc_call *parent_call,
                                    uint32_t propagation_mask,
                                    grpc_completion_queue *cq,
                                    const char *method, const char *host,
                                    gpr_timespec deadline, void *reserved) {
  GRPC_API_TRACE(
      "grpc_channel_create_call("
      "channel=%p, parent_call=%p, propagation_mask=%x, cq=%p, method=%s, "
      "host=%s, "
      "deadline=gpr_timespec { tv_sec: %lld, tv_nsec: %d, clock_type: %d }, "
      "reserved=%p)",
      10, (channel, parent_call, (unsigned)propagation_mask, cq, method, host,
           (long long)deadline.tv_sec, (int)deadline.tv_nsec,
           (int)deadline.clock_type, reserved));
  GPR_ASSERT(!reserved);
  return grpc_channel_create_call_internal(
      channel, parent_call, propagation_mask, cq,
      grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH,
                                        grpc_mdstr_from_string(method)),
      host ? grpc_mdelem_from_metadata_strings(GRPC_MDSTR_AUTHORITY,
                                               grpc_mdstr_from_string(host))
           : NULL,
      deadline);
}

void grpc_call_destroy(grpc_call *c) {
  int cancel;
  grpc_call *parent = c->parent;
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;

  GPR_TIMER_BEGIN("grpc_call_destroy", 0);
  GRPC_API_TRACE("grpc_call_destroy(c=%p)", 1, (c));

  if (parent) {
    gpr_mu_lock(&parent->mu);
    if (c == parent->first_child) {
      parent->first_child = c->sibling_next;
      if (c == parent->first_child) {
        parent->first_child = NULL;
      }
      c->sibling_prev->sibling_next = c->sibling_next;
      c->sibling_next->sibling_prev = c->sibling_prev;
    }
    gpr_mu_unlock(&parent->mu);
    GRPC_CALL_INTERNAL_UNREF(&exec_ctx, parent, "child");
  }

  gpr_mu_lock(&c->mu);
  GPR_ASSERT(!c->destroy_called);
  c->destroy_called = 1;
  if (c->have_alarm) {
    grpc_timer_cancel(&exec_ctx, &c->alarm);
  }
  cancel = !c->received_final_op;
  gpr_mu_unlock(&c->mu);
  if (cancel) grpc_call_cancel(c, NULL);
  GRPC_CALL_INTERNAL_UNREF(&exec_ctx, c, "destroy");
  grpc_exec_ctx_finish(&exec_ctx);
  GPR_TIMER_END("grpc_call_destroy", 0);
}

/* Create a client channel:
   Asynchronously: - resolve target
                   - connect to it (trying alternatives as presented)
                   - perform handshakes */
grpc_channel *grpc_insecure_channel_create(const char *target,
                                           const grpc_channel_args *args,
                                           void *reserved) {
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
  GRPC_API_TRACE(
      "grpc_insecure_channel_create(target=%p, args=%p, reserved=%p)", 3,
      (target, args, reserved));
  GPR_ASSERT(!reserved);

  client_channel_factory *f = gpr_malloc(sizeof(*f));
  memset(f, 0, sizeof(*f));
  f->base.vtable = &client_channel_factory_vtable;
  gpr_ref_init(&f->refs, 1);
  f->merge_args = grpc_channel_args_copy(args);

  grpc_channel *channel = client_channel_factory_create_channel(
      &exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, NULL);
  if (channel != NULL) {
    f->master = channel;
    GRPC_CHANNEL_INTERNAL_REF(f->master, "grpc_insecure_channel_create");
  }
  grpc_client_channel_factory_unref(&exec_ctx, &f->base);

  grpc_exec_ctx_finish(&exec_ctx);

  return channel != NULL ? channel : grpc_lame_client_channel_create(
                                         target, GRPC_STATUS_INTERNAL,
                                         "Failed to create client channel");
}

grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) {
  size_t i;

  GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved));

  grpc_server *server = gpr_malloc(sizeof(grpc_server));

  GPR_ASSERT(grpc_is_initialized() && "call grpc_init()");

  memset(server, 0, sizeof(grpc_server));

  gpr_mu_init(&server->mu_global);
  gpr_mu_init(&server->mu_call);

  /* decremented by grpc_server_destroy */
  gpr_ref_init(&server->internal_refcount, 1);
  server->root_channel_data.next = server->root_channel_data.prev =
      &server->root_channel_data;

  /* TODO(ctiller): expose a channel_arg for this */
  server->max_requested_calls = 32768;
  server->request_freelist =
      gpr_stack_lockfree_create(server->max_requested_calls);
  for (i = 0; i < (size_t)server->max_requested_calls; i++) {
    gpr_stack_lockfree_push(server->request_freelist, (int)i);
  }
  request_matcher_init(&server->unregistered_request_matcher,
                       server->max_requested_calls);
  server->requested_calls = gpr_malloc(server->max_requested_calls *
                                       sizeof(*server->requested_calls));

  server->channel_args = grpc_channel_args_copy(args);

  return server;
}

grpc_call_credentials *grpc_composite_call_credentials_create(
    grpc_call_credentials *creds1, grpc_call_credentials *creds2,
    void *reserved) {
  size_t i;
  size_t creds_array_byte_size;
  grpc_call_credentials_array creds1_array;
  grpc_call_credentials_array creds2_array;
  grpc_composite_call_credentials *c;
  GRPC_API_TRACE(
      "grpc_composite_call_credentials_create(creds1=%p, creds2=%p, "
      "reserved=%p)",
      3, (creds1, creds2, reserved));
  GPR_ASSERT(reserved == NULL);
  GPR_ASSERT(creds1 != NULL);
  GPR_ASSERT(creds2 != NULL);
  c = gpr_zalloc(sizeof(grpc_composite_call_credentials));
  c->base.type = GRPC_CALL_CREDENTIALS_TYPE_COMPOSITE;
  c->base.vtable = &composite_call_credentials_vtable;
  gpr_ref_init(&c->base.refcount, 1);
  creds1_array = get_creds_array(&creds1);
  creds2_array = get_creds_array(&creds2);
  c->inner.num_creds = creds1_array.num_creds + creds2_array.num_creds;
  creds_array_byte_size = c->inner.num_creds * sizeof(grpc_call_credentials *);
  c->inner.creds_array = gpr_zalloc(creds_array_byte_size);
  for (i = 0; i < creds1_array.num_creds; i++) {
    grpc_call_credentials *cur_creds = creds1_array.creds_array[i];
    c->inner.creds_array[i] = grpc_call_credentials_ref(cur_creds);
  }
  for (i = 0; i < creds2_array.num_creds; i++) {
    grpc_call_credentials *cur_creds = creds2_array.creds_array[i];
    c->inner.creds_array[i + creds1_array.num_creds] =
        grpc_call_credentials_ref(cur_creds);
  }
  return &c->base;
}

char *grpc_call_get_peer(grpc_call *call) {
  grpc_call_element *elem = CALL_ELEM_FROM_CALL(call, 0);
  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
  char *result = elem->filter->get_peer(&exec_ctx, elem);
  GRPC_API_TRACE("grpc_call_get_peer(%p)", 1, (call));
  grpc_exec_ctx_finish(&exec_ctx);
  return result;
}

void grpc_register_plugin(void (*init)(void), void (*destroy)(void)) {
  GRPC_API_TRACE("grpc_register_plugin(init=%lx, destroy=%lx)", 2,
                 ((unsigned long)init, (unsigned long)destroy));
  GPR_ASSERT(g_number_of_plugins != MAX_PLUGINS);
  g_all_of_the_plugins[g_number_of_plugins].init = init;
  g_all_of_the_plugins[g_number_of_plugins].destroy = destroy;
  g_number_of_plugins++;
}

void grpc_server_register_non_listening_completion_queue(
    grpc_server *server, grpc_completion_queue *cq, void *reserved) {
  GRPC_API_TRACE(
      "grpc_server_register_non_listening_completion_queue(server=%p, cq=%p, "
      "reserved=%p)",
      3, (server, cq, reserved));
  register_completion_queue(server, cq, true, reserved);
}

void grpc_server_register_completion_queue(grpc_server *server,
                                           grpc_completion_queue *cq,
                                           void *reserved) {
  GRPC_API_TRACE(
      "grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3,
      (server, cq, reserved));
  register_completion_queue(server, cq, false, reserved);
}

void grpc_register_plugin(void (*init)(void), void (*destroy)(void)) {
  GRPC_API_TRACE("grpc_register_plugin(init=%p, destroy=%p)", 2,
                 ((void *)(intptr_t)init, (void *)(intptr_t)destroy));
  GPR_ASSERT(g_number_of_plugins != MAX_PLUGINS);
  g_all_of_the_plugins[g_number_of_plugins].init = init;
  g_all_of_the_plugins[g_number_of_plugins].destroy = destroy;
  g_number_of_plugins++;
}

grpc_auth_property_iterator grpc_auth_context_property_iterator(
    const grpc_auth_context *ctx) {
  grpc_auth_property_iterator it = empty_iterator;
  GRPC_API_TRACE("grpc_auth_context_property_iterator(ctx=%p)", 1, (ctx));
  if (ctx == NULL) return it;
  it.ctx = ctx;
  return it;
}