view release on metacpan or  search on metacpan

ClickHouse.xs  view on Meta::CPAN

    SV *on_log;                     /* native SERVER_LOG packets */
    SV *on_failover;                /* multi-host: ($oh, $op, $nh, $np, $msg) */
    char *last_tls_error;           /* OpenSSL error from last failed handshake */
    char    **failover_hosts;       /* parallel arrays: hosts + ports.        */
    unsigned int *failover_ports;   /* NULL if multi-host failover disabled.  */
    int       failover_n;
    int       failover_idx;
    unsigned int failover_default_port;
    double query_start_time;        /* ev_now() captured in pipeline_advance */
    int tls_skip_verify;
    double query_timeout;
    size_t max_query_size;          /* 0 = unlimited; client-side croak guard */
    size_t max_recv_buffer;         /* 0 = unlimited; defensive recv ceiling */
    int http_basic_auth;            /* 0=X-ClickHouse-{User,Key} (default);
                                     * 1=Authorization: Basic ... (for proxies) */
    int auto_reconnect;
    uint32_t decode_flags;
    AV *native_col_names;   /* column names from last native result */
    AV *native_col_types;   /* column type strings from last native result */
    SV *on_drain;           /* callback fired when pending_count drops to 0 */
    char *last_query_id;    /* query_id of the last dispatched query */
    SV *on_trace;           /* debug trace callback */
    ev_timer ka_timer;      /* keepalive timer */
    double keepalive;       /* keepalive interval (0 = disabled) */
    int ka_timing;
    unsigned int ka_in_flight; /* keepalive pings sent but not yet ack'd */
    int callback_depth;
    /* error info from last SERVER_EXCEPTION or HTTP error */
    int32_t last_error_code;
    /* profile info from last SERVER_PROFILE_INFO */
    uint64_t profile_rows;
    uint64_t profile_bytes;
    uint64_t profile_rows_before_limit;
    /* totals / extremes from last native query */
    AV *native_totals;
    AV *native_extremes;
    /* reconnect backoff */
    double reconnect_delay;
    double reconnect_max_delay;
    double reconnect_jitter;        /* fractional [0, 1+]: actual delay
                                       picks uniformly in [d, d*(1+jitter)] */
    int reconnect_attempts;
    int reconnect_max_attempts;   /* 0 = unlimited */
    int pending_addendum_finish;  /* set when addendum partially written;
                                   * io_cb completes finish_connect after drain */
    ev_timer reconnect_timer;
    int reconnect_timing;
    /* on_progress throttling (0 = fire every packet) */
    double progress_period;
    double progress_last;          /* ev_now() of last on_progress dispatch */
    uint64_t progress_acc[5];      /* coalesced totals since last dispatch */
    /* LowCardinality cross-block dictionary state */
    SV ***lc_dicts;           /* array of dictionaries, one per column */
    uint64_t *lc_dict_sizes;  /* size of each dictionary */
    int lc_num_cols;          /* number of columns with LC state */
};

struct ev_ch_cb_s {
    SV          *cb;
    int          raw;        /* return raw response body instead of parsed rows */
    SV          *on_data;    /* per-query streaming callback (fires per block) */
    SV          *on_complete;/* per-query on_query_complete override (or NULL) */
    double       query_timeout;  /* per-query timeout (0=use default) */
    ngx_queue_t  queue;
};

struct ev_ch_send_s {
    char        *data;      /* full HTTP request or native packet */
    size_t       data_len;
    SV          *cb;
    char        *insert_data;     /* deferred TSV data for native INSERT */
    size_t       insert_data_len;
    SV          *insert_av;      /* deferred AV* data for native INSERT */
    int          raw;            /* return raw response body */
    SV          *on_data;        /* per-query streaming callback */
    SV          *on_complete;    /* per-query on_query_complete override */
    double       query_timeout;  /* per-query timeout */
    char        *query_id;       /* query_id for tracking */
    ngx_queue_t  queue;
};

/* Forward declarations for helpers defined further down (or in xs/io.c)
 * but called from earlier code in this file or from xs/*.c included
 * before the definition site. */
static void timer_cb(EV_P_ ev_timer *w, int revents);
static void stop_keepalive(ev_clickhouse_t *self);
static void schedule_reconnect(ev_clickhouse_t *self);
static void lc_free_dicts(ev_clickhouse_t *self);
static void start_reading(ev_clickhouse_t *self);
static void stop_reading(ev_clickhouse_t *self);
static void start_writing(ev_clickhouse_t *self);
static void stop_writing(ev_clickhouse_t *self);
static void emit_error(ev_clickhouse_t *self, const char *msg);
static void emit_trace(ev_clickhouse_t *self, const char *fmt, ...);
static int cleanup_connection(ev_clickhouse_t *self);
static int  cancel_pending(ev_clickhouse_t *self, const char *errmsg);
static int  check_destroyed(ev_clickhouse_t *self);
static char *safe_strdup(const char *s);
static void failover_free(ev_clickhouse_t *self);
static int  finish_connect(ev_clickhouse_t *self);
static int  try_write(ev_clickhouse_t *self);
static int  pipeline_advance(ev_clickhouse_t *self);

/* Two helpers shared between the PL_dirty and normal arms of DESTROY:
 * adding a new on_* slot or persistent string requires touching only
 * one place. on_failover is omitted from CONNECTION_HANDLERS because
 * failover_free() handles it together with the host ring. */
#define CLEAR_CONNECTION_HANDLERS(self) do { \
    CLEAR_SV((self)->on_connect); \
    CLEAR_SV((self)->on_error); \
    CLEAR_SV((self)->on_progress); \
    CLEAR_SV((self)->on_disconnect); \
    CLEAR_SV((self)->on_query_complete); \
    CLEAR_SV((self)->on_query_start); \
    CLEAR_SV((self)->on_log); \
    CLEAR_SV((self)->on_drain); \
    CLEAR_SV((self)->on_trace); \
} while (0)

#define CLEAR_PERSISTENT_STATE(self) do { \
    CLEAR_STR((self)->last_tls_error); \
    CLEAR_STR((self)->last_query_id); \
    CLEAR_STR((self)->host); \
    CLEAR_STR((self)->user); \
    CLEAR_STR((self)->password); \
    CLEAR_STR((self)->database); \
    CLEAR_STR((self)->session_id); \
    CLEAR_STR((self)->query_log_comment); \
    CLEAR_STR((self)->tls_ca_file); \
    CLEAR_STR((self)->tls_cert_file); \
    CLEAR_STR((self)->tls_key_file); \
    CLEAR_STR((self)->server_name); \
    CLEAR_STR((self)->server_display_name); \
    CLEAR_STR((self)->server_timezone); \
    CLEAR_STR((self)->insert_err); \