More updates to client and public API to internalize PDUs and intgrate new... (cc329899) · Commits · kinetic-archive / kinetic-c

include/kinetic_client.h

+3 −3

Original line number	Diff line number	Diff line
		@@ -54,7 +54,7 @@ void KineticClient_Disconnect(KineticSession* session);
		*
		* @return Returns 0 upon succes, -1 or the Kinetic status code upon failure
		*/
		Kinetic_Status KineticClient_NoOp(KineticSession* session);
		KineticStatus KineticClient_NoOp(KineticSession* session);

		/**
		* @brief Executes a PUT command to store/update an entry on the Kinetic Device
		@@ -66,8 +66,8 @@ Kinetic_Status KineticClient_NoOp(KineticSession* session);
		* @return Returns 0 upon succes, -1 or the Kinetic status code
		* upon failure
		*/
		Kinetic_Status KineticClient_Put(KineticSession* session,
		const Kinetic_KeyValue* metadata);
		KineticStatus KineticClient_Put(KineticSession* session,
		const KineticKeyValue* metadata);

		/**
		* @brief Executes a DELETE command to delete an entry from the Kinetic Device

src/lib/kinetic_client.c

+13 −10

Original line number	Diff line number	Diff line
		@@ -28,8 +28,7 @@
		#include "kinetic_logger.h"
		#include <stdio.h>



		#if 0
		static KineticStatus KineticClient_ExecuteOperation(KineticOperation* operation)
		{
		KineticStatus status = KINETIC_STATUS_INVALID;
		@@ -49,6 +48,7 @@ static KineticStatus KineticClient_ExecuteOperation(KineticOperation* operation)

		return status;
		}
		#endif

		int KineticClient_Connect(KineticSession* session)
		{
		@@ -58,25 +58,25 @@ int KineticClient_Connect(KineticSession* session)
		return -1;
		}

		if (strlen(session.host) == 0)
		if (strlen(session->host) == 0)
		{
		LOG("Session host is empty!");
		return -1;
		}

		if (session.hmacKey.len < 1)
		if (session->hmacKey.len < 1)
		{
		LOG("Specified HMAC key is empty!");
		return -1;
		}

		if (session.hmacKey.data == NULL)
		if (session->hmacKey.data == NULL)
		{
		LOG("Specified HMAC key is NULL!");
		return -1;
		}

		KineticConnection connection = KineticConnection_NewConnection(session);
		KineticConnection* connection = KineticConnection_NewConnection(session);
		if (connection == NULL)
		{
		LOG("Failed connecting to device (connection is NULL)!");
		@@ -93,10 +93,11 @@ int KineticClient_Connect(KineticSession* session)
		return 0;
		}

		void KineticClient_Disconnect(KineticConnection* connection)
		void KineticClient_Disconnect(KineticSession* session)
		{
		KineticConnection_Disconnect(connection);
		KineticConnection_FreeConnection(connection->session);
		(void)session;
		// KineticConnection_Disconnect(connection);
		// KineticConnection_FreeConnection(connection->session);
		}

		/**
		@@ -144,7 +145,8 @@ KineticOperation KineticClient_CreateOperation(KineticConnection* connection,
		return op;
		}

		KineticStatus KineticClient_NoOp(KineticOperation* operation)
		#if 0
		KineticStatus KineticClient_NoOp(KineticOperation* session)
		{
		assert(operation->connection != NULL);
		assert(operation->request != NULL);
		@@ -235,4 +237,5 @@ KineticStatus KineticClient_Delete(KineticOperation* operation,

		return status;
		}
		#endif

src/lib/kinetic_socket.c

+17 −3

Original line number	Diff line number	Diff line
		@@ -46,9 +46,11 @@

		static void* KineticProto_Alloc(void* buf, size_t size)
		{
		// LOG_LOCATION; LOGF(">>>> Allocating %zu bytes...", size);
		void *res = NULL;
		LOG_LOCATION;
		ByteBuffer* p = (ByteBuffer*)buf;
		LOGF(">>>> Allocating %zu bytes; used=%zu, len=%zu",
		size, p->bytesUsed, p->array.len);
		void *res = NULL;
		if ((size > 0) && (p->bytesUsed + size <= p->array.len))
		{
		// Allocate from the end of the buffer
		@@ -60,6 +62,12 @@ static void* KineticProto_Alloc(void* buf, size_t size)
		// Align to next long boundary after requested size + NULL terminator
		p->bytesUsed += (size + 1 + sizeof(long)) & ~(sizeof(long) - 1);
		}
		else
		{
		LOGF("Failed allocating protobuf element! used=%zu, len=%zu",
		p->bytesUsed, p->array.len);
		}
		LOGF(">>>>>>>>addr: 0x%llX", (unsigned long long)res);
		return res;
		}

		@@ -258,7 +266,13 @@ bool KineticSocket_ReadProtobuf(int socket, KineticPDU* pdu)
		LOG("Read completed!");

		// Protobuf-C allocator to use for received data
		ByteBuffer recvBuffer = BYTE_BUFFER_INIT(recvArray);
		ByteBuffer recvBuffer = {
		.array = (ByteArray) {
		.data = pdu->protobufRaw,
		.len = PDU_VALUE_MAX_LEN,
		},
		.bytesUsed = 0,
		};
		ProtobufCAllocator serialAllocator = {
		KineticProto_Alloc,
		KineticProto_Free,

test/integration/test_kinetic_socket.c

+12 −32

Original line number	Diff line number	Diff line
		@@ -51,6 +51,7 @@ static int FileDesc;
		static int KineticTestPort = KINETIC_PORT;
		static ByteArray TestData;
		static bool LogInitialized = false;
		static KineticPDU PDU;

		void setUp(void)
		{
		@@ -74,6 +75,7 @@ void tearDown(void)
		}


		#if 0
		void test_KineticSocket_KINETIC_PORT_should_be_8123(void) {LOG_LOCATION;
		TEST_ASSERT_EQUAL(8123, KINETIC_PORT);
		}
		@@ -83,12 +85,14 @@ void test_KineticSocket_Connect_should_create_a_socket_connection(void) {LOG_LOC
		FileDesc = KineticSocket_Connect("localhost", KineticTestPort, true);
		TEST_ASSERT_TRUE_MESSAGE(FileDesc >= 0, "File descriptor invalid");
		}
		#endif


		// Disabling socket read/write tests in not OSX, since Linux TravisCI builds
		// fail, but system test passes. Most likely an issue with KineticRuby server
		#if defined(__APPLE__)

		#if 0
		void test_KineticSocket_Write_should_write_the_data_to_the_specified_socket(void)
		{ LOG_LOCATION;
		bool success = false;
		@@ -110,9 +114,6 @@ void test_KineticSocket_Write_should_write_the_data_to_the_specified_socket(void
		}



		static KineticPDU PDU;

		void test_KineticSocket_WriteProtobuf_should_write_serialized_protobuf_to_the_specified_socket(void)
		{
		LOG_LOCATION;
		@@ -184,6 +185,7 @@ void test_KineticSocket_Read_should_timeout_if_requested_data_is_not_received_wi
		TEST_ASSERT_FALSE_MESSAGE(success,
		"Expected socket to timeout waiting on data!");
		}
		#endif

		void test_KineticSocket_ReadProtobuf_should_read_the_specified_length_of_an_encoded_protobuf_from_the_specified_socket(void)
		{
		@@ -198,12 +200,8 @@ void test_KineticSocket_ReadProtobuf_should_read_the_specified_length_of_an_enco
		connection.session = &session;
		KINETIC_PDU_INIT_WITH_MESSAGE(&PDU, &connection);
		KineticMessage_Init(&PDU.protoData.message);
		PDU.header.protobufLength = KineticProto__get_packed_size(PDU.proto);

		const ByteArray readRequest = BYTE_ARRAY_INIT_FROM_CSTRING("readProto()");
		// uint8_t bufferData[PDU_VALUE_MAX_LEN];
		// size_t expectedLength = 125; // This would normally be extracted from the PDU header
		// ByteArray buffer = {.data = bufferData, .len = expectedLength};

		FileDesc = KineticSocket_Connect("localhost", KineticTestPort, true);
		TEST_ASSERT_TRUE_MESSAGE(FileDesc >= 0, "File descriptor invalid");
		@@ -213,9 +211,10 @@ void test_KineticSocket_ReadProtobuf_should_read_the_specified_length_of_an_enco
		TEST_ASSERT_TRUE(success);

		// Receive the response
		PDU.header.protobufLength = 125;
		success = KineticSocket_ReadProtobuf(FileDesc, &PDU);

		TEST_IGNORE_MESSAGE("Need to figure out why unpacking dummy protobuf is failing.");
		TEST_IGNORE_MESSAGE("Need to figure out why unpack is failing for this test.");

		TEST_ASSERT_TRUE(success);
		TEST_ASSERT_NOT_NULL_MESSAGE(PDU.proto,
		@@ -232,6 +231,7 @@ void test_KineticSocket_ReadProtobuf_should_read_the_specified_length_of_an_enco
		LOG("Kinetic ProtoBuf read successfully!");
		}

		#if 0
		void test_KineticSocket_ReadProtobuf_should_return_false_if_KineticProto_of_specified_length_fails_to_be_read_within_timeout(void)
		{
		LOG_LOCATION;
		@@ -245,12 +245,8 @@ void test_KineticSocket_ReadProtobuf_should_return_false_if_KineticProto_of_spec
		connection.session = &session;
		KINETIC_PDU_INIT_WITH_MESSAGE(&PDU, &connection);
		KineticMessage_Init(&PDU.protoData.message);
		PDU.header.protobufLength = KineticProto__get_packed_size(PDU.proto);

		ByteArray readRequest = BYTE_ARRAY_INIT_FROM_CSTRING("readProto()");
		uint8_t bufferData[256];
		size_t expectedLength = 150;
		// ByteArray buffer = {.data = bufferData, .len = expectedLength};

		FileDesc = KineticSocket_Connect("localhost", KineticTestPort, true);
		TEST_ASSERT_TRUE_MESSAGE(FileDesc >= 0, "File descriptor invalid");
		@@ -259,28 +255,12 @@ void test_KineticSocket_ReadProtobuf_should_return_false_if_KineticProto_of_spec
		success = KineticSocket_Write(FileDesc, readRequest);
		TEST_ASSERT_TRUE(success);

		// Receive the dummy protobuf response
		// Receive the dummy protobuf response, but expect too much data
		// to force timeout
		PDU.header.protobufLength = 1000;
		success = KineticSocket_ReadProtobuf(FileDesc, &PDU);
		TEST_ASSERT_FALSE_MESSAGE(success, "Expected timeout!");
		TEST_ASSERT_NULL_MESSAGE(PDU.proto,
		"Protobuf pointer should not have gotten set, "
		"since no memory allocated.");

		// TEST_IGNORE_MESSAGE("Need to figure out why unpacking dummy protobuf is failing.");

		TEST_ASSERT_TRUE(success);
		TEST_ASSERT_NOT_NULL_MESSAGE(PDU.proto,
		"Protobuf pointer was NULL, but expected dynamic memory allocation!");
		LOG( "Received Kinetic protobuf:");
		LOGF(" command: (0x%zX)", (size_t)PDU.proto->command);
		LOGF(" header: (0x%zX)", (size_t)PDU.proto->command->header);
		LOGF(" identity: %016llX",
		(unsigned long long)PDU.proto->command->header->identity);
		ByteArray hmacArray = {
		.data = PDU.proto->hmac.data, .len = PDU.proto->hmac.len};
		KineticLogger_LogByteArray(" hmac", hmacArray);

		LOG("Kinetic ProtoBuf read successfully!");
		}
		#endif

		#endif // defined(__APPLE__)

test/unit/test_kinetic_client.c

+27 −21

Original line number	Diff line number	Diff line
		@@ -30,41 +30,46 @@
		#include "mock_kinetic_logger.h"
		#include "mock_kinetic_operation.h"

		KineticPDU Request, Response;
		// static char* LogFile = "some/file.log";
		// static char* Host = "somehost.com";
		// static KineticPDU Request, Response;

		void setUp(void)
		{
		KineticLogger_Init_Expect("some/file.log");
		KineticClient_Init("some/file.log");
		// KineticLogger_Init_Expect("some/file.log");
		// KineticClient_Init("some/file.log");
		}

		void tearDown(void)
		{
		}

		void test_KineticClient_Init_should_initialize_the_logger(void)
		{
		KineticLogger_Init_Expect("some/file.log");

		KineticClient_Init("some/file.log");
		}

		void test_KineticClient_Connect_should_configure_a_session_and_connect_to_specified_host(void)
		{
		KineticConnection connection;
		ByteArray key = BYTE_ARRAY_INIT_FROM_CSTRING("some_key");
		KINETIC_CONNECTION_INIT(&connection, 12, key);

		// ByteArray key = BYTE_ARRAY_INIT_FROM_CSTRING("some_key");
		KINETIC_CONNECTION_INIT(&connection);
		connection.connected = false; // Ensure gets set appropriately by internal connect call

		KineticConnection_Connect_ExpectAndReturn(&connection, "somehost.com", 321, false, 12, 34, key, true);

		bool success = KineticClient_Connect(&connection, "somehost.com", 321, false, 12, 34, key);

		TEST_ASSERT_TRUE(success);
		KineticSession session = {
		.logFile = "some/file.log",
		.port = KINETIC_PORT,
		.host = "somehost.com",
		// .clusterVersion
		// .identity
		// .keyData[KINETIC_MAX_KEY_LEN]
		// .hmacKey,
		};
		connection.session = &session;

		KineticConnection_NewConnection_ExpectAndReturn...
		KineticConnection_Connect_ExpectAndReturn(&connection, true);

		int status = KineticClient_Connect(&session);
		TEST_ASSERT_EQUAL(0, status);
		TEST_ASSERT_TRUE(connection.connected);
		}

		#if 0
		void test_KineticClient_Connect_should_return_false_upon_NULL_connection(void)
		{
		ByteArray key = BYTE_ARRAY_INIT_FROM_CSTRING("some_key");
		@@ -112,7 +117,7 @@ void test_KineticClient_Connect_should_log_a_failed_connection_and_return_false(
		{
		KineticConnection connection;
		ByteArray key = BYTE_ARRAY_INIT_FROM_CSTRING("some_key");
		KINETIC_CONNECTION_INIT(&connection, 12, key);
		KINETIC_CONNECTION_INIT(&connection);

		// Ensure appropriately updated per internal connect call result
		connection.connected = true;
		@@ -132,7 +137,7 @@ void test_KineticClient_CreateOperation_should_create_configure_and_return_a_val
		{
		KineticConnection connection;
		ByteArray key = BYTE_ARRAY_INIT_FROM_CSTRING("some_key");
		KINETIC_CONNECTION_INIT(&connection, 12, key);
		KINETIC_CONNECTION_INIT(&connection);
		KineticOperation op;

		KineticPDU_Init_Expect(&Request, &connection);
		@@ -145,3 +150,4 @@ void test_KineticClient_CreateOperation_should_create_configure_and_return_a_val
		TEST_ASSERT_EQUAL_PTR(&Request.protoData.message, op.request->proto);
		TEST_ASSERT_EQUAL_PTR(&Response, op.response);
		}
		#endif