view release on metacpan or  search on metacpan

etcd_common.h  view on Meta::CPAN

    struct keepalive_call *next;
    int auto_reconnect;
    int established;
    int reconnect_attempt;
    unsigned attempt_epoch;
    ev_timer reconnect_timer;
    ev_timer renew_timer;
    int client_owns;           /* dual ownership, see watch_call_t */
    int perl_owns;
} keepalive_call_t;

typedef struct observe_params {
    char *name;
    size_t name_len;
} observe_params_t;

typedef struct observe_call {
    call_base_t base;  /* Must be first */
    grpc_call *call;
    SV *callback;
    grpc_metadata_array initial_metadata;
    grpc_metadata_array trailing_metadata;
    grpc_byte_buffer *recv_buffer;
    grpc_status_code status;
    grpc_slice status_details;
    int active;
    struct ev_etcd_struct *client;
    struct observe_call *next;
    int auto_reconnect;
    int established;
    int reconnect_attempt;
    unsigned attempt_epoch;
    ev_timer reconnect_timer;
    observe_params_t params;
    int client_owns;           /* dual ownership, see watch_call_t */
    int perl_owns;
} observe_call_t;

typedef struct ev_etcd_struct {
    grpc_channel *channel;
    channel_ref_t *channel_ref;
    grpc_completion_queue *cq;

    /* cq_thread hands completions to the EV thread via event_queue + cq_async */
    pthread_t cq_thread;
    pthread_mutex_t queue_mutex; /* Protects event_queue */
    ev_async cq_async;
    queued_event_t *event_queue;
    queued_event_t *event_queue_tail;

    pending_call_t *pending_calls;
    watch_call_t *watches;
    keepalive_call_t *keepalives;
    observe_call_t *observes;
    int active;
    int in_callback;  /* Guard against freeing client during event processing */
    char *auth_token;
    size_t auth_token_len;
    int timeout_seconds;

    char **endpoints;
    int endpoint_count;
    int current_endpoint;
    unsigned channel_gen;  /* bumped on every endpoint switch */

    grpc_channel_credentials *creds;  /* NULL = insecure */
    char *tls_server_name;
    int keepalive_ms;          /* 0 = no keepalive pings */
    int keepalive_timeout_ms;

    int max_retries;
    /* Bumped by each answer without a leader: voids reconnects counted before it */
    unsigned no_leader_epoch;

    ev_timer health_timer;
    int is_healthy;
    SV *health_callback;
    pid_t owner_pid;
    struct ev_etcd_struct *next_live;  /* this process's live clients */
} ev_etcd_t;

static inline void link_pending_call(ev_etcd_t *client, pending_call_t *pc) {
    pc->next = client->pending_calls;
    if (pc->next)
        pc->next->pprev = &pc->next;
    pc->pprev = &client->pending_calls;
    client->pending_calls = pc;
}

static inline void unlink_pending_call(pending_call_t *pc) {
    if (!pc->pprev)
        return;
    *pc->pprev = pc->next;
    if (pc->next)
        pc->next->pprev = pc->pprev;
    pc->pprev = NULL;
}

typedef ev_etcd_t *EV__Etcd;
typedef watch_call_t *EV__Etcd__Watch;
typedef keepalive_call_t *EV__Etcd__Keepalive;
typedef observe_call_t *EV__Etcd__Observe;

static inline void init_call_base(call_base_t *base, call_type_t type) {
    base->type = type;
}

static inline void etcd_call_acquire(ev_etcd_t *client, call_base_t *base) {
    base->channel_gen = client->channel_gen;
    base->channel_ref = client->channel_ref;
    base->channel_ref->refs++;
}

#define VALIDATE_CALLBACK(cb) \
    do { \
        if (!SvROK(cb) || SvTYPE(SvRV(cb)) != SVt_PVCV) { \
            croak("callback must be a code reference"); \
        } \
    } while (0)

/* 32-bit IV perls go through NV, exact only to 2^53: fine for revisions,
 * lossy for lease, member and cluster ids, which use all 63/64 bits */
#if IVSIZE >= 8
#  define newSVi64(v) newSViv((IV)(v))
#  define newSVu64(v) newSVuv((UV)(v))
#  define SvI64(sv)   ((int64_t)SvIV(sv))
#  define SvI64_nomg(sv) ((int64_t)SvIV_nomg(sv))
#  define SvU64(sv)   ((uint64_t)SvUV(sv))
#else
#  define newSVi64(v) newSVnv((NV)(v))
#  define newSVu64(v) newSVnv((NV)(v))
#  define SvI64(sv)   ((int64_t)SvNV(sv))
#  define SvI64_nomg(sv) ((int64_t)SvNV_nomg(sv))
#  define SvU64(sv)   ((uint64_t)SvNV(sv))
#endif

const char* grpc_status_name(grpc_status_code code);
int is_retryable_status(grpc_status_code code);
int is_permanent_status(grpc_status_code code);
int etcd_is_no_leader(grpc_status_code status, grpc_slice status_details);
SV* create_error_hv(pTHX_ grpc_status_code code, const char *message, size_t message_len, const char *source);
SV* create_pending_error_hv(pTHX_ pending_call_t *pc, grpc_status_code code, const char *source);

SV* kv_to_hashref(pTHX_ Mvccpb__KeyValue *kv);
SV* event_to_hashref(pTHX_ Mvccpb__Event *event);
void add_header_to_hv(pTHX_ HV *result, Etcdserverpb__ResponseHeader *header);

channel_ref_t *etcd_create_channel(ev_etcd_t *client, const char *target);
void etcd_channel_release(channel_ref_t *ref, int destroy);
void etcd_call_release(call_base_t *base);
void etcd_rotate_endpoint(ev_etcd_t *client);
void etcd_endpoint_failed(ev_etcd_t *client, const call_base_t *base,
                          grpc_status_code status, grpc_slice status_details);
void etcd_stream_failed(ev_etcd_t *client, unsigned channel_gen, int established,
                        grpc_status_code status, grpc_slice status_details);

void setup_auth_metadata(ev_etcd_t *client, grpc_op *op, call_base_t *base);
void setup_stream_metadata(ev_etcd_t *client, grpc_op *op, call_base_t *base);

extern grpc_slice METHOD_KV_RANGE;
extern grpc_slice METHOD_KV_PUT;
extern grpc_slice METHOD_KV_DELETE;
extern grpc_slice METHOD_KV_COMPACT;
extern grpc_slice METHOD_KV_TXN;
extern grpc_slice METHOD_WATCH;
extern grpc_slice METHOD_LEASE_GRANT;
extern grpc_slice METHOD_LEASE_REVOKE;
extern grpc_slice METHOD_LEASE_KEEPALIVE;
extern grpc_slice METHOD_LEASE_TTL;
extern grpc_slice METHOD_LEASE_LEASES;
extern grpc_slice METHOD_MAINTENANCE_STATUS;
extern grpc_slice METHOD_AUTH_AUTHENTICATE;
extern grpc_slice METHOD_AUTH_USER_ADD;
extern grpc_slice METHOD_AUTH_USER_DELETE;
extern grpc_slice METHOD_AUTH_USER_CHANGE_PASSWORD;
extern grpc_slice METHOD_AUTH_USER_GET;
extern grpc_slice METHOD_AUTH_USER_LIST;
extern grpc_slice METHOD_AUTH_USER_GRANT_ROLE;
extern grpc_slice METHOD_AUTH_USER_REVOKE_ROLE;
extern grpc_slice METHOD_AUTH_ENABLE;
extern grpc_slice METHOD_AUTH_DISABLE;
extern grpc_slice METHOD_AUTH_ROLE_ADD;
extern grpc_slice METHOD_AUTH_ROLE_DELETE;
extern grpc_slice METHOD_AUTH_ROLE_GET;
extern grpc_slice METHOD_AUTH_ROLE_LIST;
extern grpc_slice METHOD_AUTH_ROLE_GRANT_PERM;
extern grpc_slice METHOD_AUTH_ROLE_REVOKE_PERM;
extern grpc_slice METHOD_LOCK;
extern grpc_slice METHOD_UNLOCK;
extern grpc_slice METHOD_ELECTION_CAMPAIGN;
extern grpc_slice METHOD_ELECTION_PROCLAIM;
extern grpc_slice METHOD_ELECTION_LEADER;
extern grpc_slice METHOD_ELECTION_RESIGN;
extern grpc_slice METHOD_ELECTION_OBSERVE;
extern grpc_slice METHOD_CLUSTER_MEMBER_ADD;
extern grpc_slice METHOD_CLUSTER_MEMBER_REMOVE;
extern grpc_slice METHOD_CLUSTER_MEMBER_UPDATE;
extern grpc_slice METHOD_CLUSTER_MEMBER_LIST;
extern grpc_slice METHOD_CLUSTER_MEMBER_PROMOTE;
extern grpc_slice METHOD_MAINTENANCE_ALARM;
extern grpc_slice METHOD_MAINTENANCE_DEFRAGMENT;
extern grpc_slice METHOD_MAINTENANCE_HASH_KV;
extern grpc_slice METHOD_MAINTENANCE_MOVE_LEADER;
extern grpc_slice METHOD_AUTH_STATUS;

void init_method_slices(void);

#define SERIALIZE_PROTOBUF_TO_SLICE(slice_var, size_func, pack_func, req_ptr) \
    do { \
        size_t _req_len = size_func(req_ptr); \
        slice_var = grpc_slice_malloc(_req_len); \
        pack_func(req_ptr, GRPC_SLICE_START_PTR(slice_var)); \