nicdex/eventstore-client-c
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Moved main to test

Browse Source 
Moved esc.h to include
Added append_to_stream on connection
This commit is contained in:
Nicolas Dextraze
2018-03-24 09:30:29 -07:00
parent 4eef29c0da
commit aa0db4f07a
42 changed files with 1461 additions and 699 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@@ -10,7 +10,7 @@ if(WIN32)
link_directories(c:/Users/nicol/dev/thirdparty/protobuf-c/protobuf-c/.libs)
endif()
add_executable(esc src/main.c src/utils/mutex.c src/utils/mutex.h src/utils/thread.c src/utils/thread.h src/utils/socket.c src/utils/socket.h src/esc.c src/esc.h src/proto.c src/proto.h src/utils/uuid.c src/utils/uuid.h src/utils/buffer.c src/utils/buffer.h src/tcp_package.c src/tcp_package.h src/utils/string.c src/utils/string.h src/utils/queue.c src/utils/queue.h src/tcp_messages.c src/tcp_messages.h src/utils/debug.h src/utils/debug.c)
add_executable(esc test/main.c src/utils/mutex.c src/utils/mutex.h src/utils/thread.c src/utils/thread.h src/utils/socket.c src/utils/socket.h include/esc.h src/proto.c src/proto.h src/utils/uuid.c src/utils/uuid.h src/utils/buffer.c src/utils/buffer.h src/tcp_package.c src/tcp_package.h src/utils/string.c src/utils/string.h src/utils/queue.c src/utils/queue.h src/tcp_messages.c src/tcp_messages.h src/utils/debug.h src/utils/debug.c src/position.c src/position.h src/credentials.c src/credentials.h src/connection.c src/connection.h src/utils/bool.h src/endpoint_discoverer.h src/utils/error.c src/utils/error.h src/endpoint_discoverer.c src/proto_helper.h src/proto_helper.c src/results.h src/results.c src/event_data.c src/event_data.h src/utils/array.c src/utils/array.h src/utils/event.c src/utils/event.h)
if(WIN32)
target_link_libraries(esc wsock32 ws2_32 libprotobuf-c.a)

112
include/esc.h Normal file
View File

@@ -0,0 +1,112 @@
//
// Created by nicol on 2018-03-18.
//
#ifndef ESC_ESC_H
#define ESC_ESC_H
#include <stdint.h>
// bool
typedef int bool_t;
#define BOOL_TRUE 1
#define BOOL_FALSE 0
// array
typedef struct st_array array_t;
typedef void (*array_deallocator)(void*);
array_t* array_create(size_t n, ...);
void array_destroy(array_t* array, array_deallocator destroyer);
// uuid
typedef struct st_uuid uuid_t;
uuid_t* uuid_create();
void uuid_destroy(uuid_t*);
// buffer
typedef struct st_buffer buffer_t;
void buffer_destroy(buffer_t*);
buffer_t* buffer_from_string(const char* str);
size_t buffer_size(buffer_t* b);
// error
#define ERROR_MSG_SIZE 1024
typedef struct st_error {
const char* file;
int line;
int code;
char message[ERROR_MSG_SIZE];
} error_t;
// esc
#define ESC_VERSION_NOSTREAM -1
#define ESC_VERSION_EMPTYSTREAM -1
#define ESC_VERSION_ANY -2
typedef struct st_esc_position esc_position_t;
typedef struct st_esc_connection_settings esc_connection_settings_t;
typedef struct st_connection esc_connection_t;
typedef struct st_credentials esc_credentials_t;
struct st_recorded_event {
uuid_t* event_id;
const char* event_type;
int64_t event_number;
const char *event_stream_id;
int64_t created_epoch;
buffer_t* data;
buffer_t* metadata;
};
typedef struct st_recorded_event esc_recorded_event_t;
struct st_resolved_event {
esc_recorded_event_t* event;
esc_recorded_event_t* link;
esc_position_t* original_position;
};
typedef struct st_resolved_event esc_resolved_event_t;
struct st_all_events_slice {
char* read_direction;
esc_position_t* from_position;
esc_position_t* next_position;
size_t n_events;
esc_resolved_event_t** events;
int is_end_of_stream;
};
typedef struct st_all_events_slice esc_all_events_slice_t;
typedef struct st_esc_write_result {
int64_t next_expected_version;
esc_position_t* log_position;
} esc_write_result_t;
typedef struct st_esc_event_data esc_event_data_t;
esc_connection_settings_t* const esc_default_connection_settings;
// Connection
esc_connection_t* esc_connection_create(esc_connection_settings_t* connection_settings, const char* addr, const char* connection_name);
void esc_connection_destroy(esc_connection_t* conn);
int esc_connection_connect(esc_connection_t* conn);
void esc_connection_close(esc_connection_t* conn);
esc_credentials_t* esc_credentials_create(const char* username, const char* password);
void esc_credentials_destroy(esc_credentials_t* creds);
esc_event_data_t* esc_event_data_create(uuid_t* event_id, const char* event_type, bool_t is_json, buffer_t* data, buffer_t* metadata);
void esc_event_data_destroy(esc_event_data_t*);
esc_write_result_t* esc_append_to_stream(esc_connection_t* conn, const char* stream, int64_t expected_version, array_t* events);
void esc_write_result_destroy(esc_write_result_t* write_result);
esc_all_events_slice_t* esc_connection_read_all_forward(esc_connection_t* conn, esc_position_t* last_checkpoint, unsigned int count, esc_credentials_t* credentials);
void esc_all_events_slice_destroy(esc_all_events_slice_t* all_events_slice);
error_t* esc_connection_last_error(esc_connection_t* conn);
// Formatting
const char* uuid_format(uuid_t* uuid, char* buf, size_t buf_size);
const char* esc_position_format(esc_position_t* position, char* buffer, size_t buf_size);
#endif //ESC_ESC_H

322
src/connection.c Normal file
View File

@@ -0,0 +1,322 @@
//
// Created by nicolas on 22/03/18.
//
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "connection.h"
#include "tcp_package.h"
#include "tcp_messages.h"
#include "credentials.h"
#include "proto_helper.h"
#include "utils/array.h"
#include "utils/debug.h"
#include "utils/error.h"
#include "utils/string.h"
#include "utils/socket.h"
// usleep
#ifdef _WIN32
#include <windows.h>
#define usleep Sleep
#endif
#ifdef __linux__
#include <unistd.h>
#endif
struct st_esc_connection_settings default_connection_settings = {
BOOL_FALSE
};
esc_connection_settings_t* esc_default_connection_settings = &default_connection_settings;
//TODO partial transfer
ssize_t connection_send_tcp_package(esc_connection_t* conn, tcp_package_t* pkg) {
char uuid_buf[37];
#ifdef _WIN32
fprintf(stderr, "connection_send_tcp_package: %s %u %s %llu ", get_string_for_tcp_message(pkg->command), pkg->flags, esc_uuid_format(pkg->correlation_id, uuid_buf, 37), buffer_size(pkg->data));
#else
fprintf(stderr, "connection_send_tcp_package: %s %u %s %lu ", get_string_for_tcp_message(pkg->command), pkg->flags,
uuid_format(pkg->correlation_id, uuid_buf, 37), buffer_size(pkg->data));
#endif
buffer_t* send_buffer = tcp_package_to_buffer(pkg);
size_t send_buffer_size = buffer_size(send_buffer);
uint32_t size = (uint32_t)send_buffer_size;
ssize_t rc;
if ((rc = socket_send(conn->tcp_conn, (char *) &size, sizeof(uint32_t))) <= 0) {
buffer_destroy(send_buffer);
fprintf(stderr, "%d\n", socket_error(conn->tcp_conn));
return -1;
}
uint8_t* send_buffer_data = buffer_data(send_buffer);
if ((rc = socket_send(conn->tcp_conn, (char *) send_buffer_data, send_buffer_size)) <= 0) {
buffer_destroy(send_buffer);
fprintf(stderr, "%d\n", socket_error(conn->tcp_conn));
return -2;
}
buffer_destroy(send_buffer);
fprintf(stderr, "0\n");
return 0;
}
//TODO partial transfer
tcp_package_t* connection_recv_tcp_package(esc_connection_t* conn) {
uint32_t recv_size;
ssize_t rc;
if ((rc = socket_recv(conn->tcp_conn, (char *)&recv_size, sizeof(uint32_t))) != 4) {
#ifdef _WIN32
fprintf(stderr, "connection_recv_tcp_package: %lld %d\n", rc, socket_error(conn->tcp_conn));
#else
fprintf(stderr, "connection_recv_tcp_package: %ld %d\n", rc, socket_error(conn->tcp_conn));
#endif
return 0;
}
buffer_t* recv_buffer = buffer_create(recv_size);
size_t recv_buffer_size = recv_size;
uint8_t* recv_buffer_data = buffer_data(recv_buffer);
while(recv_size > 0) {
size_t pos = recv_buffer_size - recv_size;
rc = socket_recv(conn->tcp_conn, (char *)&recv_buffer_data[pos], recv_size);
recv_size -= rc;
}
tcp_package_t* recv_pkg = tcp_package_from_buffer(recv_buffer);
buffer_destroy(recv_buffer);
char uuid_buf[37];
#ifdef _WIN32
fprintf(stderr, "connection_recv_tcp_package: %s %u %s %llu\n", get_string_for_tcp_message(recv_pkg->command), recv_pkg->flags, esc_uuid_format(recv_pkg->correlation_id, uuid_buf, 37), buffer_size(recv_pkg->data));
#else
fprintf(stderr, "connection_recv_tcp_package: %s %u %s %lu\n", get_string_for_tcp_message(recv_pkg->command), recv_pkg->flags,
uuid_format(recv_pkg->correlation_id, uuid_buf, 37), buffer_size(recv_pkg->data));
#endif
return recv_pkg;
}
void* connection_thread(void* arg) {
esc_connection_t* conn = arg;
while(conn->stop == BOOL_FALSE) {
if (socket_writable(conn->tcp_conn)) {
tcp_package_t* send_pkg = queue_dequeue(conn->send_queue);
if (send_pkg != 0) {
ssize_t rc = connection_send_tcp_package(conn, send_pkg);
tcp_package_destroy(send_pkg);
if (rc != 0) {
fprintf(stderr, "failed to send pkg.");
}
}
}
if (socket_readable(conn->tcp_conn)) {
tcp_package_t* recv_pkg = connection_recv_tcp_package(conn);
if (recv_pkg == 0) {
//TODO Handle connection lost
conn->stop = 1;
} else if (recv_pkg->command == MESSAGE_HEARTBEATREQUEST) {
tcp_package_t* heartbeat_pkg = tcp_package_create(MESSAGE_HEARTBEATRESPONSE, recv_pkg->correlation_id, buffer_create(0));
queue_enqueue(conn->send_queue, heartbeat_pkg);
tcp_package_destroy(recv_pkg);
} else {
queue_enqueue(conn->recv_queue, recv_pkg);
}
}
usleep(1);
}
printf("manager thread stopped\n");
event_set(conn->stopped_event);
}
bool_t find_tcp_package_by_correlation_id(void* _item, void* _correlation_id) {
tcp_package_t* item = _item;
uuid_t* correlation_id = _correlation_id;
if (uuid_compare(item->correlation_id, correlation_id) == 0) {
return BOOL_TRUE;
}
return BOOL_FALSE;
}
//TODO add timeout
tcp_package_t* connection_wait_for(esc_connection_t* conn, uuid_t* correlation_id) {
tcp_package_t* found = 0;
while(found == 0) {
found = queue_remove(conn->recv_queue, find_tcp_package_by_correlation_id, correlation_id);
usleep(1);
}
return found;
}
esc_connection_t* esc_connection_create(esc_connection_settings_t* connection_settings, const char* addr, const char* connection_name) {
struct st_connection* conn = malloc(sizeof(struct st_connection));
conn->settings = connection_settings;
if (connection_name == 0 || strcmp(connection_name, "") == 0) {
uuid_t* uuid = uuid_create();
char buf[40];
uuid_format(uuid, buf, 40);
conn->name = string_copy(buf);
uuid_destroy(uuid);
} else {
conn->name = connection_name;
}
conn->send_queue = queue_create();
conn->recv_queue = queue_create();
conn->stop = 0;
conn->stopped_event = event_create();
conn->last_error = 0;
if (strncmp(addr, "tcp://", 6) != 0) {
conn->last_error = error_create(1, "invalid schema for address: %s", addr);
return conn;
}
char* pos = strrchr(addr, ':');
if (pos == 0) {
conn->last_error = error_create(2, "missing port in address: %s", addr);
return conn;
}
char* host = malloc(pos - addr - 5);
strncpy(host, addr+6, pos-addr-6);
host[pos-addr-6] = 0;
unsigned short port = (unsigned short)atoi(pos+1);
esc_static_endpoint_discoverer_t discover_data = {
{host, port}, connection_settings->use_ssl_connection
};
conn->discoverer_data = malloc(sizeof(struct st_esc_static_endpoint_discoverer));
memcpy(conn->discoverer_data, &discover_data, sizeof(struct st_esc_static_endpoint_discoverer));
conn->discover = (esc_endpoint_discoverer_t)esc_static_discover;
return conn;
}
void esc_connection_destroy(esc_connection_t* conn) {
esc_connection_close(conn);
socket_destroy(conn->tcp_conn);
free(((struct st_esc_static_endpoint_discoverer*)conn->discoverer_data)->tcp_endpoint.host);
free(conn->discoverer_data);
thread_destroy(conn->manager_thread);
queue_destroy(conn->recv_queue);
queue_destroy(conn->send_queue);
event_destroy(conn->stopped_event);
free((void*)conn->name);
free(conn);
}
// return 0 on success
// return non-zero on failure and sets last_error on connection
int esc_connection_connect(esc_connection_t* conn) {
// Discover endpoint
esc_node_endpoints_t* endpoints = conn->discover(conn->discoverer_data, 0);
if (endpoints == 0) {
return -1;
}
// Establish Tcp Connection
conn->tcp_conn = socket_create(SOCKET_TYPE_TCP);
ip_endpoint_t endpoint = conn->settings->use_ssl_connection ? endpoints->secure_tcp_endpoint : endpoints->tcp_endpoint;
if (socket_connect(conn->tcp_conn, endpoint.host, endpoint.port) != 0) {
conn->last_error = error_create(socket_error(conn->tcp_conn), "can't connect to %s:%d", endpoint.host, endpoint.port);
return -1;
}
free(endpoints);
// Start manager thread
conn->manager_thread = thread_create(connection_thread, conn);
if (thread_start(conn->manager_thread)) {
conn->last_error = error_create(thread_error(conn->manager_thread), "can't start manager thread.", "");
return -1;
}
// Identify
uuid_t* correlation_id = uuid_create();
buffer_t* msg_buf = esc_identify_client_pack(conn->name);
tcp_package_t* send_pkg = tcp_package_create(MESSAGE_IDENTIFYCLIENT, correlation_id, msg_buf);
queue_enqueue(conn->send_queue, send_pkg);
tcp_package_t* recv_pkg = connection_wait_for(conn, correlation_id);
if (recv_pkg->command != MESSAGE_CLIENTIDENTIFIED) {
conn->last_error = error_create(0, "server error: %d.", recv_pkg->command);
uuid_destroy(correlation_id);
tcp_package_destroy(recv_pkg);
return -1;
}
uuid_destroy(correlation_id);
tcp_package_destroy(recv_pkg);
return 0;
}
esc_all_events_slice_t* esc_connection_read_all_forward(esc_connection_t* conn, esc_position_t* last_checkpoint, unsigned int count, esc_credentials_t* credentials) {
uuid_t* correlation_id = uuid_create();
buffer_t* msg_buf = esc_read_all_forward_pack(last_checkpoint, count, 0);
tcp_package_t* send_pkg = tcp_package_create_authenticated(MESSAGE_READALLEVENTSFORWARD, correlation_id, msg_buf, credentials->username, credentials->password);
queue_enqueue(conn->send_queue, send_pkg);
tcp_package_t* recv_pkg = connection_wait_for(conn, correlation_id);
if (recv_pkg->command != MESSAGE_READALLEVENTSFORWARDCOMPLETED) {
conn->last_error = error_create(recv_pkg->command, "wrong message from the server: %s", get_string_for_tcp_message(recv_pkg->command));
tcp_package_destroy(recv_pkg);
uuid_destroy(correlation_id);
return 0;
}
esc_all_events_slice_t* result;
inspection_result_t* inspection_result = esc_all_events_slice_unpack(recv_pkg->data, &result);
// TODO fully handle inspection results
if (inspection_result->error) {
conn->last_error = error_copy(inspection_result->error);
inspection_result_destroy(inspection_result);
tcp_package_destroy(recv_pkg);
uuid_destroy(correlation_id);
return 0;
}
inspection_result_destroy(inspection_result);
tcp_package_destroy(recv_pkg);
uuid_destroy(correlation_id);
return result;
}
esc_write_result_t* esc_append_to_stream(esc_connection_t* conn, const char* stream, int64_t expected_version, array_t* events) {
uuid_t* correlation_id = uuid_create();
buffer_t* msg_buf = esc_append_to_stream_pack(stream, expected_version, events);
tcp_package_t* send_pkg = tcp_package_create(MESSAGE_WRITEEVENTS, correlation_id, msg_buf);
queue_enqueue(conn->send_queue, send_pkg);
tcp_package_t* recv_pkg = connection_wait_for(conn, correlation_id);
if (recv_pkg->command != MESSAGE_WRITEEVENTSCOMPLETED) {
conn->last_error = error_create(recv_pkg->command, "server error: %s", get_string_for_tcp_message(recv_pkg->command));
tcp_package_destroy(recv_pkg);
uuid_destroy(correlation_id);
return 0;
}
esc_write_result_t* result;
inspection_result_t* inspection_result = esc_write_result_unpack(recv_pkg->data, &result);
// TODO fully handle inspection results
if (inspection_result->error) {
conn->last_error = error_copy(inspection_result->error);
inspection_result_destroy(inspection_result);
tcp_package_destroy(recv_pkg);
uuid_destroy(correlation_id);
return 0;
}
inspection_result_destroy(inspection_result);
tcp_package_destroy(recv_pkg);
uuid_destroy(correlation_id);
return result;
}
void esc_connection_close(esc_connection_t* conn) {
if (conn->stop == 1) return;
conn->stop = 1;
event_wait(conn->stopped_event);
socket_close(conn->tcp_conn);
}
inline error_t* esc_connection_last_error(esc_connection_t* conn) {
return conn->last_error;
}

41
src/connection.h Normal file
View File

@@ -0,0 +1,41 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_ESC_CONNECTION_H
#define ESC_ESC_CONNECTION_H
#include <protobuf-c/protobuf-c.h>
#include "utils/bool.h"
#include "utils/socket.h"
#include "utils/thread.h"
#include "utils/mutex.h"
#include "utils/queue.h"
#include "utils/error.h"
#include "endpoint_discoverer.h"
#include "utils/event.h"
typedef struct st_esc_connection_settings {
bool_t use_ssl_connection;
} esc_connection_settings_t;
esc_connection_settings_t* esc_default_connection_settings;
typedef struct st_connection {
esc_connection_settings_t* settings;
const char* name;
void* discoverer_data;
esc_endpoint_discoverer_t discover;
socket_t* tcp_conn;
thread_t* manager_thread;
bool_t stop;
queue_t* send_queue;
queue_t* recv_queue;
error_t* last_error;
event_t* stopped_event;
} esc_connection_t;
error_t* esc_connection_last_error(esc_connection_t* conn);
void esc_connection_close(esc_connection_t* conn);
#endif //ESC_ESC_CONNECTION_H

21
src/credentials.c Normal file
View File

@@ -0,0 +1,21 @@
//
// Created by nicolas on 22/03/18.
//
#include <stdlib.h>
#include "utils/debug.h"
#include "utils/string.h"
#include "credentials.h"
esc_credentials_t* esc_credentials_create(const char* username, const char* password) {
esc_credentials_t* creds = malloc(sizeof(struct st_credentials));
creds->username = string_copy(username);
creds->password = string_copy(password);
return creds;
}
void esc_credentials_destroy(esc_credentials_t* creds) {
free((void*)creds->username);
free((void*)creds->password);
free(creds);
}

17
src/credentials.h Normal file
View File

@@ -0,0 +1,17 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_ESC_CREDENTIALS_H
#define ESC_ESC_CREDENTIALS_H
struct st_credentials {
const char* username;
const char* password;
};
typedef struct st_credentials esc_credentials_t;
esc_credentials_t* esc_credentials_create(const char* username, const char* password);
void esc_credentials_destroy(esc_credentials_t* creds);
#endif //ESC_ESC_CREDENTIALS_H

16
src/endpoint_discoverer.c Normal file
View File

@@ -0,0 +1,16 @@
//
// Created by nicolas on 22/03/18.
//
#include "utils/debug.h"
#include "endpoint_discoverer.h"
const esc_node_endpoints_t* esc_static_discover(const esc_static_endpoint_discoverer_t* discover_data, const ip_endpoint_t* failed_tcp_endpoint) {
esc_node_endpoints_t* result = malloc(sizeof(struct st_esc_node_endpoints));
if (discover_data->use_ssl_connection) {
result->secure_tcp_endpoint = discover_data->tcp_endpoint;
} else {
result->tcp_endpoint = discover_data->tcp_endpoint;
}
return result;
}

30
src/endpoint_discoverer.h Normal file
View File

@@ -0,0 +1,30 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_ESC_ENDPOINT_DISCOVERER_H
#define ESC_ESC_ENDPOINT_DISCOVERER_H
#include "utils/bool.h"
typedef struct st_ip_endpoint {
char* host;
unsigned short port;
} ip_endpoint_t;
typedef struct st_esc_node_endpoints {
ip_endpoint_t tcp_endpoint;
ip_endpoint_t secure_tcp_endpoint;
} esc_node_endpoints_t;
typedef esc_node_endpoints_t* (*esc_endpoint_discoverer_t)(const void* discover_data, const ip_endpoint_t* failed_tcp_endpoint);
struct st_esc_static_endpoint_discoverer {
ip_endpoint_t tcp_endpoint;
bool_t use_ssl_connection;
};
typedef struct st_esc_static_endpoint_discoverer esc_static_endpoint_discoverer_t;
const esc_node_endpoints_t* esc_static_discover(const esc_static_endpoint_discoverer_t* discover_data, const ip_endpoint_t* failed_tcp_endpoint);
#endif //ESC_ESC_ENDPOINT_DISCOVERER_H

410
src/esc.c
View File

@@ -1,410 +0,0 @@
//
// Created by nicol on 2018-03-18.
//
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include "esc.h"
#include "utils/debug.h"
#include "utils/socket.h"
#include "proto.h"
#include "tcp_package.h"
#include "utils/thread.h"
#include "utils/mutex.h"
#include "utils/string.h"
#include "utils/queue.h"
#include "tcp_messages.h"
#ifdef _WIN32
#include <windows.h>
#define usleep Sleep
#endif
#ifdef __linux__
#include <unistd.h>
#endif
typedef int bool_t;
const bool_t BOOL_TRUE = 1;
const bool_t BOOL_FALSE = 0;
const esc_error_t _error_create(const char* file, int line, int code, char* format, ...) {
va_list vl;
va_start(vl, format);
esc_error_t err;
err.file = file;
err.line = line;
err.code = code;
vsnprintf(err.message, ERROR_MSG_SIZE, format, vl);
return err;
}
#define error_create(code, fmt, args...) _error_create(__FILE__, __LINE__, code, fmt, args)
struct st_connection_settings {
bool_t use_ssl_connection;
};
const esc_connection_settings_t default_connection_settings = {
};
const esc_connection_settings_t* esc_default_connection_settings = &default_connection_settings;
struct st_tcp_endpoint {
char* host;
unsigned short port;
};
typedef struct st_tcp_endpoint tcp_endpoint_t;
struct st_node_endpoints {
tcp_endpoint_t tcp_endpoint;
tcp_endpoint_t secure_tcp_endpoint;
};
typedef struct st_node_endpoints node_endpoints_t;
typedef const node_endpoints_t* (*endpoint_discoverer_t)(const void* discover_data, const tcp_endpoint_t* failed_tcp_endpoint);
struct st_connection {
esc_connection_settings_t settings;
const char* name;
void* discoverer_data;
endpoint_discoverer_t discover;
socket_t tcp_conn;
ProtobufCAllocator protobuf_c_allocator;
thread_t manager_thread;
bool_t stop;
mutex_t recv_peek_lock;
queue_t send_queue;
queue_t recv_queue;
bool_t has_error;
esc_error_t last_error;
};
struct st_static_endpoint_discoverer {
tcp_endpoint_t tcp_endpoint;
bool_t use_ssl_connection;
};
typedef struct st_static_endpoint_discoverer static_endpoint_discoverer_t;
const node_endpoints_t* static_discover(const static_endpoint_discoverer_t* discover_data, const tcp_endpoint_t* failed_tcp_endpoint) {
node_endpoints_t* result = malloc(sizeof(struct st_node_endpoints));
if (discover_data->use_ssl_connection) {
result->secure_tcp_endpoint = discover_data->tcp_endpoint;
} else {
result->tcp_endpoint = discover_data->tcp_endpoint;
}
return result;
}
//TODO partial transfer
ssize_t connection_send_tcp_package(esc_connection_t conn, tcp_package_t pkg) {
char uuid_buf[37];
#ifdef _WIN32
fprintf(stderr, "connection_send_tcp_package: %s %u %s %llu ", get_string_for_tcp_message(pkg->command), pkg->flags, esc_uuid_format(pkg->correlation_id, uuid_buf, 37), buffer_size(pkg->data));
#else
fprintf(stderr, "connection_send_tcp_package: %s %u %s %lu ", get_string_for_tcp_message(pkg->command), pkg->flags, esc_uuid_format(pkg->correlation_id, uuid_buf, 37), buffer_size(pkg->data));
#endif
buffer_t send_buffer = tcp_package_to_buffer(pkg);
size_t send_buffer_size = buffer_size(send_buffer);
uint32_t size = (uint32_t)send_buffer_size;
ssize_t rc;
if ((rc = socket_send(conn->tcp_conn, (char *) &size, sizeof(uint32_t))) <= 0) {
buffer_destroy(send_buffer);
fprintf(stderr, "%d\n", socket_error(conn->tcp_conn));
return -1;
}
uint8_t* send_buffer_data = buffer_data(send_buffer);
if ((rc = socket_send(conn->tcp_conn, (char *) send_buffer_data, send_buffer_size)) <= 0) {
buffer_destroy(send_buffer);
fprintf(stderr, "%d\n", socket_error(conn->tcp_conn));
return -2;
}
buffer_destroy(send_buffer);
fprintf(stderr, "0\n");
return 0;
}
//TODO partial transfer
tcp_package_t connection_recv_tcp_package(esc_connection_t conn) {
uint32_t recv_size;
ssize_t rc;
if ((rc = socket_recv(conn->tcp_conn, (char *)&recv_size, sizeof(uint32_t))) != 4) {
#ifdef _WIN32
fprintf(stderr, "connection_recv_tcp_package: %lld %d\n", rc, socket_error(conn->tcp_conn));
#else
fprintf(stderr, "connection_recv_tcp_package: %ld %d\n", rc, socket_error(conn->tcp_conn));
#endif
return 0;
}
buffer_t recv_buffer = buffer_create(recv_size);
size_t recv_buffer_size = recv_size;
uint8_t* recv_buffer_data = buffer_data(recv_buffer);
while(recv_size > 0) {
size_t pos = recv_buffer_size - recv_size;
rc = socket_recv(conn->tcp_conn, (char *)&recv_buffer_data[pos], recv_size);
recv_size -= rc;
}
tcp_package_t recv_pkg = tcp_package_from_buffer(recv_buffer);
buffer_destroy(recv_buffer);
char uuid_buf[37];
#ifdef _WIN32
fprintf(stderr, "connection_recv_tcp_package: %s %u %s %llu\n", get_string_for_tcp_message(recv_pkg->command), recv_pkg->flags, esc_uuid_format(recv_pkg->correlation_id, uuid_buf, 37), buffer_size(recv_pkg->data));
#else
fprintf(stderr, "connection_recv_tcp_package: %s %u %s %lu\n", get_string_for_tcp_message(recv_pkg->command), recv_pkg->flags, esc_uuid_format(recv_pkg->correlation_id, uuid_buf, 37), buffer_size(recv_pkg->data));
#endif
return recv_pkg;
}
inline void connection_enqueue_send(esc_connection_t conn, tcp_package_t pkg) {
queue_enqueue(conn->send_queue, pkg);
}
tcp_package_t connection_wait_for(esc_connection_t conn, esc_uuid_t correlation_id) {
tcp_package_t found = 0;
while(found == 0) {
mutex_lock(conn->recv_peek_lock);
tcp_package_t peek = queue_peek(conn->recv_queue);
if (peek && esc_uuid_compare(peek->correlation_id, correlation_id) == 0) {
queue_dequeue(conn->recv_queue);
found = peek;
}
mutex_unlock(conn->recv_peek_lock);
usleep(1);
}
return found;
}
void* protobuf_c_alloc(void *alloc_data, size_t size) {
return malloc(size);
}
void protobuf_c_free(void *alloc_data, void* p) {
free(p);
}
esc_connection_t esc_connection_create(const esc_connection_settings_t* connection_settings, const char* addr, const char* connection_name) {
struct st_connection* conn = malloc(sizeof(struct st_connection));
conn->settings = *connection_settings;
conn->protobuf_c_allocator.alloc = protobuf_c_alloc;
conn->protobuf_c_allocator.free = protobuf_c_free;
conn->protobuf_c_allocator.allocator_data = 0;
if (connection_name == 0 || strcmp(connection_name, "") == 0) {
esc_uuid_t uuid = esc_uuid_create();
char buf[40];
esc_uuid_format(uuid, buf, 40);
conn->name = string_copy(buf);
esc_uuid_destroy(uuid);
} else {
conn->name = connection_name;
}
conn->send_queue = queue_create();
conn->recv_peek_lock = mutex_create();
conn->recv_queue = queue_create();
conn->stop = 0;
conn->has_error = 0;
memset(&conn->last_error, 0, sizeof(struct st_error));
if (strncmp(addr, "tcp://", 6) != 0) {
conn->has_error = 1;
conn->last_error = error_create(1, "invalid schema for address: %s", addr);
return conn;
}
char* pos = strrchr(addr, ':');
if (pos == 0) {
conn->has_error = 1;
conn->last_error = error_create(2, "missing port in address: %s", addr);
return conn;
}
char* host = malloc(pos - addr - 5);
strncpy(host, addr+6, pos-addr-6);
host[pos-addr-6] = 0;
unsigned short port = (unsigned short)atoi(pos+1);
static_endpoint_discoverer_t discover_data = {
{host, port}, connection_settings->use_ssl_connection
};
conn->discoverer_data = malloc(sizeof(struct st_static_endpoint_discoverer));
memcpy(conn->discoverer_data, &discover_data, sizeof(struct st_static_endpoint_discoverer));
conn->discover = (endpoint_discoverer_t)static_discover;
return conn;
}
void* connection_thread(void* arg) {
esc_connection_t conn = arg;
while(conn->stop == BOOL_FALSE) {
if (socket_writable(conn->tcp_conn)) {
tcp_package_t send_pkg = queue_dequeue(conn->send_queue);
if (send_pkg != 0) {
ssize_t rc = connection_send_tcp_package(conn, send_pkg);
if (rc == 0) {
tcp_package_destroy(send_pkg);
} else {
fprintf(stderr, "failed to send pkg.");
}
}
}
if (socket_readable(conn->tcp_conn)) {
tcp_package_t recv_pkg = connection_recv_tcp_package(conn);
if (recv_pkg == 0) {
//TODO Handle connection lost
conn->stop = 1;
} else if (recv_pkg->command == MESSAGE_HEARTBEATREQUEST) {
tcp_package_t heartbeat_pkg = tcp_package_create(MESSAGE_HEARTBEATRESPONSE, recv_pkg->correlation_id, buffer_create(0));
queue_enqueue(conn->send_queue, heartbeat_pkg);
} else {
queue_enqueue(conn->recv_queue, recv_pkg);
}
}
usleep(1);
}
printf("manager thread stopping\n");
}
// return 0 on success
// return non-zero on failure and sets last_error on connection
int esc_connection_connect(esc_connection_t conn) {
// Discover endpoint
const node_endpoints_t* endpoints = conn->discover(conn->discoverer_data, 0);
if (endpoints == 0) {
return -1;
}
// Establish Tcp Connection
conn->tcp_conn = socket_create(SOCKET_TYPE_TCP);
tcp_endpoint_t endpoint = conn->settings.use_ssl_connection ? endpoints->secure_tcp_endpoint : endpoints->tcp_endpoint;
if (socket_connect(conn->tcp_conn, endpoint.host, endpoint.port) != 0) {
conn->last_error = error_create(socket_error(conn->tcp_conn), "can't connect to %s:%d", endpoint.host, endpoint.port);
return -1;
}
// Start manager thread
conn->manager_thread = thread_create(connection_thread, conn);
if (thread_start(conn->manager_thread)) {
conn->last_error = error_create(thread_error(conn->manager_thread), "can't start manager thread.", "");
return -1;
}
// Identify
// build message
EventStore__Client__Messages__IdentifyClient identify_client;
event_store__client__messages__identify_client__init(&identify_client);
identify_client.connection_name = (char*)conn->name;
identify_client.version = 1;
size_t s = event_store__client__messages__identify_client__get_packed_size(&identify_client);
uint8_t buffer[s];
event_store__client__messages__identify_client__pack(&identify_client, buffer);
// build tcp_package
esc_uuid_t correlation_id = esc_uuid_create();
tcp_package_t send_pkg = tcp_package_create(MESSAGE_IDENTIFYCLIENT, correlation_id, buffer_from(buffer, s));
queue_enqueue(conn->send_queue, send_pkg);
tcp_package_t recv_pkg = connection_wait_for(conn, correlation_id);
if (recv_pkg->command != MESSAGE_CLIENTIDENTIFIED) {
conn->last_error = error_create(0, "server error: %d.", recv_pkg->command);
tcp_package_destroy(recv_pkg);
return -1;
}
tcp_package_destroy(recv_pkg);
return 0;
}
esc_credentials_t esc_credentials_create(const char* username, const char* password) {
esc_credentials_t creds = malloc(sizeof(struct st_credentials));
creds->username = string_copy(username);
creds->password = string_copy(password);
return creds;
}
const esc_recorded_event_t* recorded_event_create(EventStore__Client__Messages__EventRecord* msg) {
if (msg == 0) return 0;
esc_recorded_event_t* ev = malloc(sizeof(struct st_recorded_event));
ev->event_id = esc_uuid_from(msg->event_id.data, msg->event_id.len);
ev->event_type = string_copy(msg->event_type);
ev->event_number = msg->event_number;
ev->event_stream_id = string_copy(msg->event_stream_id);
ev->created_epoch = msg->has_created_epoch ? msg->created_epoch : 0;
ev->data = buffer_from(msg->data.data, msg->data.len);
ev->metadata = buffer_from(msg->metadata.data, msg->metadata.len);
return ev;
}
esc_resolved_event_t* resolved_event_create(EventStore__Client__Messages__ResolvedEvent* msg) {
esc_resolved_event_t* ev = malloc(sizeof(struct st_resolved_event));
ev->original_position.prepare_position = msg->prepare_position;
ev->original_position.commit_position = msg->commit_position;
ev->event = recorded_event_create(msg->event);
ev->link = recorded_event_create(msg->link);
return ev;
}
esc_all_events_slice_t esc_connection_read_all_forward(esc_connection_t conn, const esc_position_t* last_checkpoint, unsigned int count, esc_credentials_t credentials) {
//TODO function for packing from client struct direct to protobuf data
EventStore__Client__Messages__ReadAllEvents send_msg;
event_store__client__messages__read_all_events__init(&send_msg);
send_msg.prepare_position = last_checkpoint ? last_checkpoint->prepare_position : 0;
send_msg.commit_position = last_checkpoint ? last_checkpoint->commit_position : 0;
send_msg.max_count = count;
send_msg.require_master = 0;
send_msg.resolve_link_tos = 0;
size_t s = event_store__client__messages__read_all_events__get_packed_size(&send_msg);
uint8_t buffer[s];
event_store__client__messages__read_all_events__pack(&send_msg, buffer);
esc_uuid_t correlation_id = esc_uuid_create();
tcp_package_t send_pkg = tcp_package_create_authenticated(MESSAGE_READALLEVENTSFORWARD, correlation_id, buffer_from(buffer, s), credentials->username, credentials->password);
queue_enqueue(conn->send_queue, send_pkg);
tcp_package_t recv_pkg = connection_wait_for(conn, correlation_id);
if (recv_pkg->command != MESSAGE_READALLEVENTSFORWARDCOMPLETED) {
conn->last_error = error_create(recv_pkg->command, "server error: %s", get_string_for_tcp_message(recv_pkg->command));
return 0;
}
//TODO function for unpacking from protobuf data to client struct
size_t data_size = buffer_size(recv_pkg->data);
uint8_t* data = buffer_data(recv_pkg->data);
EventStore__Client__Messages__ReadAllEventsCompleted *recv_msg =
event_store__client__messages__read_all_events_completed__unpack(&conn->protobuf_c_allocator, data_size, data);
esc_all_events_slice_t result = malloc(sizeof(struct st_all_events_slice));
result->read_direction = "forward";
result->is_end_of_stream = (recv_msg->events == 0 || recv_msg->n_events == 0) ? 1 : 0;
result->from_position.prepare_position = recv_msg->prepare_position;
result->from_position.commit_position = recv_msg->commit_position;
result->next_position.prepare_position = recv_msg->next_prepare_position;
result->next_position.commit_position = recv_msg->next_commit_position;
result->n_events = recv_msg->n_events;
result->events = malloc(recv_msg->n_events * sizeof(void*));
for (size_t i = 0; i < recv_msg->n_events; i++) {
result->events[i] = resolved_event_create(recv_msg->events[i]);
}
event_store__client__messages__read_all_events_completed__free_unpacked(recv_msg, &conn->protobuf_c_allocator);
return result;
}
void esc_connection_close(esc_connection_t conn) {
conn->stop = 1;
socket_close(conn->tcp_conn);
}
inline esc_error_t esc_connection_last_error(esc_connection_t conn) {
return conn->last_error;
}
const char* esc_position_format(const esc_position_t* position, char* buffer, size_t buf_size) {
#ifdef _WIN32
snprintf(buffer, buf_size, "%llu/%llu", position->prepare_position, position->commit_position);
#else
snprintf(buffer, buf_size, "%lu/%lu", position->prepare_position, position->commit_position);
#endif
return buffer;
}

80
src/esc.h
View File

@@ -1,80 +0,0 @@
//
// Created by nicol on 2018-03-18.
//
#ifndef ESC_ESC_H
#define ESC_ESC_H
#include "utils/uuid.h"
#include "utils/buffer.h"
#define ERROR_MSG_SIZE 1024
struct st_error {
const char* file;
int line;
int code;
char message[ERROR_MSG_SIZE];
};
typedef struct st_error esc_error_t;
struct st_connection_settings;
typedef struct st_connection_settings esc_connection_settings_t;
struct st_connection;
typedef struct st_connection* esc_connection_t;
struct st_credentials {
const char* username;
const char* password;
};
typedef struct st_credentials* esc_credentials_t;
struct st_esc_position {
int64_t prepare_position;
int64_t commit_position;
};
typedef struct st_esc_position esc_position_t;
struct st_recorded_event {
esc_uuid_t event_id;
const char* event_type;
int64_t event_number;
const char *event_stream_id;
int64_t created_epoch;
buffer_t data;
buffer_t metadata;
};
typedef struct st_recorded_event esc_recorded_event_t;
struct st_resolved_event {
const esc_recorded_event_t* event;
const esc_recorded_event_t* link;
esc_position_t original_position;
};
typedef struct st_resolved_event esc_resolved_event_t;
struct st_all_events_slice {
char* read_direction;
esc_position_t from_position;
esc_position_t next_position;
size_t n_events;
esc_resolved_event_t** events;
int is_end_of_stream;
};
typedef struct st_all_events_slice* esc_all_events_slice_t;
const esc_connection_settings_t* esc_default_connection_settings;
// Connection
esc_connection_t esc_connection_create(const esc_connection_settings_t* connection_settings, const char* addr, const char* connection_name);
int esc_connection_connect(esc_connection_t conn);
esc_credentials_t esc_credentials_create(const char* username, const char* password);
esc_all_events_slice_t esc_connection_read_all_forward(esc_connection_t conn, const esc_position_t* last_checkpoint, unsigned int count, esc_credentials_t credentials);
void esc_connection_close(esc_connection_t conn);
esc_error_t esc_connection_last_error(esc_connection_t conn);
// Utils
const char* esc_position_format(const esc_position_t* position, char* buffer, size_t buf_size);
#endif //ESC_ESC_H

25
src/event_data.c Normal file
View File

@@ -0,0 +1,25 @@
//
// Created by nicolas on 23/03/18.
//
#include "event_data.h"
#include "utils/debug.h"
#include "utils/string.h"
esc_event_data_t* esc_event_data_create(uuid_t* event_id, const char* event_type, bool_t is_json, buffer_t* data, buffer_t* metadata) {
esc_event_data_t* event_data = malloc(sizeof(esc_event_data_t));
event_data->event_id = uuid_copy(event_id);
event_data->event_type = string_copy(event_type);
event_data->is_json = is_json;
event_data->data = data ? buffer_copy(data) : 0;
event_data->metadata = metadata ? buffer_copy(metadata) : 0;
return event_data;
}
void esc_event_data_destroy(esc_event_data_t* event_data) {
uuid_destroy(event_data->event_id);
free((void*)event_data->event_type);
if (event_data->data) buffer_destroy(event_data->data);
if (event_data->metadata) buffer_destroy(event_data->metadata);
free(event_data);
}

23
src/event_data.h Normal file
View File

@@ -0,0 +1,23 @@
//
// Created by nicolas on 23/03/18.
//
#ifndef ESC_EVENT_DATA_H
#define ESC_EVENT_DATA_H
#include "utils/uuid.h"
#include "utils/bool.h"
#include "utils/buffer.h"
typedef struct st_esc_event_data {
uuid_t* event_id;
const char* event_type;
bool_t is_json;
buffer_t* data;
buffer_t* metadata;
} esc_event_data_t;
esc_event_data_t* esc_event_data_create(uuid_t* event_id, const char* event_type, bool_t is_json, buffer_t* data, buffer_t* metadata);
void esc_event_data_destroy(esc_event_data_t*);
#endif //ESC_EVENT_DATA_H

68
src/main.c
View File

@@ -1,68 +0,0 @@
#include <stdio.h>
#include "utils/debug.h"
#include "esc.h"
#ifdef _WIN32
#include <winsock.h>
#include <windows.h>
#define usleep Sleep
#define sleep(x) Sleep(x*1000)
#endif
#ifdef __linux__
#include <unistd.h>
#endif
int main() {
#ifdef _WIN32
WSADATA wsaData;
WSAStartup(MAKEWORD(2,0), &wsaData);
#endif
esc_connection_t conn = esc_connection_create(esc_default_connection_settings, "tcp://127.0.0.1:1113", NULL);
if (esc_connection_connect(conn) != 0) {
esc_error_t err = esc_connection_last_error(conn);
fprintf(stderr, "Error: %s code=%d file=%s line=%d", err.message, err.code, err.file, err.line);
return -2;
}
esc_credentials_t credentials = esc_credentials_create("admin", "changeit");
esc_all_events_slice_t result = 0;
do {
result = esc_connection_read_all_forward(conn, result ? &result->next_position : NULL, 100, credentials);
if (result == 0) {
esc_error_t err = esc_connection_last_error(conn);
fprintf(stderr, "Error: %s code=%d file=%s line=%d", err.message, err.code, err.file, err.line);
return -3;
}
char posbuf1[44];
char posbuf2[44];
printf("%s %s %s %u\n", result->read_direction,
esc_position_format(&result->from_position, posbuf1, 44),
esc_position_format(&result->next_position, posbuf2, 44),
result->is_end_of_stream);
char uuid_buf[37];
for (size_t i = 0; i < result->n_events; i++) {
#ifdef _WIN32
printf("%s %s %lld@%s %llu %llu\n", esc_uuid_format(result->events[i]->event->event_id, uuid_buf, 37),
#else
printf("%s %s %ld@%s %lu %lu\n", esc_uuid_format(result->events[i]->event->event_id, uuid_buf, 37),
#endif
result->events[i]->event->event_type,
result->events[i]->event->event_number,
result->events[i]->event->event_stream_id,
buffer_size(result->events[i]->event->data),
buffer_size(result->events[i]->event->metadata));
}
} while(result->is_end_of_stream == 0);
sleep(10);
esc_connection_close(conn);
dbg_list_allocs();
#ifdef _WIN32
WSACleanup();
#endif
return 0;
}

37
src/position.c Normal file
View File

@@ -0,0 +1,37 @@
//
// Created by nicolas on 22/03/18.
//
#include <stdint.h>
#include <stddef.h>
#include <stdio.h>
#include "position.h"
#include "utils/debug.h"
esc_position_t* esc_position_create(int64_t prepare_position, int64_t commit_position) {
esc_position_t* res = malloc(sizeof(struct st_esc_position));
res->prepare_position = prepare_position;
res->commit_position = commit_position;
return res;
}
inline void esc_position_destroy(esc_position_t* pos) {
free(pos);
}
int esc_position_compare(esc_position_t* left, esc_position_t* right) {
if ((left->commit_position < right->commit_position) || ((left->commit_position == right->commit_position) && (left->prepare_position < right->prepare_position)))
return -1;
if ((left->commit_position > right->commit_position) || ((left->commit_position == right->commit_position) && (left->prepare_position > right->prepare_position)))
return 1;
return 0;
}
const char* esc_position_format(esc_position_t* position, char* buffer, size_t buf_size) {
#ifdef _WIN32
snprintf(buffer, buf_size, "%llu/%llu", position->prepare_position, position->commit_position);
#else
snprintf(buffer, buf_size, "%lu/%lu", position->prepare_position, position->commit_position);
#endif
return buffer;
}

20
src/position.h Normal file
View File

@@ -0,0 +1,20 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_ESC_POSITION_H
#define ESC_ESC_POSITION_H
#include <stdint.h>
typedef struct st_esc_position {
int64_t prepare_position;
int64_t commit_position;
} esc_position_t;
esc_position_t* esc_position_create(int64_t prepare_position, int64_t commit_position);
void esc_position_destroy(esc_position_t*);
int esc_position_compare(esc_position_t* left, esc_position_t* right);
const char* esc_position_format(esc_position_t* position, char* buffer, size_t buf_len);
#endif //ESC_ESC_POSITION_H

213
src/proto_helper.c Normal file
View File

@@ -0,0 +1,213 @@
//
// Created by nicolas on 22/03/18.
//
#include "utils/debug.h"
#include "utils/string.h"
#include "results.h"
#include "proto_helper.h"
#include "proto.h"
#include "utils/bool.h"
#include "event_data.h"
void* protobuf_c_alloc(void *alloc_data, size_t size) {
return malloc(size);
}
void protobuf_c_free(void *alloc_data, void* p) {
free(p);
}
ProtobufCAllocator protobuf_allocator = {
protobuf_c_alloc, protobuf_c_free, 0
};
inspection_result_t* inspection_result_create(operation_decision_t decision, const char* description, error_t* error) {
inspection_result_t* inspection_result = malloc(sizeof(inspection_result_t));
inspection_result->decision = decision;
inspection_result->description = description;
inspection_result->error = error;
return inspection_result;
}
void inspection_result_destroy(inspection_result_t* inspection_result) {
free(inspection_result->error);
free(inspection_result);
}
buffer_t* esc_identify_client_pack(const char* connection_name) {
EventStore__Client__Messages__IdentifyClient identify_client;
event_store__client__messages__identify_client__init(&identify_client);
identify_client.connection_name = (char*)connection_name;
identify_client.version = 1;
size_t msg_size = event_store__client__messages__identify_client__get_packed_size(&identify_client);
uint8_t msg_buf[msg_size];
event_store__client__messages__identify_client__pack(&identify_client, msg_buf);
return buffer_copyfrom(msg_buf, msg_size);
}
esc_recorded_event_t* esc_recorded_event_unpack(EventStore__Client__Messages__EventRecord* msg) {
if (msg == 0) return 0;
esc_recorded_event_t* ev = malloc(sizeof(struct st_recorded_event));
ev->event_id = uuid_from(msg->event_id.data, msg->event_id.len);
ev->event_type = string_copy(msg->event_type);
ev->event_number = msg->event_number;
ev->event_stream_id = string_copy(msg->event_stream_id);
ev->created_epoch = msg->has_created_epoch ? msg->created_epoch : 0;
ev->data = buffer_copyfrom(msg->data.data, msg->data.len);
ev->metadata = buffer_copyfrom(msg->metadata.data, msg->metadata.len);
//event_store__client__messages__event_record__free_unpacked(msg, &protobuf_allocator);
return ev;
}
esc_resolved_event_t* esc_resolved_event_unpack(EventStore__Client__Messages__ResolvedEvent* msg) {
esc_resolved_event_t* ev = malloc(sizeof(struct st_resolved_event));
ev->original_position = esc_position_create(msg->prepare_position, msg->commit_position);
ev->event = esc_recorded_event_unpack(msg->event);
ev->link = esc_recorded_event_unpack(msg->link);
//event_store__client__messages__resolved_event__free_unpacked(msg, &protobuf_allocator);
return ev;
}
const char* get_string_for_all_events_result(int result) {
switch(result) {
case EVENT_STORE__CLIENT__MESSAGES__READ_ALL_EVENTS_COMPLETED__READ_ALL_RESULT__Success: return "Success";
case EVENT_STORE__CLIENT__MESSAGES__READ_ALL_EVENTS_COMPLETED__READ_ALL_RESULT__NotModified: return "Not Modified";
case EVENT_STORE__CLIENT__MESSAGES__READ_ALL_EVENTS_COMPLETED__READ_ALL_RESULT__AccessDenied: return "Access Denied";
case EVENT_STORE__CLIENT__MESSAGES__READ_ALL_EVENTS_COMPLETED__READ_ALL_RESULT__Error: return "Error";
default: return "Unknown";
}
}
inspection_result_t* esc_all_events_slice_unpack(buffer_t* buffer, esc_all_events_slice_t** all_events_slice_p) {
EventStore__Client__Messages__ReadAllEventsCompleted *recv_msg =
event_store__client__messages__read_all_events_completed__unpack(
&protobuf_allocator,
buffer_size(buffer),
buffer_data(buffer));
inspection_result_t* inspection_result;
switch (recv_msg->result) {
case EVENT_STORE__CLIENT__MESSAGES__READ_ALL_EVENTS_COMPLETED__READ_ALL_RESULT__Success: {
esc_all_events_slice_t *all_events_slice = malloc(sizeof(struct st_all_events_slice));
all_events_slice->read_direction = "forward";
all_events_slice->is_end_of_stream = (recv_msg->events == 0 || recv_msg->n_events == 0) ? 1 : 0;
all_events_slice->from_position = esc_position_create(recv_msg->prepare_position,
recv_msg->commit_position);
all_events_slice->next_position = esc_position_create(recv_msg->next_prepare_position,
recv_msg->next_commit_position);
all_events_slice->n_events = recv_msg->n_events;
all_events_slice->events = malloc(recv_msg->n_events * sizeof(void *));
for (size_t i = 0; i < recv_msg->n_events; i++) {
all_events_slice->events[i] = esc_resolved_event_unpack(recv_msg->events[i]);
}
*all_events_slice_p = all_events_slice;
inspection_result = inspection_result_create(Operation_Decision_EndOperation, "Success", 0);
break;
}
case EVENT_STORE__CLIENT__MESSAGES__READ_ALL_EVENTS_COMPLETED__READ_ALL_RESULT__Error: {
error_t* error = error_create(recv_msg->result, "Server error: %s", recv_msg->error);
inspection_result = inspection_result_create(Operation_Decision_EndOperation, "Error", error);
break;
}
case EVENT_STORE__CLIENT__MESSAGES__READ_ALL_EVENTS_COMPLETED__READ_ALL_RESULT__AccessDenied: {
error_t* error = error_create(recv_msg->result, "Access denied: %s %s", "Read", "All");
inspection_result = inspection_result_create(Operation_Decision_EndOperation, "AccessDenied", error);
break;
}
default: {
error_t* error = error_create(recv_msg->result, "Unexpected ReadStreamResult %s", get_string_for_all_events_result(recv_msg->result));
inspection_result = inspection_result_create(Operation_Decision_EndOperation, "Unexpected", error);
break;
}
}
event_store__client__messages__read_all_events_completed__free_unpacked(recv_msg, &protobuf_allocator);
return inspection_result;
}
buffer_t* esc_read_all_forward_pack(esc_position_t* last_checkpoint, int32_t count, bool_t resolve_link_tos) {
EventStore__Client__Messages__ReadAllEvents send_msg;
event_store__client__messages__read_all_events__init(&send_msg);
send_msg.prepare_position = last_checkpoint ? last_checkpoint->prepare_position : 0;
send_msg.commit_position = last_checkpoint ? last_checkpoint->commit_position : 0;
send_msg.max_count = count;
send_msg.require_master = 0;
send_msg.resolve_link_tos = resolve_link_tos;
size_t s = event_store__client__messages__read_all_events__get_packed_size(&send_msg);
uint8_t buffer[s];
event_store__client__messages__read_all_events__pack(&send_msg, buffer);
return buffer_copyfrom(buffer, s);
}
const char* get_string_for_write_result(int write_result) {
switch(write_result) {
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__Success: return "Success";
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__PrepareTimeout: return "PrepareTimeout";
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__CommitTimeout: return "CommitTimeout";
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__ForwardTimeout: return "ForwardTimeout";
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__WrongExpectedVersion: return "WrongExpectedVersion";
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__StreamDeleted: return "StreamDeleted";
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__InvalidTransaction: return "InvalidTransaction";
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__AccessDenied: return "AccessDenied";
default: return "Unknown";
}
}
inspection_result_t* esc_write_result_unpack(buffer_t* buffer, esc_write_result_t** write_result_p) {
EventStore__Client__Messages__WriteEventsCompleted* recv_msg =
event_store__client__messages__write_events_completed__unpack(&protobuf_allocator, buffer_size(buffer), buffer_data(buffer));
inspection_result_t* inspection_result;
switch(recv_msg->result) {
case EVENT_STORE__CLIENT__MESSAGES__OPERATION_RESULT__Success: {
esc_write_result_t *write_result = malloc(sizeof(esc_write_result_t));
write_result->next_expected_version = recv_msg->last_event_number;
write_result->log_position = esc_position_create(
recv_msg->prepare_position ? recv_msg->prepare_position : -1,
recv_msg->commit_position ? recv_msg->commit_position : -1);
*write_result_p = write_result;
inspection_result = inspection_result_create(Operation_Decision_EndOperation, "Success", 0);
break;
}
default: {
error_t* error = error_create(recv_msg->result, "Unexpected OperationResult %s", get_string_for_write_result(recv_msg->result));
inspection_result = inspection_result_create(Operation_Decision_EndOperation, "Unexpected", error);
break;
}
}
event_store__client__messages__write_events_completed__free_unpacked(recv_msg, &protobuf_allocator);
return inspection_result;
}
buffer_t* esc_append_to_stream_pack(const char* stream, int64_t expected_version, array_t* events) {
EventStore__Client__Messages__WriteEvents send_msg;
event_store__client__messages__write_events__init(&send_msg);
send_msg.event_stream_id = (char*)stream;
send_msg.expected_version = expected_version;
send_msg.require_master = 0;
send_msg.n_events = events->size;
send_msg.events = malloc(events->size * sizeof(EventStore__Client__Messages__NewEvent*));
for (size_t i = 0; i < events->size; i++) {
esc_event_data_t* event_data = events->data[i];
EventStore__Client__Messages__NewEvent* new_event = malloc(sizeof(EventStore__Client__Messages__NewEvent));
event_store__client__messages__new_event__init(new_event);
new_event->event_id.data = (uint8_t *)event_data->event_id;
new_event->event_id.len = 16;
new_event->metadata.data = event_data->metadata ? buffer_data(event_data->metadata) : 0;
new_event->metadata.len = event_data->metadata ? buffer_size(event_data->metadata) : 0;
new_event->data.data = event_data->data ? buffer_data(event_data->data) : 0;
new_event->data.len = event_data->data ? buffer_size(event_data->data) : 0;
new_event->event_type = (char*)event_data->event_type;
new_event->data_content_type = event_data->is_json ? 1 : 0;
new_event->metadata_content_type = event_data->is_json ? 1 : 0;
new_event->has_metadata = event_data->metadata ? 1 : 0;
send_msg.events[i] = new_event;
}
size_t size = event_store__client__messages__write_events__get_packed_size(&send_msg);
uint8_t buffer[size];
event_store__client__messages__write_events__pack(&send_msg, buffer);
for (size_t i = 0; i < events->size; i++) {
free(send_msg.events[i]);
}
free(send_msg.events);
return buffer_copyfrom(buffer, size);
}

39
src/proto_helper.h Normal file
View File

@@ -0,0 +1,39 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_PROTO_HELPER_H
#define ESC_PROTO_HELPER_H
#include "utils/array.h"
#include "utils/buffer.h"
#include "utils/bool.h"
#include "utils/error.h"
#include "results.h"
typedef enum {
Operation_Decision_DoNothing,
Operation_Decision_EndOperation,
Operation_Decision_Retry,
Operation_Decision_Reconnect,
Operation_Decision_Subscribed
} operation_decision_t;
typedef struct {
operation_decision_t decision;
const char* description;
error_t* error;
} inspection_result_t;
void inspection_result_destroy(inspection_result_t*);
buffer_t* esc_identify_client_pack(const char* connection_name);
inspection_result_t* esc_all_events_slice_unpack(buffer_t* buffer, esc_all_events_slice_t** all_events_slice_p);
buffer_t* esc_read_all_forward_pack(esc_position_t* last_checkpoint, int32_t count, bool_t resolve_link_tos);
const char* get_string_for_all_events_result(int result);
inspection_result_t* esc_write_result_unpack(buffer_t* buffer, esc_write_result_t** write_result_p);
buffer_t* esc_append_to_stream_pack(const char* stream, int64_t expected_version, array_t* events);
const char* get_string_for_write_result(int result);
#endif //ESC_PROTO_HELPER_H

39
src/results.c Normal file
View File

@@ -0,0 +1,39 @@
//
// Created by nicolas on 22/03/18.
//
#include "utils/debug.h"
#include "position.h"
#include "results.h"
void esc_recorded_event_destroy(esc_recorded_event_t* recorded_event) {
uuid_destroy(recorded_event->event_id);
free((void*)recorded_event->event_type);
free((void*)recorded_event->event_stream_id);
buffer_destroy(recorded_event->data);
buffer_destroy(recorded_event->metadata);
free(recorded_event);
}
void esc_resolved_event_destroy(esc_resolved_event_t* resolved_event) {
if (resolved_event->original_position) esc_position_destroy(resolved_event->original_position);
if (resolved_event->event) esc_recorded_event_destroy(resolved_event->event);
if (resolved_event->link) esc_recorded_event_destroy(resolved_event->link);
free(resolved_event);
}
void esc_all_events_slice_destroy(esc_all_events_slice_t* all_events_slice) {
esc_position_destroy(all_events_slice->from_position);
esc_position_destroy(all_events_slice->next_position);
for (size_t i = 0; i < all_events_slice->n_events; i++) {
esc_resolved_event_destroy(all_events_slice->events[i]);
}
free(all_events_slice->events);
free(all_events_slice);
}
void esc_write_result_destroy(esc_write_result_t* write_result) {
esc_position_destroy(write_result->log_position);
free(write_result);
}

50
src/results.h Normal file
View File

@@ -0,0 +1,50 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_RESULTS_H
#define ESC_RESULTS_H
#include "utils/uuid.h"
#include "utils/buffer.h"
#include "position.h"
struct st_recorded_event {
uuid_t* event_id;
const char* event_type;
int64_t event_number;
const char *event_stream_id;
int64_t created_epoch;
buffer_t* data;
buffer_t* metadata;
};
typedef struct st_recorded_event esc_recorded_event_t;
struct st_resolved_event {
esc_recorded_event_t* event;
esc_recorded_event_t* link;
esc_position_t* original_position;
};
typedef struct st_resolved_event esc_resolved_event_t;
struct st_all_events_slice {
char* read_direction;
esc_position_t* from_position;
esc_position_t* next_position;
size_t n_events;
esc_resolved_event_t** events;
int is_end_of_stream;
};
typedef struct st_all_events_slice esc_all_events_slice_t;
typedef struct st_esc_write_result_t {
int64_t next_expected_version;
esc_position_t* log_position;
} esc_write_result_t;
void esc_recorded_event_destroy(esc_recorded_event_t* recorded_event);
void esc_resolved_event_destroy(esc_resolved_event_t* resolved_event);
void esc_all_events_slice_destroy(esc_all_events_slice_t* all_events_slice);
void esc_write_result_destroy(esc_write_result_t* write_result);
#endif //ESC_RESULTS_H

28
src/tcp_package.c
View File

@@ -16,37 +16,37 @@ const uint32_t CorrelationOffset = 2;
const uint32_t AuthOffset = 18;
const uint32_t MandatorySize = 18;
tcp_package_t tcp_package_create(uint8_t command, esc_uuid_t correlation_id, buffer_t data) {
tcp_package_t pkg = malloc(sizeof(struct st_tcp_package));
tcp_package_t* tcp_package_create(uint8_t command, uuid_t* correlation_id, buffer_t* data) {
tcp_package_t* pkg = malloc(sizeof(struct st_tcp_package));
pkg->command = command;
pkg->flags = 0;
pkg->correlation_id = esc_uuid_copy(correlation_id);
pkg->correlation_id = uuid_copy(correlation_id);
pkg->login = 0;
pkg->password = 0;
pkg->data = data;
return pkg;
}
tcp_package_t tcp_package_create_authenticated(uint8_t command, esc_uuid_t correlation_id, buffer_t data, const char* username, const char* password) {
tcp_package_t pkg = malloc(sizeof(struct st_tcp_package));
tcp_package_t* tcp_package_create_authenticated(uint8_t command, uuid_t* correlation_id, buffer_t* data, const char* username, const char* password) {
tcp_package_t* pkg = malloc(sizeof(struct st_tcp_package));
pkg->command = command;
pkg->flags = 1;
pkg->correlation_id = esc_uuid_copy(correlation_id);
pkg->correlation_id = uuid_copy(correlation_id);
pkg->login = string_copy(username);
pkg->password = string_copy(password);
pkg->data = data;
return pkg;
}
void tcp_package_destroy(tcp_package_t pkg) {
void tcp_package_destroy(tcp_package_t* pkg) {
free((void*)pkg->login);
free((void*)pkg->password);
esc_uuid_destroy(pkg->correlation_id);
uuid_destroy(pkg->correlation_id);
buffer_destroy(pkg->data);
free(pkg);
}
buffer_t tcp_package_to_buffer(tcp_package_t pkg) {
buffer_t* tcp_package_to_buffer(tcp_package_t* pkg) {
size_t data_size = buffer_size(pkg->data);
uint8_t* buf = malloc(MandatorySize + data_size + (pkg->flags ? 257*2 : 0));
buf[CommandOffset] = pkg->command;
@@ -73,10 +73,12 @@ buffer_t tcp_package_to_buffer(tcp_package_t pkg) {
}
fprintf(stderr, "\n");
*/
return buffer_from(buf, size);
buffer_t* res = buffer_copyfrom(buf, size);
free(buf);
return res;
}
tcp_package_t tcp_package_from_buffer(buffer_t buffer) {
tcp_package_t* tcp_package_from_buffer(buffer_t* buffer) {
size_t buf_size = buffer_size(buffer);
uint8_t* buf_data = buffer_data(buffer);
/*
@@ -85,10 +87,10 @@ tcp_package_t tcp_package_from_buffer(buffer_t buffer) {
}
fprintf(stderr, "\n");
*/
tcp_package_t pkg = malloc(sizeof(struct st_tcp_package));
tcp_package_t* pkg = malloc(sizeof(struct st_tcp_package));
pkg->command = buf_data[CommandOffset];
pkg->flags = buf_data[FlagsOffset];
pkg->correlation_id = esc_uuid_from(&buf_data[CorrelationOffset], 16);
pkg->correlation_id = uuid_from(&buf_data[CorrelationOffset], 16);
if (pkg->flags) {
size_t l_len = buf_data[AuthOffset];
size_t p_len = buf_data[AuthOffset+1+l_len];

16
src/tcp_package.h
View File

@@ -11,17 +11,17 @@
struct st_tcp_package {
uint8_t command;
uint8_t flags;
esc_uuid_t correlation_id;
uuid_t* correlation_id;
const char* login;
const char* password;
buffer_t data;
buffer_t* data;
};
typedef struct st_tcp_package* tcp_package_t;
typedef struct st_tcp_package tcp_package_t;
tcp_package_t tcp_package_create(uint8_t command, esc_uuid_t correlation_id, buffer_t data);
tcp_package_t tcp_package_create_authenticated(uint8_t command, esc_uuid_t correlation_id, buffer_t data, const char* username, const char* password);
void tcp_package_destroy(tcp_package_t pkg);
buffer_t tcp_package_to_buffer(tcp_package_t pkg);
tcp_package_t tcp_package_from_buffer(buffer_t buffer);
tcp_package_t* tcp_package_create(uint8_t command, uuid_t* correlation_id, buffer_t* data);
tcp_package_t* tcp_package_create_authenticated(uint8_t command, uuid_t* correlation_id, buffer_t* data, const char* username, const char* password);
void tcp_package_destroy(tcp_package_t* pkg);
buffer_t* tcp_package_to_buffer(tcp_package_t* pkg);
tcp_package_t* tcp_package_from_buffer(buffer_t* buffer);
#endif //ESC_TCP_PACKAGE_H

33
src/utils/array.c Normal file
View File

@@ -0,0 +1,33 @@
//
// Created by nicolas on 23/03/18.
//
#include <stdarg.h>
#include "array.h"
#include "debug.h"
array_t* array_create(size_t n, ...) {
va_list vl;
va_start(vl, n);
array_t* arr = malloc(sizeof(array_t));
arr->size = n;
arr->data = malloc(n * sizeof(void*));
for (size_t i = 0; i < n; i++) {
void* item = va_arg(vl, void*);
arr->data[i] = item;
}
return arr;
}
void array_destroy(array_t* array, array_deallocator destroyer) {
if (destroyer) {
for (size_t i=0;i<array->size;i++) {
destroyer(array->data[i]);
}
}
free(array->data);
free(array);
}

20
src/utils/array.h Normal file
View File

@@ -0,0 +1,20 @@
//
// Created by nicolas on 23/03/18.
//
#ifndef ESC_ARRAY_H
#define ESC_ARRAY_H
#include <stddef.h>
typedef struct st_array {
size_t size;
void** data;
} array_t;
typedef void (*array_deallocator)(void*);
array_t* array_create(size_t n, ...);
void array_destroy(array_t* array, array_deallocator destroyer);
#endif //ESC_ARRAY_H

13
src/utils/bool.h Normal file
View File

@@ -0,0 +1,13 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_BOOL_H
#define ESC_BOOL_H
typedef int bool_t;
#define BOOL_TRUE 1
#define BOOL_FALSE 0
#endif //ESC_BOOL_H

26
src/utils/buffer.c
View File

@@ -13,31 +13,31 @@ struct st_buffer {
uint8_t own_data;
};
buffer_t buffer_create(size_t size) {
buffer_t buf = malloc(sizeof(struct st_buffer));
buffer_t* buffer_create(size_t size) {
buffer_t* buf = malloc(sizeof(struct st_buffer));
buf->size = size;
buf->data = malloc(size);
buf->own_data = 1;
return buf;
}
inline void buffer_destroy(buffer_t buffer) {
inline void buffer_destroy(buffer_t* buffer) {
if (buffer->own_data) {
free(buffer->data);
}
free(buffer);
}
buffer_t buffer_from(const uint8_t* data, size_t size) {
buffer_t buf = malloc(sizeof(struct st_buffer));
buffer_t* buffer_from(const uint8_t* data, size_t size) {
buffer_t* buf = malloc(sizeof(struct st_buffer));
buf->size = size;
buf->data = (uint8_t*)data;
buf->own_data = 0;
return buf;
}
buffer_t buffer_copyfrom(const uint8_t* data, size_t size) {
buffer_t buf = malloc(sizeof(struct st_buffer));
buffer_t* buffer_copyfrom(const uint8_t* data, size_t size) {
buffer_t* buf = malloc(sizeof(struct st_buffer));
buf->size = size;
buf->data = malloc(size);
memcpy(buf->data, data, size);
@@ -45,10 +45,18 @@ buffer_t buffer_copyfrom(const uint8_t* data, size_t size) {
return buf;
}
inline size_t buffer_size(buffer_t buffer) {
inline size_t buffer_size(buffer_t* buffer) {
return buffer->size;
}
inline uint8_t* buffer_data(buffer_t buffer) {
inline uint8_t* buffer_data(buffer_t* buffer) {
return buffer->data;
}
buffer_t* buffer_from_string(const char* str) {
return buffer_copyfrom((uint8_t*)str, strlen(str));
}
buffer_t* buffer_copy(buffer_t* buffer) {
return buffer_copyfrom(buffer->data, buffer->size);
}

16
src/utils/buffer.h
View File

@@ -8,13 +8,15 @@
#include <stdint.h>
#include <stddef.h>
typedef struct st_buffer* buffer_t;
typedef struct st_buffer buffer_t;
buffer_t buffer_create(size_t size);
void buffer_destroy(buffer_t buffer);
buffer_t buffer_from(const uint8_t* data, size_t size);
buffer_t buffer_copyfrom(const uint8_t* data, size_t size);
size_t buffer_size(buffer_t buffer);
uint8_t* buffer_data(buffer_t buffer);
buffer_t* buffer_create(size_t size);
void buffer_destroy(buffer_t* buffer);
buffer_t* buffer_from(const uint8_t* data, size_t size);
buffer_t* buffer_from_string(const char* str);
buffer_t* buffer_copyfrom(const uint8_t* data, size_t size);
buffer_t* buffer_copy(buffer_t* other);
size_t buffer_size(buffer_t* buffer);
uint8_t* buffer_data(buffer_t* buffer);
#endif //ESC_BUFFER_H

19
src/utils/debug.c
View File

@@ -4,10 +4,14 @@
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
void* (*_malloc)(size_t) = malloc;
void (*_free)(void*) = free;
#include "debug.h"
#define DBG_ALLOC_INC_SIZE 1024
struct st_alloc {
const char* file;
int line;
@@ -20,8 +24,9 @@ struct st_alloc* _allocations = 0;
void* dbg_malloc(const char* file, int line, size_t size) {
if (_allocations_count == _allocations_size) {
_allocations = _allocations_size == 0 ? _malloc(1024*sizeof(struct st_alloc)) : realloc(_allocations, (_allocations_size + 1024)*sizeof(struct st_alloc));
_allocations_size += 1024;
_allocations = _allocations_size == 0 ? _malloc(DBG_ALLOC_INC_SIZE*sizeof(struct st_alloc)) : realloc(_allocations, (_allocations_size + DBG_ALLOC_INC_SIZE)*sizeof(struct st_alloc));
memset(&_allocations[_allocations_size], 0, DBG_ALLOC_INC_SIZE * sizeof(struct st_alloc));
_allocations_size += DBG_ALLOC_INC_SIZE;
}
void* ptr = _malloc(size);
struct st_alloc* alloc = &_allocations[_allocations_count++];
@@ -33,20 +38,24 @@ void* dbg_malloc(const char* file, int line, size_t size) {
}
void dbg_free(void* ptr) {
struct st_alloc* found = 0;
for (size_t i = 0; i < _allocations_count; i++) {
if (_allocations[i].ptr == ptr) _allocations[i].ptr = 0;
if (_allocations[i].ptr == ptr) found = &_allocations[i];
}
if (found) found->ptr = 0;
_free(ptr);
}
void dbg_list_allocs() {
size_t total = 0;
size_t count = 0;
printf("Memory leaks:\n");
for (size_t i = 0; i < _allocations_count; i++) {
if (_allocations[i].ptr == 0) continue;
printf("%lu %p (%lu) @ %s:%d\n", i, _allocations[i].ptr, _allocations[i].size, _allocations[i].file, _allocations[i].line);
total += _allocations[i].size;
count++;
}
printf("Total memory leaked: %lu\n", total);
printf("Totals: nb=%lu memory=%lu\n", count, total);
}

29
src/utils/error.c Normal file
View File

@@ -0,0 +1,29 @@
//
// Created by nicolas on 22/03/18.
//
#include <stdarg.h>
#include <stdio.h>
#include <assert.h>
#include <string.h>
#include "error.h"
#include "debug.h"
error_t* _error_create(const char* file, int line, int code, char* format, ...) {
va_list vl;
va_start(vl, format);
error_t* err = malloc(sizeof(struct st_error));
assert(err != 0);
err->file = file;
err->line = line;
err->code = code;
vsnprintf(err->message, ERROR_MSG_SIZE, format, vl);
return err;
}
#define error_create(code, fmt, args...) _error_create(__FILE__, __LINE__, code, fmt, args)
error_t* error_copy(error_t* err) {
error_t* copy = malloc(sizeof(error_t));
memcpy(copy, err, sizeof(error_t));
return copy;
}

21
src/utils/error.h Normal file
View File

@@ -0,0 +1,21 @@
//
// Created by nicolas on 22/03/18.
//
#ifndef ESC_ERROR_H
#define ESC_ERROR_H
#define ERROR_MSG_SIZE 1024
struct st_error {
const char* file;
int line;
int code;
char message[ERROR_MSG_SIZE];
};
typedef struct st_error error_t;
error_t* _error_create(const char* file, int line, int code, char* format, ...);
#define error_create(code, fmt, args...) _error_create(__FILE__, __LINE__, code, fmt, args)
error_t* error_copy(error_t* error);
#endif //ESC_ERROR_H

37
src/utils/event.c Normal file
View File

@@ -0,0 +1,37 @@
//
// Created by nicolas on 23/03/18.
//
#include "event.h"
#include "debug.h"
#ifdef __linux__
#include <pthread.h>
struct st_event {
pthread_mutex_t mutex;
pthread_cond_t handle;
};
event_t* event_create() {
event_t* event = malloc(sizeof(event_t));
pthread_mutex_init(&event->mutex, 0);
pthread_cond_init(&event->handle, 0);
return event;
}
void event_destroy(event_t* event) {
pthread_mutex_destroy(&event->mutex);
pthread_cond_destroy(&event->handle);
free(event);
}
void event_wait(event_t* event) {
pthread_cond_wait(&event->handle, &event->mutex);
}
void event_set(event_t* event) {
pthread_cond_broadcast(&event->handle);
}
#endif

15
src/utils/event.h Normal file
View File

@@ -0,0 +1,15 @@
//
// Created by nicolas on 23/03/18.
//
#ifndef ESC_EVENT_H
#define ESC_EVENT_H
typedef struct st_event event_t;
event_t* event_create();
void event_destroy(event_t* event);
void event_wait(event_t* event);
void event_set(event_t* event);
#endif //ESC_EVENT_H

21
src/utils/mutex.c
View File

@@ -16,21 +16,21 @@ struct st_mutex {
CRITICAL_SECTION handle;
};
mutex_t mutex_create() {
mutex_t mutex = malloc(sizeof(struct st_mutex));
mutex_t* mutex_create() {
mutex_t* mutex = malloc(sizeof(struct st_mutex));
InitializeCriticalSection(&mutex->handle);
return mutex;
}
void mutex_lock(mutex_t mutex) {
void mutex_lock(mutex_t* mutex) {
EnterCriticalSection(&mutex->handle);
}
void mutex_unlock(mutex_t mutex) {
void mutex_unlock(mutex_t* mutex) {
LeaveCriticalSection(&mutex->handle);
}
void mutex_destroy(mutex_t mutex) {
void mutex_destroy(mutex_t* mutex) {
DeleteCriticalSection(&mutex->handle);
}
#endif
@@ -41,22 +41,23 @@ struct st_mutex {
pthread_mutex_t handle;
};
mutex_t mutex_create() {
mutex_t mutex = malloc(sizeof(struct st_mutex));
mutex_t* mutex_create() {
mutex_t* mutex = malloc(sizeof(struct st_mutex));
pthread_mutex_init(&mutex->handle, 0);
return mutex;
}
void mutex_lock(mutex_t mutex) {
void mutex_lock(mutex_t* mutex) {
pthread_mutex_lock(&mutex->handle);
}
void mutex_unlock(mutex_t mutex) {
void mutex_unlock(mutex_t* mutex) {
pthread_mutex_unlock(&mutex->handle);
}
void mutex_destroy(mutex_t mutex) {
void mutex_destroy(mutex_t* mutex) {
pthread_mutex_destroy(&mutex->handle);
free(mutex);
}
#endif

10
src/utils/mutex.h
View File

@@ -5,11 +5,11 @@
#ifndef ESC_MUTEX_H
#define ESC_MUTEX_H
typedef struct st_mutex* mutex_t;
typedef struct st_mutex mutex_t;
mutex_t mutex_create();
void mutex_lock(mutex_t mutex);
void mutex_unlock(mutex_t mutex);
void mutex_destroy(mutex_t mutex);
mutex_t* mutex_create();
void mutex_lock(mutex_t* mutex);
void mutex_unlock(mutex_t* mutex);
void mutex_destroy(mutex_t* mutex);
#endif //ESC_MUTEX_H

43
src/utils/queue.c
View File

@@ -14,14 +14,14 @@ struct st_node {
struct st_node* next;
};
struct st_queue {
mutex_t lock;
mutex_t* lock;
size_t size;
struct st_node* start;
struct st_node* end;
};
queue_t queue_create() {
queue_t q = malloc(sizeof(struct st_queue));
queue_t* queue_create() {
queue_t* q = malloc(sizeof(struct st_queue));
assert(q != 0);
q->lock = mutex_create();
q->size = 0;
@@ -30,7 +30,7 @@ queue_t queue_create() {
return q;
}
void queue_enqueue(queue_t q, void* item) {
void queue_enqueue(queue_t* q, void* item) {
struct st_node* node = malloc(sizeof(struct st_node));
assert(node != 0);
node->item = item;
@@ -48,7 +48,7 @@ void queue_enqueue(queue_t q, void* item) {
mutex_unlock(q->lock);
}
void* queue_dequeue(queue_t q) {
void* queue_dequeue(queue_t* q) {
mutex_lock(q->lock);
if (q->start == 0) {
mutex_unlock(q->lock);
@@ -66,21 +66,21 @@ void* queue_dequeue(queue_t q) {
return item;
}
void* queue_peek(queue_t q) {
void* queue_peek(queue_t* q) {
mutex_lock(q->lock);
void* item = q->start ? q->start->item : 0;
mutex_unlock(q->lock);
return item;
}
inline size_t queue_size(queue_t q) {
inline size_t queue_size(queue_t* q) {
mutex_lock(q->lock);
size_t size = q->size;
mutex_unlock(q->lock);
return size;
}
void queue_destroy(queue_t q) {
void queue_destroy(queue_t* q) {
mutex_lock(q->lock);
struct st_node* p = q->start;
while(p) {
@@ -89,5 +89,32 @@ void queue_destroy(queue_t q) {
p = next;
}
mutex_unlock(q->lock);
mutex_destroy(q->lock);
free(q);
}
void* queue_remove(queue_t* q, find_predicate predicate, void* arg) {
void* found = 0;
mutex_lock(q->lock);
struct st_node* p = q->start;
struct st_node* prev = 0;
while(p) {
if (predicate(p->item, arg) == BOOL_TRUE) {
found = p->item;
if (prev) {
prev->next = p->next;
} else {
q->start = p->next;
}
if (q->end == p) {
q->end = prev;
}
free(p);
break;
}
prev = p;
p = p->next;
}
mutex_unlock(q->lock);
return found;
}

19
src/utils/queue.h
View File

@@ -5,13 +5,18 @@
#ifndef ESC_QUEUE_H
#define ESC_QUEUE_H
typedef struct st_queue* queue_t;
#include "bool.h"
queue_t queue_create();
void queue_enqueue(queue_t q, void* item);
void* queue_dequeue(queue_t q);
void* queue_peek(queue_t q);
size_t queue_size(queue_t q);
void queue_destroy(queue_t q);
typedef struct st_queue queue_t;
queue_t* queue_create();
void queue_destroy(queue_t* q);
void queue_enqueue(queue_t* q, void* item);
void* queue_dequeue(queue_t* q);
void* queue_peek(queue_t* q);
size_t queue_size(queue_t* q);
typedef bool_t (*find_predicate)(void* item, void* arg);
void* queue_remove(queue_t* q, find_predicate, void* arg);
#endif //ESC_QUEUE_H

41
src/utils/socket.c
View File

@@ -17,9 +17,9 @@ struct st_socket {
int last_error;
};
socket_t socket_create(int type) {
socket_t* socket_create(int type) {
if (type == SOCKET_TYPE_TCP) {
socket_t s = malloc(sizeof(struct st_socket));
socket_t* s = malloc(sizeof(struct st_socket));
s->af = AF_INET;
s->type = SOCK_STREAM;
s->proto = IPPROTO_TCP;
@@ -32,7 +32,7 @@ socket_t socket_create(int type) {
return NULL;
}
int socket_close(socket_t s) {
int socket_close(socket_t* s) {
int rc = closesocket(s->s);
if (rc != 0) {
s->last_error = WSAGetLastError();
@@ -40,7 +40,7 @@ int socket_close(socket_t s) {
return rc;
}
int socket_connect(socket_t s, char* addr, unsigned short port) {
int socket_connect(socket_t* s, char* addr, unsigned short port) {
char port_s[6];
ADDRINFO hints;
memset(&hints, 0, sizeof(ADDRINFO));
@@ -60,7 +60,7 @@ int socket_connect(socket_t s, char* addr, unsigned short port) {
return 0;
}
ssize_t socket_send(socket_t s, char* data, size_t len) {
ssize_t socket_send(socket_t* s, char* data, size_t len) {
ssize_t rc = send(s->s, data, (int)len, 0);
if (rc < 0) {
s->last_error = WSAGetLastError();
@@ -68,7 +68,7 @@ ssize_t socket_send(socket_t s, char* data, size_t len) {
return rc;
}
ssize_t socket_recv(socket_t s, char* buf, size_t len) {
ssize_t socket_recv(socket_t* s, char* buf, size_t len) {
ssize_t rc = recv(s->s, buf, (int)len, 0);
if (rc < 0) {
s->last_error = WSAGetLastError();
@@ -76,7 +76,7 @@ ssize_t socket_recv(socket_t s, char* buf, size_t len) {
return rc;
}
int socket_readable(socket_t s) {
int socket_readable(socket_t* s) {
fd_set readable;
FD_ZERO(&readable);
FD_SET(s->s, &readable);
@@ -89,7 +89,7 @@ int socket_readable(socket_t s) {
return rc;
}
int socket_writable(socket_t s) {
int socket_writable(socket_t* s) {
fd_set writable;
FD_ZERO(&writable);
FD_SET(s->s, &writable);
@@ -102,7 +102,7 @@ int socket_writable(socket_t s) {
return rc;
}
int socket_error(socket_t s) {
int socket_error(socket_t* s) {
return s->last_error;
}
#endif
@@ -122,9 +122,9 @@ struct st_socket {
int last_error;
};
socket_t socket_create(int type) {
socket_t* socket_create(int type) {
if (type == SOCKET_TYPE_TCP) {
socket_t s = malloc(sizeof(struct st_socket));
socket_t* s = malloc(sizeof(struct st_socket));
s->af = AF_INET;
s->type = SOCK_STREAM;
s->proto = IPPROTO_TCP;
@@ -137,12 +137,12 @@ socket_t socket_create(int type) {
return NULL;
}
inline void socket_destroy(socket_t s) {
inline void socket_destroy(socket_t* s) {
socket_close(s);
free(s);
}
int socket_close(socket_t s) {
int socket_close(socket_t* s) {
if (close(s->s) == -1) {
s->last_error = errno;
return -1;
@@ -150,7 +150,7 @@ int socket_close(socket_t s) {
return 0;
}
int socket_connect(socket_t s, char* addr, unsigned short port) {
int socket_connect(socket_t* s, char* addr, unsigned short port) {
struct addrinfo hints;
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = s->af;
@@ -161,17 +161,20 @@ int socket_connect(socket_t s, char* addr, unsigned short port) {
snprintf(port_s, 6, "%d", port);
int rc;
if ((rc = getaddrinfo(addr, port_s, &hints, &result)) != 0) {
if (result) freeaddrinfo(result);
s->last_error = rc;
return -1;
}
if (connect(s->s, result->ai_addr, (int)result->ai_addrlen) != 0) {
freeaddrinfo(result);
s->last_error = errno;
return -1;
}
freeaddrinfo(result);
return 0;
}
ssize_t socket_send(socket_t s, char* data, size_t len) {
ssize_t socket_send(socket_t* s, char* data, size_t len) {
ssize_t rc = send(s->s, data, len, 0);
if (rc < 0) {
s->last_error = errno;
@@ -179,7 +182,7 @@ ssize_t socket_send(socket_t s, char* data, size_t len) {
return rc;
}
ssize_t socket_recv(socket_t s, char* buf, size_t len) {
ssize_t socket_recv(socket_t* s, char* buf, size_t len) {
ssize_t rc = recv(s->s, buf, len, 0);
if (rc < 0) {
s->last_error = errno;
@@ -187,7 +190,7 @@ ssize_t socket_recv(socket_t s, char* buf, size_t len) {
return rc;
}
int socket_writable(socket_t s) {
int socket_writable(socket_t* s) {
fd_set writable;
FD_ZERO(&writable);
FD_SET(s->s, &writable);
@@ -200,7 +203,7 @@ int socket_writable(socket_t s) {
return rc;
}
int socket_readable(socket_t s) {
int socket_readable(socket_t* s) {
fd_set readable;
FD_ZERO(&readable);
FD_SET(s->s, &readable);
@@ -213,7 +216,7 @@ int socket_readable(socket_t s) {
return rc;
}
inline int socket_error(socket_t s) {
inline int socket_error(socket_t* s) {
return s->last_error;
}
#endif

20
src/utils/socket.h
View File

@@ -6,21 +6,21 @@
#define ESC_SOCKET_H
struct st_socket;
typedef struct st_socket* socket_t;
typedef struct st_socket socket_t;
enum {
SOCKET_TYPE_TCP = 1,
SOCKET_TYPE_UDP = 2
};
socket_t socket_create(int type);
void socket_destroy(socket_t s);
int socket_close(socket_t s);
int socket_connect(socket_t s, char* addr, unsigned short port);
ssize_t socket_send(socket_t s, char* data, size_t len);
ssize_t socket_recv(socket_t s, char* buf, size_t len);
int socket_writable(socket_t s);
int socket_readable(socket_t s);
int socket_error(socket_t s);
socket_t* socket_create(int type);
void socket_destroy(socket_t* s);
int socket_close(socket_t* s);
int socket_connect(socket_t* s, char* addr, unsigned short port);
ssize_t socket_send(socket_t* s, char* data, size_t len);
ssize_t socket_recv(socket_t* s, char* buf, size_t len);
int socket_writable(socket_t* s);
int socket_readable(socket_t* s);
int socket_error(socket_t* s);
#endif //ESC_SOCKET_H

26
src/utils/thread.c
View File

@@ -17,8 +17,8 @@ struct st_thread {
int last_error;
};
thread_t thread_create(thread_start_t thread_start, thread_arg_t thread_arg) {
thread_t thread = malloc(sizeof(struct st_thread));
thread_t* thread_create(thread_start_t thread_start, thread_arg_t thread_arg) {
thread_t* thread = malloc(sizeof(struct st_thread));
thread->handle = 0;
thread->start = thread_start;
thread->arg = thread_arg;
@@ -26,7 +26,7 @@ thread_t thread_create(thread_start_t thread_start, thread_arg_t thread_arg) {
return thread;
}
int thread_start(thread_t thread) {
int thread_start(thread_t* thread) {
thread->handle = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)thread->start, thread->arg, 0, &thread->id);
if (thread->handle == NULL) {
DWORD rc = GetLastError();
@@ -36,7 +36,7 @@ int thread_start(thread_t thread) {
return 0;
}
int thread_wait(thread_t thread) {
int thread_wait(thread_t* thread) {
if (WaitForSingleObject(thread->handle, INFINITE) == WAIT_FAILED) {
DWORD rc = GetLastError();
thread->last_error = (int)rc;
@@ -45,13 +45,13 @@ int thread_wait(thread_t thread) {
return 0;
}
int thread_destroy(thread_t thread) {
int thread_destroy(thread_t* thread) {
if (thread_wait(thread) != 0) SuspendThread(thread->handle);
CloseHandle(thread->handle);
free(thread);
}
int thread_error(thread_t thread) {
int thread_error(thread_t* thread) {
return thread->last_error;
}
@@ -68,8 +68,8 @@ struct st_thread {
int joined;
};
thread_t thread_create(thread_start_t thread_start, thread_arg_t thread_arg) {
thread_t thread = malloc(sizeof(struct st_thread));
thread_t* thread_create(thread_start_t thread_start, thread_arg_t thread_arg) {
thread_t* thread = malloc(sizeof(struct st_thread));
thread->handle = 0;
thread->start = thread_start;
thread->arg = thread_arg;
@@ -79,7 +79,7 @@ thread_t thread_create(thread_start_t thread_start, thread_arg_t thread_arg) {
return thread;
}
int thread_start(thread_t thread) {
int thread_start(thread_t* thread) {
int rc = pthread_create(&thread->handle, NULL, thread->start, thread->arg);
if (rc != 0) {
thread->last_error = rc;
@@ -88,7 +88,7 @@ int thread_start(thread_t thread) {
return 0;
}
int thread_wait(thread_t thread) {
int thread_wait(thread_t* thread) {
if (thread->joined) {
return 0;
}
@@ -102,7 +102,7 @@ int thread_wait(thread_t thread) {
return 0;
}
int thread_destroy(thread_t thread) {
int thread_destroy(thread_t* thread) {
if (thread_wait(thread)) {
int rc = pthread_cancel(thread->handle);
if (rc != 0) {
@@ -114,11 +114,11 @@ int thread_destroy(thread_t thread) {
return 0;
}
inline int thread_error(thread_t thread) {
inline int thread_error(thread_t* thread) {
return thread->last_error;
}
inline int thread_result(thread_t thread, void** result) {
inline int thread_result(thread_t* thread, void** result) {
if (thread_wait(thread)) {
return -1;
}

14
src/utils/thread.h
View File

@@ -7,13 +7,13 @@
typedef void* (*thread_start_t)(void*);
typedef void* thread_arg_t;
typedef struct st_thread* thread_t;
typedef struct st_thread thread_t;
thread_t thread_create(thread_start_t thread_start, thread_arg_t thread_arg);
int thread_start(thread_t thread);
int thread_wait(thread_t thread);
int thread_destroy(thread_t thread);
int thread_error(thread_t thread);
int thread_result(thread_t thread, void** result);
thread_t* thread_create(thread_start_t thread_start, thread_arg_t thread_arg);
int thread_start(thread_t* thread);
int thread_wait(thread_t* thread);
int thread_destroy(thread_t* thread);
int thread_error(thread_t* thread);
int thread_result(thread_t* thread, void** result);
#endif //ESC_THREAD_H

18
src/utils/uuid.c
View File

@@ -18,39 +18,39 @@ struct st_uuid {
};
//TODO: fixme - this is not secure, use crypto random bytes instead of rand
esc_uuid_t esc_uuid_create() {
uuid_t* uuid_create() {
if (srand_called == 0) {
srand(time(NULL));
srand_called = 1;
}
esc_uuid_t result = malloc(sizeof(struct st_uuid));
uuid_t* result = malloc(sizeof(struct st_uuid));
for(int i=0;i<UUID_SIZE;i++)
result->data[i] = (uint8_t)(rand() & 0xff);
return result;
}
inline void esc_uuid_destroy(esc_uuid_t uuid) {
inline void uuid_destroy(uuid_t* uuid) {
free(uuid);
}
esc_uuid_t esc_uuid_from(const uint8_t* src, size_t size) {
uuid_t* uuid_from(const uint8_t *src, size_t size) {
if (size < UUID_SIZE) {
return 0;
}
esc_uuid_t uuid = malloc(sizeof(struct st_uuid));
uuid_t* uuid = malloc(sizeof(struct st_uuid));
memcpy(uuid->data, src, UUID_SIZE);
return uuid;
}
esc_uuid_t esc_uuid_copy(esc_uuid_t other) {
return esc_uuid_from((uint8_t*)other->data, UUID_SIZE);
uuid_t* uuid_copy(uuid_t* other) {
return uuid_from((uint8_t *) other->data, UUID_SIZE);
}
int esc_uuid_compare(esc_uuid_t left, esc_uuid_t right) {
int uuid_compare(uuid_t* left, uuid_t* right) {
return memcmp(left->data, right->data, UUID_SIZE);
}
const char* esc_uuid_format(esc_uuid_t uuid, char* buffer, size_t buf_size) {
const char* uuid_format(uuid_t* uuid, char *buffer, size_t buf_size) {
#ifdef _WIN32
snprintf(buffer, buf_size, "%08lx-%04hx-%04hx-%04hx-%08lx%04hx",
*(long*)&uuid->data[0],

17
src/utils/uuid.h
View File

@@ -5,15 +5,16 @@
#ifndef ESC_UUID_H
#define ESC_UUID_H
#include <stdint.h>
#include <stdint.h> //uint8_t
#include <stddef.h> //size_t
typedef struct st_uuid* esc_uuid_t;
typedef struct st_uuid uuid_t;
esc_uuid_t esc_uuid_create();
void esc_uuid_destroy(esc_uuid_t uuid);
esc_uuid_t esc_uuid_from(const uint8_t* src, size_t size);
esc_uuid_t esc_uuid_copy(esc_uuid_t other);
int esc_uuid_compare(esc_uuid_t left, esc_uuid_t right);
const char* esc_uuid_format(esc_uuid_t uuid, char* buffer, size_t len);
uuid_t* uuid_create();
void uuid_destroy(uuid_t* uuid);
uuid_t* uuid_from(const uint8_t *src, size_t size);
uuid_t* uuid_copy(uuid_t* other);
int uuid_compare(uuid_t* left, uuid_t* right);
const char* uuid_format(uuid_t* uuid, char *buffer, size_t len);
#endif //ESC_UUID_H

89
test/main.c Normal file
View File

@@ -0,0 +1,89 @@
#include <stdio.h>
#include "../src/utils/debug.h"
#include "../include/esc.h"
#ifdef _WIN32
#include <winsock.h>
#include <windows.h>
#define usleep Sleep
#define sleep(x) Sleep(x*1000)
#endif
#ifdef __linux__
#include <unistd.h>
#endif
int main() {
#ifdef _WIN32
WSADATA wsaData;
WSAStartup(MAKEWORD(2,0), &wsaData);
#endif
esc_connection_t* conn = esc_connection_create(esc_default_connection_settings, "tcp://127.0.0.1:1113", NULL);
if (esc_connection_connect(conn) != 0) {
error_t* err = esc_connection_last_error(conn);
fprintf(stderr, "Error: %s code=%d file=%s line=%d\n", err->message, err->code, err->file, err->line);
return -2;
}
uuid_t *event_id = uuid_create();
buffer_t *data = buffer_from_string("{\"a\":\"1\"}");
esc_event_data_t* event_data = esc_event_data_create(event_id, "test_event", BOOL_TRUE, data, 0);
array_t* events = array_create(1, event_data);
esc_write_result_t* write_result = esc_append_to_stream(conn, "test-123", ESC_VERSION_ANY, events);
if (write_result == 0) {
error_t* err = esc_connection_last_error(conn);
fprintf(stderr, "Error: %s code=%d file=%s line=%d\n", err->message, err->code, err->file, err->line);
return -3;
}
esc_write_result_destroy(write_result);
array_destroy(events, (array_deallocator)esc_event_data_destroy);
uuid_destroy(event_id);
buffer_destroy(data);
esc_credentials_t* credentials = esc_credentials_create("admin", "changeit");
esc_all_events_slice_t* result = 0;
do {
esc_all_events_slice_t* old_result = result;
result = esc_connection_read_all_forward(conn, old_result ? old_result->next_position : NULL, 100, credentials);
if (old_result) esc_all_events_slice_destroy(old_result);
if (result == 0) {
error_t* err = esc_connection_last_error(conn);
fprintf(stderr, "Error: %s code=%d file=%s line=%d\n", err->message, err->code, err->file, err->line);
break;
}
char posbuf1[44];
char posbuf2[44];
printf("%s %s %s %u\n", result->read_direction,
esc_position_format(result->from_position, posbuf1, 44),
esc_position_format(result->next_position, posbuf2, 44),
result->is_end_of_stream);
char uuid_buf[37];
for (size_t i = 0; i < result->n_events; i++) {
#ifdef _WIN32
printf("%s %s %lld@%s %llu %llu\n", esc_uuid_format(result->events[i]->event->event_id, uuid_buf, 37),
#else
printf("%s %s %ld@%s %lu %lu\n", uuid_format(result->events[i]->event->event_id, uuid_buf, 37),
#endif
result->events[i]->event->event_type,
result->events[i]->event->event_number,
result->events[i]->event->event_stream_id,
buffer_size(result->events[i]->event->data),
buffer_size(result->events[i]->event->metadata));
}
} while(result->is_end_of_stream == 0);
if (result) esc_all_events_slice_destroy(result);
sleep(5);
esc_connection_destroy(conn);
esc_credentials_destroy(credentials);
dbg_list_allocs();
#ifdef _WIN32
WSACleanup();
#endif
return 0;
}