EV-Etcd
view release on metacpan or search on metacpan
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);
if (opts && SvROK(opts) && SvTYPE(SvRV(opts)) == SVt_PVHV) {
grpc_call_cancel(pc->call, NULL);
}
pc = pc->next;
}
if (client->cq) {
grpc_completion_queue_shutdown(client->cq);
}
pthread_join(client->cq_thread, NULL);
pthread_mutex_lock(&client->queue_mutex);
queued_event_t *qe = client->event_queue;
while (qe) {
queued_event_t *next = qe->next;
free(qe);
qe = next;
}
client->event_queue = NULL;
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 */
( run in 0.621 second using v1.01-cache-2.11-cpan-3f962061493 )