Commit f34f9ed9 authored by Greg Williams's avatar Greg Williams
Browse files

Flushed out some details for sending/receiving with KineticConnection in prep... 

Flushed out some details for sending/receiving with KineticConnection in prep for socket comm integration.
parent b0829d24

src/lib/kinetic_api.c

+10 −12
Original line number Diff line number Diff line
@@ -19,7 +19,7 @@ 
*/



#include "kinetic_api.h"

#include "kinetic_connction.h"

#include "kinetic_connection.h"

#include "kinetic_pdu.h"

#include "kinetic_logger.h"

#include <stdio.h>
@@ -49,29 +49,27 @@ const KineticConnection KineticApi_Connect(const char* host, int port, bool bloc 
	return connection;

}



KineticProto_Status_StatusCode KineticApi_SendNoop(

    KineticPDU* const request, KineticPDU* const response,

    uint8_t* const requestBuffer, uint8_t* const responseBuffer)

KineticProto_Status_StatusCode KineticApi_SendNoop(KineticPDU* request, KineticPDU* response)

{

    KineticProto_Status_StatusCode status =

        KINETIC_PROTO_STATUS_STATUS_CODE_INVALID_STATUS_CODE;



    // Initialize request

    KineticExchange_Init(request->exchange, 1234, 5678, connection);

    KineticMessage_Init(request->message);

    KineticMessage_BuildNoop(request->message);

    KineticExchange_Init(request->exchange, 1234, 5678, request->exchange->connection);

    KineticMessage_Init(request->protobuf);

    KineticMessage_BuildNoop(request->protobuf);



    KineticPDU_Init(request, request->exchange, requestBuffer, request->message, NULL, 0);

    KineticPDU_Init(request, request->exchange, request->protobuf, NULL, 0);



    // Send the request

    KineticConnection_SendPDU(request);



    // Associate response with same exchange as request

    response->exchange = request->exchange;

    if (KineticConnection_ReceivePDU(connection, response))

    KineticPDU_Init(response, response->exchange, response->protobuf, NULL, 0);

    if (KineticConnection_ReceivePDU(response))

    {

        status = KINETIC_PROTO_STATUS_STATUS_CODE_SUCCESS;

        status = response->protobuf->command.status->code;

    }



	return KINETIC_PROTO_STATUS_STATUS_CODE_NOT_ATTEMPTED;

	return status;

}

src/lib/kinetic_api.h

+1 −3
Original line number Diff line number Diff line
@@ -26,8 +26,6 @@ 


void KineticApi_Init(const char* log_file);

const KineticConnection KineticApi_Connect(const char* host, int port, bool blocking);

KineticProto_Status_StatusCode KineticApi_SendNoop(

    KineticPDU* const request, KineticPDU* const response,

    uint8_t* const requestBuffer, uint8_t* const responseBuffer);

KineticProto_Status_StatusCode KineticApi_SendNoop(KineticPDU* request, KineticPDU* response);



#endif // _KINETIC_API_H

src/lib/kinetic_connection.c

+2 −2
Original line number Diff line number Diff line
@@ -52,12 +52,12 @@ bool KineticConnection_SendPDU(KineticPDU* const request) 
{

    // .... NEED TO SEND THE MESSAGE STILL!!!!!!!



    return (request->message->status.code == KINETIC_PROTO_STATUS_STATUS_CODE_SUCCESS);

    return (request->protobuf->status.code == KINETIC_PROTO_STATUS_STATUS_CODE_SUCCESS);

}



bool KineticConnection_ReceivePDU(KineticPDU* const response)

{

    // KineticPDU_Init(&PDUOut, (uint8_t*)0x12345678, &MessageOut.proto, (uint8_t*)0xDEADBEEF, 789);



    return (response->message->status.code == KINETIC_PROTO_STATUS_STATUS_CODE_SUCCESS);

    return (response->protobuf->status.code == KINETIC_PROTO_STATUS_STATUS_CODE_SUCCESS);

}

src/lib/kinetic_message.c

+1 −1
Original line number Diff line number Diff line
@@ -40,5 +40,5 @@ void KineticMessage_Init(KineticMessage* const message) 


void KineticMessage_BuildNoop(KineticMessage* const message)

{

    assert(false); // Need to complete!!!

    // assert(false); // Need to complete!!!

}

src/lib/kinetic_pdu.c

+21 −39
Original line number Diff line number Diff line
@@ -21,23 +21,21 @@ 
#include "kinetic_pdu.h"

#include <string.h>



static void KineticPDU_PackInt64(uint8_t* const buffer, int64_t value);

static int64_t KineticPDU_UnpackInt64(const uint8_t* const buffer);

static void KineticPDU_PackInt32(uint8_t* const buffer, int32_t value);

static int32_t KineticPDU_UnpackInt32(const uint8_t* const buffer);



void KineticPDU_Init(

    KineticPDU* const pdu,

    KineticExchange* const exchange,

    uint8_t* const buffer,

    KineticMessage* const message,

    const uint8_t* const value,

    int64_t valueLength)

    uint8_t* const value,

    int32_t valueLength)

{

    size_t protoPackedSize;



    assert(pdu != NULL);

    assert(message != NULL);

    assert(exchange != NULL);

    assert(buffer != NULL);



    // Initialize the instance struct

    memset(pdu, 0, sizeof(KineticPDU));
@@ -45,56 +43,40 @@ void KineticPDU_Init( 
    // Associate with the specified exchange

    pdu->exchange = exchange;



    // Configure the protobuf header with exchange info

    KineticExchange_ConfigureHeader(exchange, &message->header);

    // Associate with the specified message

    pdu->protobuf = message;



    // Assign the data buffer pointer

    pdu->data = buffer;

    // Configure the protobuf header with exchange info

    KineticExchange_ConfigureHeader(exchange, &pdu->protobuf->header);



    // Setup the PDU header fields

    pdu->prefix = &buffer[0];

    *pdu->prefix = (uint8_t)'F';

    pdu->protoLength = &buffer[1];

    protoPackedSize = KineticProto_get_packed_size(&message->proto);

    KineticPDU_PackInt64(pdu->protoLength, protoPackedSize); // Pack protobuf length field

    pdu->valueLength = &buffer[1 + sizeof(int64_t)];

    KineticPDU_PackInt64(pdu->valueLength, valueLength); // Pack value length field

    pdu->header.versionPrefix = (uint8_t)'F';



    // Populate with the encoded kinetic protocol buffer

    pdu->proto = &buffer[PDU_HEADER_LEN];

    KineticProto_pack(&message->proto, pdu->proto);



    // Populate with the value data, if specified

    if (value == NULL)

    {

        pdu->value = NULL;

        pdu->valueLength = 0;

    }

    else

    {

        pdu->value = &buffer[PDU_HEADER_LEN + protoPackedSize];

        memcpy(pdu->value, value, valueLength);

    }

    // Associate with protobuf

    pdu->protobufLength = KineticProto_get_packed_size(&pdu->protobuf->proto);

    KineticPDU_PackInt32((uint8_t*)&pdu->header.protobufLength, pdu->protobufLength); // Pack protobuf length field



    // Associate PDU with message

    pdu->message = message;

    // Associate with payload (if present)

    pdu->valueLength = valueLength;

    KineticPDU_PackInt32((uint8_t*)&pdu->header.valueLength, pdu->valueLength); // Pack value length field

    pdu->value = value;

}



void KineticPDU_PackInt64(uint8_t* const buffer, int64_t value)

void KineticPDU_PackInt32(uint8_t* const buffer, int32_t value)

{

    int i;

    for (i = sizeof(int64_t) - 1; i >= 0; i--)

    for (i = sizeof(int32_t) - 1; i >= 0; i--)

    {

        buffer[i] = value & 0xFF;

        value = value >> 8;

    }

}



int64_t KineticPDU_UnpackInt64(const uint8_t* const buffer)

int32_t KineticPDU_UnpackInt32(const uint8_t* const buffer)

{

    int i;

    int64_t value = 0;

    for (i = 0; i < sizeof(int64_t); i++)

    int32_t value = 0;

    for (i = 0; i < sizeof(int32_t); i++)

    {

        value <<= 8;

        value += buffer[i];