main.c

/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <pthread.h>
#include <stdio.h>
#include <string.h>
#include <strings.h>
#include <errno.h>
#include <stdlib.h>
#include <stdint.h>
#include <stddef.h>
#include <stdbool.h>
#include <time.h>
#include <unistd.h>
#include <regex.h>

#include <hbase/hbase.h>

/* Found under /libhbase/src/test/native/common */
#include <byte_buffer.h>
#include "hedis.h"

#line __LINE__ "main.c"

/*
 * Sample code to illustrate usage of libhbase APIs
 */

#ifdef __cplusplus
extern  "C" {
#endif

#define CHECK_API_ERROR(retCode, ...) \
    HBASE_LOG_MSG((retCode ? HBASE_LOG_LEVEL_ERROR : HBASE_LOG_LEVEL_INFO), \
        __VA_ARGS__, retCode);
// test REGEX: https://regex101.com/r/bB3mQ1/4
#define HEDIS_COMMAND_PATTERN "([a-zA-Z0-9_\\-]+)@([#:a-zA-Z0-9_\\\\\\-]+)(@([@#a-zA-Z0-9_\\\\\\-]+)(:([a-zA-Z0-9_\\-]+))?)?"
#define HEDIS_COMMAND_LENGTH 6
#define HEDIS_COMMAND_TABLE_INDEX 0
#define HEDIS_COMMAND_ROWKEY_INDEX 1
#define HEDIS_COMMAND_COLUMN_FAMILY_INDEX 3
#define HEDIS_COMMAND_COLUMN_QUALIFIER_INDEX 5
#define MAX_ERROR_MSG 0x1000

typedef struct cell_data_t_ {
    bytebuffer value;
    hb_cell_t  *hb_cell;
    struct cell_data_t_ *next_cell;
} cell_data_t;

cell_data_t*
new_cell_data() {
    cell_data_t *cell_data = (cell_data_t*) calloc(1, sizeof(cell_data_t));
    cell_data->next_cell = NULL;
    return cell_data;
}

typedef struct row_data_t_ {
    bytebuffer key;
    struct cell_data_t_ *first_cell;
} row_data_t;

void
release_row_data(row_data_t *row_data) {
    if (row_data != NULL) {
        cell_data_t *cell = row_data->first_cell;
        while (cell) {
            bytebuffer_free(cell->value);
            free(cell->hb_cell);
            cell_data_t *cur_cell = cell;
            cell = cell->next_cell;
            free(cur_cell);
        }
        bytebuffer_free(row_data->key);
        free(row_data);
    }
}

hedisConfigEntry **hedis_entries;
int hedis_entry_count;
regex_t *r = NULL;
char *value;
char *char_rowkey = NULL;
char *char_family = NULL;
char *char_qualifier = NULL;
bytebuffer byte_rowkey = NULL;
bytebuffer byte_family = NULL;
bytebuffer byte_qualifier = NULL;
char *connector_zookeeper = NULL;
hb_connection_t connection = NULL;
hb_client_t client = NULL;

char *convert(const byte_t *src, size_t length) {
    char *result = malloc(sizeof(char) * length);

    sprintf(result, "%.*s", length, src);

    return result;
}

void printRow(const hb_result_t result) {
    const byte_t *key = NULL;
    size_t key_len = 0;
    hb_result_get_key(result, &key, &key_len);
    size_t cell_count = 0;
    hb_result_get_cell_count(result, &cell_count);
    HBASE_LOG_INFO("Row=\'%.*s\', cell count=%d", key_len, key, cell_count);
    const hb_cell_t **cells;
    hb_result_get_cells(result, &cells, &cell_count);
    for (size_t i = 0; i < cell_count; ++i) {
        HBASE_LOG_INFO(
            "Cell %d: family=\'%.*s\', qualifier=\'%.*s\', "
            "value=\'%.*s\', timestamp=%lld.", i,
            cells[i]->family_len, cells[i]->family,
            cells[i]->qualifier_len, cells[i]->qualifier,
            cells[i]->value_len, cells[i]->value, cells[i]->ts);
    }
}

char * to_json(const hb_result_t result) {
    const byte_t *key = NULL;
    size_t key_len = 0;

    char *json;

    json = malloc(sizeof(char) * 200);

    hb_result_get_key(result, &key, &key_len);

    if (key == NULL) {
        return NULL;
    }

    size_t cell_count = 0;

    hb_result_get_cell_count(result, &cell_count);

    const hb_cell_t **cells;

    hb_result_get_cells(result, &cells, &cell_count);

    // show JSON: {"rowkey":"thisisrowkey","columns":[{"name":"cf:gu","value":"M0000001"},{"name":"cf:nk","value":"kewang"}]}
    strcpy(json, "{\"rowkey\":\"");

    strcat(json, convert(key, key_len));
    strcat(json, "\",\"columns\":[");

    for (size_t i = 0; i < cell_count; i++) {
        strcat(json, "{\"name\":\"");
        strcat(json, convert(cells[i]->family, cells[i]->family_len));
        strcat(json, ":");
        strcat(json, convert(cells[i]->qualifier, cells[i]->qualifier_len));
        strcat(json, "\",\"value\":\"");
        strcat(json, convert(cells[i]->value, cells[i]->value_len));
        strcat(json, "\"},");
    }

    int json_length = strlen(json);

    memmove(&json[json_length - 1], &json[json_length], 1);

    strcat(json, "]}");

    return json;
}

/**
 * Get synchronizer and callback
 */
volatile bool get_done = false;
pthread_cond_t get_cv = PTHREAD_COND_INITIALIZER;
pthread_mutex_t get_mutex = PTHREAD_MUTEX_INITIALIZER;

void
get_callback(int32_t err, hb_client_t client,
             hb_get_t get, hb_result_t result, void *extra) {
    if (err == 0) {
        const char *table_name;
        size_t table_name_len;
        hb_result_get_table(result, &table_name, &table_name_len);
        HBASE_LOG_INFO("Received get callback for table=\'%.*s\'.",
                       table_name_len, table_name);

        printRow(result);

        value = to_json(result);

        printf("value: %s\n", value);

        hb_result_destroy(result);
    } else {
        HBASE_LOG_ERROR("Get failed with error code: %d.", err);
    }

    hb_get_destroy(get);

    pthread_mutex_lock(&get_mutex);
    get_done = true;
    pthread_cond_signal(&get_cv);
    pthread_mutex_unlock(&get_mutex);
}

void
wait_for_get() {
    HBASE_LOG_INFO("Waiting for get operation to complete.");
    pthread_mutex_lock(&get_mutex);
    while (!get_done) {
        pthread_cond_wait(&get_cv, &get_mutex);
    }
    pthread_mutex_unlock(&get_mutex);
    HBASE_LOG_INFO("Get operation completed.");
}

/**
 * Scan synchronizer and callbacks
 */
volatile bool scan_done = false;
pthread_cond_t scan_cv = PTHREAD_COND_INITIALIZER;
pthread_mutex_t scan_mutex = PTHREAD_MUTEX_INITIALIZER;

void scan_callback(int32_t err, hb_scanner_t scanner,
                   hb_result_t results[], size_t num_results, void *extra) {
    if (num_results) {
        const char *table_name;
        size_t table_name_len;
        hb_result_get_table(results[0], &table_name, &table_name_len);
        HBASE_LOG_INFO("Received scan_next callback for table=\'%.*s\', row count=%d.",
                       table_name_len, table_name, num_results);

        for (int i = 0; i < num_results; ++i) {
            printRow(results[i]);
            hb_result_destroy(results[i]);
        }
        hb_scanner_next(scanner, scan_callback, NULL);
    } else {
        hb_scanner_destroy(scanner, NULL, NULL);
        pthread_mutex_lock(&scan_mutex);
        scan_done = true;
        pthread_cond_signal(&scan_cv);
        pthread_mutex_unlock(&scan_mutex);
    }
}

void
wait_for_scan() {
    HBASE_LOG_INFO("Waiting for scan to complete.");
    pthread_mutex_lock(&scan_mutex);
    while (!scan_done) {
        pthread_cond_wait(&scan_cv, &scan_mutex);
    }
    pthread_mutex_unlock(&scan_mutex);
    HBASE_LOG_INFO("Scan completed.");
}

/**
 * Client destroy synchronizer and callbacks
 */
volatile bool client_destroyed = false;
pthread_cond_t client_destroyed_cv = PTHREAD_COND_INITIALIZER;
pthread_mutex_t client_destroyed_mutex = PTHREAD_MUTEX_INITIALIZER;

void
client_disconnection_callback(int32_t err,
                              hb_client_t client, void *extra) {
    HBASE_LOG_INFO("Received client disconnection callback.");
    pthread_mutex_lock(&client_destroyed_mutex);
    client_destroyed = true;
    pthread_cond_signal(&client_destroyed_cv);
    pthread_mutex_unlock(&client_destroyed_mutex);
}

void
wait_client_disconnection() {
    HBASE_LOG_INFO("Waiting for client to disconnect.");
    pthread_mutex_lock(&client_destroyed_mutex);
    while (!client_destroyed) {
        pthread_cond_wait(&client_destroyed_cv, &client_destroyed_mutex);
    }
    pthread_mutex_unlock(&client_destroyed_mutex);
    HBASE_LOG_INFO("Client disconnected.");
}

int
ensureTable(hb_connection_t connection, const char *table_name) {
    int32_t retCode = 0;
    hb_admin_t admin = NULL;

    if ((retCode = hb_admin_create(connection, &admin)) != 0) {
        HBASE_LOG_ERROR("Could not create HBase admin : errorCode = %d.", retCode);

        goto cleanup;
    }

    if ((retCode = hb_admin_table_exists(admin, NULL, table_name)) == 0) {
        HBASE_LOG_INFO("Table '%s' exists", table_name);
    } else if (retCode != ENOENT) {
        HBASE_LOG_ERROR("Error while checking if the table exists: errorCode = %d.", retCode);
    }

cleanup:
    if (admin) {
        hb_admin_destroy(admin, NULL, NULL);
    }

    return retCode;
}

int init(hedisConfigEntry **entries, int entry_count) {
    hedis_entries = entries;
    hedis_entry_count = entry_count;

    r = malloc(sizeof(regex_t));

    int status = regcomp(r, HEDIS_COMMAND_PATTERN, REG_EXTENDED | REG_NEWLINE);

    if (status != 0) {
        char error_message[MAX_ERROR_MSG];

        regerror(status, r, error_message, MAX_ERROR_MSG);

        printf("Regex error compiling '%s': %s\n", HEDIS_COMMAND_PATTERN, error_message);

        return -1;
    }

    for (int i = 0; i < hedis_entry_count; i++) {
        if (!strcasecmp(hedis_entries[i]->key, "zookeeper")) {
            connector_zookeeper = malloc(sizeof(char) * (strlen(hedis_entries[i]->value) + 1));

            strcpy(connector_zookeeper, hedis_entries[i]->value);

            break;
        }
    }

    int32_t retCode = 0;

    if ((retCode = hb_connection_create(connector_zookeeper,
                                        NULL,
                                        &connection)) != 0) {
        HBASE_LOG_ERROR("Could not create HBase connection : errorCode = %d.", retCode);

        return -1;
    }

    HBASE_LOG_INFO("Connecting to HBase cluster using Zookeeper ensemble '%s'.",
                   connector_zookeeper);

    if ((retCode = hb_client_create(connection, &client)) != 0) {
        HBASE_LOG_ERROR("Could not connect to HBase cluster : errorCode = %d.", retCode);

        return -1;
    }

    return 0;
}

int parse_hedis_command(const char * to_match, char ** str) {
    /* "P" is a pointer into the string which points to the end of the
     *        previous match. */
    const char * p = to_match;
    /* "N_matches" is the maximum number of matches allowed. */
    const int n_matches = HEDIS_COMMAND_LENGTH + 1;
    /* "M" contains the matches found. */
    regmatch_t m[n_matches];

    int i = 0;
    int nomatch = regexec(r, p, n_matches, m, 0);

    if (nomatch) {
        printf("No more matches.\n");

        return -1;
    }

    for (i = 0; i < n_matches; i++) {
        int start;
        int finish;

        if (m[i].rm_so == -1) {
            break;
        }

        start = m[i].rm_so + (p - to_match);
        finish = m[i].rm_eo + (p - to_match);

        if (i != 0) {
            int size = finish - start;

            str[i - 1] = malloc(sizeof(char) * (size + 1));

            sprintf(str[i - 1], "%.*s", size, to_match + start);
        }
    }

    p += m[0].rm_eo;

    return i - 1;
}

char *get_value(const char *str) {
    char **commands = malloc(sizeof(char *) * HEDIS_COMMAND_LENGTH);

    int command_length = parse_hedis_command(str, commands);

    if (command_length == -1) {
        return NULL;
    }

    int32_t retCode = 0;
    FILE* logFile = NULL;

    // reinitialize pthread
    get_cv = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
    get_mutex = (pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;
    client_destroyed_cv = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
    client_destroyed_mutex = (pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;

    char_rowkey = commands[HEDIS_COMMAND_ROWKEY_INDEX];

    if (command_length >= HEDIS_COMMAND_COLUMN_FAMILY_INDEX + 1) {
        char_family = commands[HEDIS_COMMAND_COLUMN_FAMILY_INDEX];
        byte_family = bytebuffer_strcpy(char_family);
    }

    if (command_length >= HEDIS_COMMAND_COLUMN_QUALIFIER_INDEX + 1) {
        char_qualifier = commands[HEDIS_COMMAND_COLUMN_QUALIFIER_INDEX];
        byte_qualifier = bytebuffer_strcpy(char_qualifier);
    }

    byte_rowkey = bytebuffer_strcpy(char_rowkey);

    const char *table_name = commands[HEDIS_COMMAND_TABLE_INDEX];
    const size_t table_name_len = strlen(table_name);

    hb_log_set_level(HBASE_LOG_LEVEL_DEBUG); // defaults to INFO
    const char *logFilePath = getenv("HBASE_LOG_FILE");
    if (logFilePath != NULL) {
        FILE* logFile = fopen(logFilePath, "a");
        if (!logFile) {
            retCode = errno;
            fprintf(stderr, "Unable to open log file \"%s\"", logFilePath);
            perror(NULL);

            goto cleanup;
        }
        hb_log_set_stream(logFile); // defaults to stderr
    }

    // fetch a row with rowkey
    hb_get_t get = NULL;
    hb_get_create(byte_rowkey->buffer, byte_rowkey->length, &get);

    if (char_family != NULL) {
        if (char_qualifier == NULL) {
            hb_get_add_column(get, (byte_t *)char_family, strlen(char_family), NULL, 0);
        } else {
            hb_get_add_column(get, (byte_t *)char_family, strlen(char_family), (byte_t *)char_qualifier, strlen(char_qualifier));
        }
    }

    hb_get_set_table(get, table_name, table_name_len);
    hb_get_set_num_versions(get, 10); // up to ten versions of each column

    get_done = false;
    hb_get_send(client, get, get_callback, NULL);
    wait_for_get();

cleanup:
    if (logFile) {
        HBASE_LOG_INFO("Close log file.");
        fclose(logFile);
    }

    if (char_qualifier != NULL) {
        bytebuffer_free(byte_qualifier);
    }

    if (char_family != NULL) {
        bytebuffer_free(byte_family);
    }

    for (int i = 0; i < command_length; i++) {
        free(commands[i]);
    }

    free(commands);

    bytebuffer_free(byte_rowkey);

    pthread_cond_destroy(&get_cv);
    pthread_mutex_destroy(&get_mutex);

    pthread_cond_destroy(&client_destroyed_cv);
    pthread_mutex_destroy(&client_destroyed_mutex);

    HBASE_LOG_INFO("Return value: %s\n", value);

    return value;
}

#ifdef __cplusplus
}
#endif