EV-Etcd
view release on metacpan or search on metacpan
return;
#ifndef __APPLE__
grpc_shutdown_blocking();
ev_etcd_grpc_held = 0;
ev_etcd_grpc_settling = 1;
#endif
}
/* Mortal, so a croak later in new() frees it */
static SV *slurp_pem(pTHX_ const char *opt, const char *path) {
PerlIO *fp = PerlIO_open(path, "r");
if (!fp)
croak("EV::Etcd: cannot open %s '%s': %s", opt, path, Strerror(errno));
SV *sv = sv_2mortal(newSVpvs(""));
char buf[8192];
SSize_t n;
while ((n = PerlIO_read(fp, buf, sizeof buf)) > 0)
sv_catpvn(sv, buf, n);
int failed = n < 0 || PerlIO_error(fp);
PerlIO_close(fp);
if (failed)
croak("EV::Etcd: cannot read %s '%s'", opt, path);
return sv;
}
/* 0 for off (non-positive or NaN), else whole milliseconds capped at a day */
static int seconds_to_ms(NV seconds) {
if (!(seconds > 0))
return 0;
if (seconds >= 86400)
return 86400000;
return seconds < 0.001 ? 1 : (int)(seconds * 1000);
}
/* A mortal copy: magic on a later option could change or free the original */
static char *option_pv(pTHX_ SV *val, STRLEN *len) {
STRLEN l;
const char *p = SvPV_nomg(val, l);
if (len) *len = l;
return SvPVX(sv_2mortal(newSVpvn(p, l)));
}
/* undef means no options; anything but a hash reference is a mistake */
static SV *opts_arg(pTHX_ SV *sv) {
SvGETMAGIC(sv);
if (!SvOK(sv))
return NULL;
if (!SvROK(sv) || SvTYPE(SvRV(sv)) != SVt_PVHV)
croak("EV::Etcd: options must be a hash reference");
return sv;
}
static int64_t hv_fetch_i64(pTHX_ HV *hv, const char *key, I32 klen) {
SV **svp = hv_fetch(hv, key, klen, 0);
if (!svp)
return 0;
SvGETMAGIC(*svp);
return SvOK(*svp) ? SvI64_nomg(*svp) : 0;
}
static int endpoint_scheme_len(const char *ep, int *is_https) {
*is_https = strnEQ(ep, "https://", 8);
return *is_https ? 8 : strnEQ(ep, "http://", 7) ? 7 : 0;
}
static void process_txn_response(pTHX_ pending_call_t *pc);
static void process_auth_response(pTHX_ pending_call_t *pc);
static void process_user_add_response(pTHX_ pending_call_t *pc);
static void process_user_delete_response(pTHX_ pending_call_t *pc);
static void process_user_change_password_response(pTHX_ pending_call_t *pc);
static void process_auth_enable_response(pTHX_ pending_call_t *pc);
static void process_auth_disable_response(pTHX_ pending_call_t *pc);
static void process_role_add_response(pTHX_ pending_call_t *pc);
static void process_role_delete_response(pTHX_ pending_call_t *pc);
static void process_role_get_response(pTHX_ pending_call_t *pc);
static void process_role_list_response(pTHX_ pending_call_t *pc);
static void process_role_grant_permission_response(pTHX_ pending_call_t *pc);
static void process_role_revoke_permission_response(pTHX_ pending_call_t *pc);
static void process_user_grant_role_response(pTHX_ pending_call_t *pc);
static void process_user_revoke_role_response(pTHX_ pending_call_t *pc);
static void process_user_get_response(pTHX_ pending_call_t *pc);
static void process_user_list_response(pTHX_ pending_call_t *pc);
static SV* response_op_to_hashref(pTHX_ Etcdserverpb__ResponseOp *op);
static void parse_request_ops(pTHX_ SV *src_av, Etcdserverpb__RequestOp ***dst_ops, size_t *dst_n);
/* Caller must Safefree the result */
static char* compute_prefix_range_end(const char *key, size_t key_len, size_t *out_len) {
size_t i = key_len;
while (i > 0 && (unsigned char)key[i - 1] == 0xFF) {
i--;
}
char *range_end;
if (i == 0) {
/* range_end "\0" means every key >= key */
Newx(range_end, 1, char);
range_end[0] = '\0';
*out_len = 1;
} else {
Newx(range_end, i, char);
memcpy(range_end, key, i);
((unsigned char *)range_end)[i - 1]++;
*out_len = i;
}
return range_end;
}
static void health_timer_callback(struct ev_loop *loop, ev_timer *w, int revents) {
dTHX;
ev_etcd_t *client = (ev_etcd_t *)((char *)w - offsetof(ev_etcd_t, health_timer));
(void)loop;
(void)revents;
if (!client->active) {
return;
}
grpc_connectivity_state state = grpc_channel_check_connectivity_state(client->channel, 0);
int was_healthy = client->is_healthy;
int is_healthy = state != GRPC_CHANNEL_TRANSIENT_FAILURE && state != GRPC_CHANNEL_SHUTDOWN;
int checked = client->current_endpoint;
if (was_healthy != is_healthy) {
client->is_healthy = is_healthy;
if (!is_healthy && client->endpoint_count > 1) {
etcd_rotate_endpoint(client);
}
if (client->health_callback) {
dSP;
CALLBACK_WINDOW_BEGIN(client);
ENTER;
SAVETMPS;
PUSHMARK(SP);
EXTEND(SP, 2);
PUSHs(sv_2mortal(newSViv(is_healthy)));
PUSHs(sv_2mortal(newSVpv(client->endpoints[checked], 0)));
PUTBACK;
CALL_SV_SAFE(client->health_callback, G_DISCARD);
FREETMPS;
LEAVE;
if (CALLBACK_WINDOW_END(client))
return;
}
}
}
/* No Perl on this thread. Until GRPC_QUEUE_SHUTDOWN: unread completions keep gRPC up */
static void *cq_thread_func(void *arg) {
ev_etcd_t *client = (ev_etcd_t *)arg;
for (;;) {
grpc_event event = grpc_completion_queue_next(client->cq,
gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
if (event.type == GRPC_QUEUE_SHUTDOWN) {
break;
}
if (event.type != GRPC_OP_COMPLETE) {
continue;
}
queued_event_t *qe = (queued_event_t *)malloc(sizeof(queued_event_t));
if (!qe) {
fprintf(stderr, "EV::Etcd: CRITICAL - malloc failed in gRPC thread, event dropped\n");
continue;
}
qe->tag = event.tag;
qe->success = event.success;
qe->next = NULL;
pthread_mutex_lock(&client->queue_mutex);
if (client->event_queue_tail) {
client->event_queue_tail->next = qe;
} else {
client->event_queue = qe;
}
client->event_queue_tail = qe;
pthread_mutex_unlock(&client->queue_mutex);
ev_async_send(EV_DEFAULT, &client->cq_async);
}
return NULL;
}
/* A DESTROY deferred from inside a callback, once in_callback is back to 0 */
void finish_client_destroy(pTHX_ ev_etcd_t *client) {
while (client->pending_calls) {
pending_call_t *pc = client->pending_calls;
client->pending_calls = pc->next;
grpc_metadata_array_destroy(&pc->initial_metadata);
grpc_metadata_array_destroy(&pc->trailing_metadata);
if (pc->recv_buffer) grpc_byte_buffer_destroy(pc->recv_buffer);
grpc_slice_unref(pc->status_details);
READ_STREAM_STATUS(kc, CALL_TYPE_LEASE_KEEPALIVE_STATUS);
process_keepalive_status(aTHX_ kc);
} else {
cleanup_keepalive(aTHX_ kc);
}
}
} else if (base->type == CALL_TYPE_ELECTION_OBSERVE_STATUS) {
observe_call_t *oc = (observe_call_t *)base;
if (oc->active) process_observe_status(aTHX_ oc);
else cleanup_observe(aTHX_ oc);
} else if (base->type == CALL_TYPE_ELECTION_OBSERVE_RECV) {
observe_call_t *oc = (observe_call_t *)base;
if (success && oc->active && oc->recv_buffer) {
process_observe_response(aTHX_ oc);
if (!client->active) return;
if (oc->active) {
observe_rearm_recv(aTHX_ oc);
} else {
cleanup_observe(aTHX_ oc);
}
} else if (oc->active) {
READ_STREAM_STATUS(oc, CALL_TYPE_ELECTION_OBSERVE_STATUS);
process_observe_status(aTHX_ oc);
} else {
cleanup_observe(aTHX_ oc);
}
} else if (base->type == CALL_TYPE_ELECTION_OBSERVE) {
observe_call_t *oc = (observe_call_t *)base;
if (success) {
if (oc->recv_buffer && oc->active) {
process_observe_response(aTHX_ oc);
if (!client->active) return;
}
if (oc->active) {
observe_rearm_recv(aTHX_ oc);
} else {
cleanup_observe(aTHX_ oc);
}
} else {
if (oc->active) {
READ_STREAM_STATUS(oc, CALL_TYPE_ELECTION_OBSERVE_STATUS);
process_observe_status(aTHX_ oc);
} else {
cleanup_observe(aTHX_ oc);
}
}
} else {
pending_call_t *pc = (pending_call_t *)base;
/* Before the handler: a DESTROY from its callback frees what is still listed */
unlink_pending_call(pc);
/* Freed at scope end, which an exit() from the callback also reaches */
ENTER;
SAVEDESTRUCTOR_X(free_handled_call, pc);
/* A failed batch can leave the zero-initialised GRPC_STATUS_OK */
grpc_status_code status = !success && pc->status == GRPC_STATUS_OK
? GRPC_STATUS_UNAVAILABLE : pc->status;
/* Before the callback, so a retry from it reaches the next endpoint */
etcd_endpoint_failed(client, &pc->base, status, pc->status_details);
if (success) {
switch (pc->base.type) {
case CALL_TYPE_RANGE:
process_range_response(aTHX_ pc);
break;
case CALL_TYPE_PUT:
process_put_response(aTHX_ pc);
break;
case CALL_TYPE_DELETE:
process_delete_response(aTHX_ pc);
break;
case CALL_TYPE_LEASE_GRANT:
process_lease_grant_response(aTHX_ pc);
break;
case CALL_TYPE_LEASE_REVOKE:
process_lease_revoke_response(aTHX_ pc);
break;
case CALL_TYPE_LEASE_TIME_TO_LIVE:
process_lease_time_to_live_response(aTHX_ pc);
break;
case CALL_TYPE_LEASE_LEASES:
process_lease_leases_response(aTHX_ pc);
break;
case CALL_TYPE_COMPACT:
process_compact_response(aTHX_ pc);
break;
case CALL_TYPE_STATUS:
process_status_response(aTHX_ pc);
break;
case CALL_TYPE_TXN:
process_txn_response(aTHX_ pc);
break;
case CALL_TYPE_AUTH:
process_auth_response(aTHX_ pc);
break;
case CALL_TYPE_USER_ADD:
process_user_add_response(aTHX_ pc);
break;
case CALL_TYPE_USER_DELETE:
process_user_delete_response(aTHX_ pc);
break;
case CALL_TYPE_USER_CHANGE_PASSWORD:
process_user_change_password_response(aTHX_ pc);
break;
case CALL_TYPE_AUTH_ENABLE:
process_auth_enable_response(aTHX_ pc);
break;
case CALL_TYPE_AUTH_DISABLE:
process_auth_disable_response(aTHX_ pc);
break;
case CALL_TYPE_ROLE_ADD:
process_role_add_response(aTHX_ pc);
break;
case CALL_TYPE_ROLE_DELETE:
process_role_delete_response(aTHX_ pc);
break;
case CALL_TYPE_ROLE_GET:
process_role_get_response(aTHX_ pc);
break;
static void process_user_get_response(pTHX_ pending_call_t *pc) {
BEGIN_RESPONSE_HANDLER(pc, "user_get");
Etcdserverpb__AuthUserGetResponse *resp;
UNPACK_RESPONSE(pc, resp, etcdserverpb__auth_user_get_response__unpack);
HV *result = newHV();
add_header_to_hv(aTHX_ result, resp->header);
AV *roles = newAV();
for (size_t i = 0; i < resp->n_roles; i++) {
av_push(roles, resp->roles[i] ? newSVpv(resp->roles[i], 0) : newSVpvn("", 0));
}
hv_store(result, "roles", 5, newRV_noinc((SV *)roles), 0);
etcdserverpb__auth_user_get_response__free_unpacked(resp, NULL);
CALL_SUCCESS_CALLBACK(pc->callback, result);
}
static void process_user_list_response(pTHX_ pending_call_t *pc) {
BEGIN_RESPONSE_HANDLER(pc, "user_list");
Etcdserverpb__AuthUserListResponse *resp;
UNPACK_RESPONSE(pc, resp, etcdserverpb__auth_user_list_response__unpack);
HV *result = newHV();
add_header_to_hv(aTHX_ result, resp->header);
AV *users = newAV();
for (size_t i = 0; i < resp->n_users; i++) {
av_push(users, resp->users[i] ? newSVpv(resp->users[i], 0) : newSVpvn("", 0));
}
hv_store(result, "users", 5, newRV_noinc((SV *)users), 0);
etcdserverpb__auth_user_list_response__free_unpacked(resp, NULL);
CALL_SUCCESS_CALLBACK(pc->callback, result);
}
MODULE = EV::Etcd PACKAGE = EV::Etcd PREFIX = ev_etcd_
PROTOTYPES: DISABLE
BOOT:
I_EV_API("EV::Etcd");
watch_init_ev_api(aTHX);
lease_init_ev_api(aTHX);
election_init_ev_api(aTHX);
init_method_slices();
ev_etcd_pid = getpid();
pthread_atfork(ev_etcd_atfork_prepare, NULL, ev_etcd_atfork_child);
SV *
ev_etcd_new(class, ...)
char *class
CODE:
{
ev_etcd_t *client;
AV *endpoints_av = NULL;
int timeout_seconds = 30;
int max_retries = 30;
NV health_interval = 0;
NV keepalive_time = 10, keepalive_timeout = 10;
SV *health_callback = NULL;
char *init_auth_token = NULL;
STRLEN init_auth_token_len = 0;
int tls = 0, saw_http = 0;
const char *tls_ca_file = NULL, *tls_cert_file = NULL, *tls_key_file = NULL;
const char *tls_server_name = NULL;
SV *ca_pem = NULL, *cert_pem = NULL, *key_pem = NULL;
int i;
if ((items - 1) % 2) croak("Odd number of options in EV::Etcd->new");
for (i = 1; i < items; i += 2) {
if (i + 1 < items) {
const char *key = SvPV_nolen(ST(i));
SV *val = ST(i + 1);
SvGETMAGIC(val);
if (strEQ(key, "endpoints")) {
if (SvOK(val)) {
if (!SvROK(val) || SvTYPE(SvRV(val)) != SVt_PVAV)
croak("EV::Etcd: endpoints must be an array reference");
endpoints_av = (AV *)sv_2mortal(SvREFCNT_inc(SvRV(val)));
if (av_len(endpoints_av) < 0)
croak("EV::Etcd: endpoints is empty");
}
} else if (strEQ(key, "timeout")) {
if (SvOK(val)) {
IV v = SvIV_nomg(val);
timeout_seconds = v < 1 ? 1 : v > INT_MAX ? INT_MAX : (int)v;
}
} else if (strEQ(key, "max_retries")) {
if (SvOK(val)) {
IV v = SvIV_nomg(val);
max_retries = v < 0 ? 0 : v > INT_MAX ? INT_MAX : (int)v;
}
} else if (strEQ(key, "health_interval")) {
if (SvOK(val)) {
health_interval = SvNV_nomg(val);
if (!(health_interval > 0)) {
health_interval = 0;
}
}
} else if (strEQ(key, "on_health_change")) {
if (SvOK(val)) {
if (!SvROK(val) || SvTYPE(SvRV(val)) != SVt_PVCV)
croak("EV::Etcd: on_health_change must be a code reference");
health_callback = sv_2mortal(newRV_inc(SvRV(val)));
}
} else if (strEQ(key, "auth_token")) {
if (SvOK(val)) {
STRLEN j;
init_auth_token = option_pv(aTHX_ val, &init_auth_token_len);
/* gRPC rejects other bytes in metadata, failing every call */
for (j = 0; j < init_auth_token_len; j++) {
if (init_auth_token[j] < 0x20 || init_auth_token[j] > 0x7E)
croak("EV::Etcd: auth_token must be printable ASCII");
}
}
} else if (strEQ(key, "tls")) {
tls = SvTRUE_nomg(val);
} else if (strEQ(key, "tls_ca_file")) {
if (SvOK(val)) tls_ca_file = option_pv(aTHX_ val, NULL);
} else if (strEQ(key, "tls_cert_file")) {
if (SvOK(val)) tls_cert_file = option_pv(aTHX_ val, NULL);
} else if (strEQ(key, "tls_key_file")) {
if (SvOK(val)) tls_key_file = option_pv(aTHX_ val, NULL);
} else if (strEQ(key, "tls_server_name")) {
if (SvOK(val)) tls_server_name = option_pv(aTHX_ val, NULL);
} else if (strEQ(key, "keepalive_time")) {
if (SvOK(val)) keepalive_time = SvNV_nomg(val);
} else if (strEQ(key, "keepalive_timeout")) {
if (SvOK(val)) keepalive_timeout = SvNV_nomg(val);
} else {
croak("Unknown option '%s' in EV::Etcd->new", key);
}
}
}
/* Only these copies are used: a tied array could hand back something else */
AV *checked = NULL;
if (endpoints_av && av_len(endpoints_av) >= 0) {
int count = av_len(endpoints_av) + 1;
checked = (AV *)sv_2mortal((SV *)newAV());
for (i = 0; i < count; i++) {
SV **ep = av_fetch(endpoints_av, i, 0);
STRLEN len;
const char *str;
int is_https, skip;
if (ep)
SvGETMAGIC(*ep);
if (!ep || !SvOK(*ep))
croak("EV::Etcd: endpoints element %d is undefined", i);
str = SvPV_nomg(*ep, len);
VALIDATE_URL_SIZE(len);
VALIDATE_NO_NUL(str, len, "endpoint");
skip = endpoint_scheme_len(str, &is_https);
if (len <= (STRLEN)skip)
croak("EV::Etcd: endpoints element %d is empty", i);
av_push(checked, newSVpvn(str + skip, len - skip));
if (!skip)
continue;
if (is_https)
tls = 1;
else
saw_http = 1;
}
}
if (tls_ca_file || tls_cert_file || tls_key_file || tls_server_name)
tls = 1;
if (tls && saw_http)
croak("EV::Etcd: http:// endpoint on a TLS client");
if (!tls_cert_file != !tls_key_file)
croak("EV::Etcd: tls_cert_file and tls_key_file must be given together");
if (tls_ca_file) ca_pem = slurp_pem(aTHX_ "tls_ca_file", tls_ca_file);
if (tls_cert_file) cert_pem = slurp_pem(aTHX_ "tls_cert_file", tls_cert_file);
if (tls_key_file) key_pem = slurp_pem(aTHX_ "tls_key_file", tls_key_file);
/* Its fork-safety croak must come before anything is allocated */
grpc_acquire(aTHX);
Newxz(client, 1, ev_etcd_t);
if (checked) {
int count = av_len(checked) + 1;
Newx(client->endpoints, count, char *);
client->endpoint_count = count;
for (i = 0; i < count; i++) {
SV *ep = *av_fetch(checked, i, 0);
client->endpoints[i] = savepvn(SvPVX(ep), SvCUR(ep));
}
} else {
Newx(client->endpoints, 1, char *);
client->endpoints[0] = savepv("127.0.0.1:2379");
client->endpoint_count = 1;
}
client->current_endpoint = 0;
client->keepalive_ms = seconds_to_ms(keepalive_time);
client->keepalive_timeout_ms = seconds_to_ms(keepalive_timeout);
if (!client->keepalive_timeout_ms)
client->keepalive_timeout_ms = 10000;
if (tls) {
grpc_ssl_pem_key_cert_pair pair;
if (cert_pem) {
pair.private_key = SvPV_nolen(key_pem);
pair.cert_chain = SvPV_nolen(cert_pem);
}
client->creds = grpc_ssl_credentials_create(
ca_pem ? SvPV_nolen(ca_pem) : NULL, cert_pem ? &pair : NULL, NULL, NULL);
if (!client->creds) {
for (int j = 0; j < client->endpoint_count; j++) {
Safefree(client->endpoints[j]);
}
Safefree(client->endpoints);
Safefree(client);
grpc_release();
croak("EV::Etcd: failed to create TLS credentials");
}
if (key_pem)
memset(SvPVX(key_pem), 0, SvCUR(key_pem));
if (tls_server_name)
client->tls_server_name = savepv(tls_server_name);
}
client->channel_ref = etcd_create_channel(client, client->endpoints[0]);
if (!client->channel_ref) {
for (int j = 0; j < client->endpoint_count; j++) {
Safefree(client->endpoints[j]);
}
Safefree(client->endpoints);
if (client->creds) grpc_channel_credentials_release(client->creds);
Safefree(client->tls_server_name);
Safefree(client);
grpc_release();
croak("Failed to create gRPC channel");
}
client->channel = client->channel_ref->channel;
client->cq = grpc_completion_queue_create_for_next(NULL);
pthread_mutex_init(&client->queue_mutex, NULL);
ev_async_init(&client->cq_async, cq_async_callback);
ev_async_start(EV_DEFAULT, &client->cq_async);
int thread_error;
WITH_SIGNALS_BLOCKED(thread_error = pthread_create(&client->cq_thread, NULL, cq_thread_func, client));
if (thread_error) {
ev_async_stop(EV_DEFAULT, &client->cq_async);
pthread_mutex_destroy(&client->queue_mutex);
grpc_completion_queue_shutdown(client->cq);
while (grpc_completion_queue_next(client->cq,
gpr_inf_past(GPR_CLOCK_REALTIME), NULL).type != GRPC_QUEUE_SHUTDOWN)
;
grpc_completion_queue_destroy(client->cq);
etcd_channel_release(client->channel_ref, 1);
for (int j = 0; j < client->endpoint_count; j++) {
Safefree(client->endpoints[j]);
}
Safefree(client->endpoints);
if (client->creds) grpc_channel_credentials_release(client->creds);
Safefree(client->tls_server_name);
Safefree(client);
grpc_release();
croak("Failed to create gRPC completion queue thread");
}
if (init_auth_token && init_auth_token_len > 0) {
Newx(client->auth_token, init_auth_token_len + 1, char);
Copy(init_auth_token, client->auth_token, init_auth_token_len, char);
client->auth_token[init_auth_token_len] = '\0';
client->auth_token_len = init_auth_token_len;
}
client->timeout_seconds = timeout_seconds;
client->active = 1;
client->owner_pid = ev_etcd_pid;
client->next_live = ev_etcd_clients;
ev_etcd_clients = client;
client->max_retries = max_retries;
client->is_healthy = 1;
if (health_callback)
client->health_callback = newSVsv(health_callback);
ev_timer_init(&client->health_timer, health_timer_callback, 0.0, 0.0);
if (health_interval > 0) {
ev_timer_set(&client->health_timer, health_interval, health_interval);
ev_timer_start(EV_DEFAULT, &client->health_timer);
}
RETVAL = sv_setref_pv(newSV(0),
sv_isobject(ST(0)) ? sv_reftype(SvRV(ST(0)), TRUE) : class, (void *)client);
}
OUTPUT:
RETVAL
void
ev_etcd_get(client, key, ...)
EV::Etcd client
SV *key
CODE:
{
SV *opts = NULL;
SV *callback;
if (items == 3) {
callback = ST(2);
} else if (items == 4) {
opts = opts_arg(aTHX_ ST(2));
callback = ST(3);
} else {
croak("Usage: $client->get($key, [\\%%opts,] $callback)");
}
VALIDATE_CALLBACK(callback);
STRLEN key_len;
const char *key_str = SvPV(key, key_len);
VALIDATE_KEY_SIZE(key_len);
client->event_queue_tail = NULL;
pthread_mutex_unlock(&client->queue_mutex);
pthread_mutex_destroy(&client->queue_mutex);
if (client->cq) {
grpc_completion_queue_destroy(client->cq);
}
/* Inside a callback, finish_client_destroy does this once it returns */
if (!client->in_callback) {
pc = client->pending_calls;
while (pc) {
pending_call_t *next = pc->next;
grpc_metadata_array_destroy(&pc->initial_metadata);
grpc_metadata_array_destroy(&pc->trailing_metadata);
if (pc->recv_buffer) {
grpc_byte_buffer_destroy(pc->recv_buffer);
}
grpc_slice_unref(pc->status_details);
if (pc->call) {
grpc_call_unref(pc->call);
}
etcd_call_release(&pc->base);
SvREFCNT_dec(pc->callback);
Safefree(pc);
pc = next;
}
/* cleanup_* frees the struct only if its Perl handle is gone, else *_DESTROY does */
while (client->watches) cleanup_watch(aTHX_ client->watches);
while (client->keepalives) cleanup_keepalive(aTHX_ client->keepalives);
while (client->observes) cleanup_observe(aTHX_ client->observes);
}
ev_timer_stop(EV_DEFAULT, &client->health_timer);
etcd_channel_release(client->channel_ref, 1);
client->channel_ref = NULL;
client->channel = NULL;
if (client->creds) {
grpc_channel_credentials_release(client->creds);
client->creds = NULL;
}
if (!client->in_callback)
grpc_release();
free_perl_resources:
etcd_channel_release(client->channel_ref, 0);
client->channel_ref = NULL;
if (client->health_callback) {
SvREFCNT_dec(client->health_callback);
client->health_callback = NULL;
}
clear_auth_token(client);
Safefree(client->tls_server_name);
client->tls_server_name = NULL;
if (client->endpoints) {
int i;
for (i = 0; i < client->endpoint_count; i++) {
if (client->endpoints[i]) {
Safefree(client->endpoints[i]);
}
}
Safefree(client->endpoints);
client->endpoints = NULL;
}
/* Inside a callback, finish_client_destroy frees it later */
if (!client->in_callback) {
Safefree(client);
}
}
MODULE = EV::Etcd PACKAGE = EV::Etcd::Watch PREFIX = ev_etcd_watch_
void
ev_etcd_watch_cancel(watch, callback)
EV::Etcd::Watch watch
SV *callback
CODE:
{
VALIDATE_CALLBACK(callback);
watch_call_t *wc = watch;
if (!wc->client_owns) {
CALL_SYNC_SUCCESS_CALLBACK(callback, newHV());
XSRETURN_EMPTY;
}
/* A fired timer stays pending until its callback: armed too, and must not resurrect */
int timer_was_armed = ev_is_active(&wc->reconnect_timer)
|| ev_is_pending(&wc->reconnect_timer);
ev_timer_stop(EV_DEFAULT, &wc->reconnect_timer);
ev_timer_stop(EV_DEFAULT, &wc->progress_timer);
if (!wc->active) {
/* In backoff nothing is in flight; otherwise the pending RECV cleans up */
if (timer_was_armed)
cleanup_watch(aTHX_ wc);
CALL_SYNC_SUCCESS_CALLBACK(callback, newHV());
XSRETURN_EMPTY;
}
wc->active = 0;
if (wc->watch_id >= 0) {
Etcdserverpb__WatchCancelRequest cancel_req = ETCDSERVERPB__WATCH_CANCEL_REQUEST__INIT;
cancel_req.watch_id = wc->watch_id;
Etcdserverpb__WatchRequest req = ETCDSERVERPB__WATCH_REQUEST__INIT;
req.request_union_case = ETCDSERVERPB__WATCH_REQUEST__REQUEST_UNION_CANCEL_REQUEST;
req.cancel_request = &cancel_req;
grpc_slice req_slice;
SERIALIZE_PROTOBUF_TO_SLICE(req_slice,
etcdserverpb__watch_request__get_packed_size,
etcdserverpb__watch_request__pack, &req);
grpc_byte_buffer *send_buffer = grpc_raw_byte_buffer_create(&req_slice, 1);
grpc_slice_unref(req_slice);
grpc_op op;
memset(&op, 0, sizeof(op));
op.op = GRPC_OP_SEND_MESSAGE;
op.data.send_message.send_message = send_buffer;
( run in 1.234 second using v1.01-cache-2.11-cpan-16aabcbfbcb )