src/main/cpp/cambria.cpp

/*******************************************************************************
 *  ============LICENSE_START=======================================================
 *  org.onap.dmaap
 *  ================================================================================
 *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
 *  ================================================================================
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *        http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *  ============LICENSE_END=========================================================
 *
 *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
 *  
 *******************************************************************************/

/*
 * This is a client library for the AT&T Cambria Event Routing Service.
 */

#include "cambria.h"

#include <arpa/inet.h>
#include <sys/socket.h>
#include <netdb.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <string.h>

#include <string>
#include <list>
#include <sstream>
#include <iomanip>
#include <algorithm>

// field used in JSON encoding to signal stream name
const char* kPartition = "cambria.partition";

// map from opaque handle to object pointer
#define toOpaque(x) ((CAMBRIA_CLIENT)x)
#define fromOpaque(x) ((cambriaClient*)x)

// trace support
extern void traceOutput ( const char* format, ... );
#ifdef CAMBRIA_TRACING
	#define TRACE traceOutput
#else
	#define TRACE 1 ? (void) 0 : traceOutput
#endif

/*
 *	internal cambria client class
 */
class cambriaClient
{
public:
	cambriaClient ( const std::string& host, int port, const std::string& topic, const std::string& format );
	~cambriaClient ();

	cambriaSendResponse* send ( const char* streamName, const char* message );
	cambriaSendResponse* send ( const char* streamName, const char** message, unsigned int count );

	cambriaGetResponse* get ( int timeoutMs, int limit );

private:

	std::string fHost;
	int fPort;
	std::string fTopic;
	std::string fFormat;

	bool buildMessageBody ( const char* streamName, const char** msgs, unsigned int count, std::string& buffer );

	void write ( int socket, const char* line );
};

cambriaClient::cambriaClient ( const std::string& host, int port, const std::string& topic, const std::string& format ) :
	fHost ( host ),
	fPort ( port ),
	fTopic ( topic ),
	fFormat ( format )
{
}

cambriaClient::~cambriaClient ()
{
}

/*
 *	This isn't quite right -- if the message already has cambria.partition,
 *	it'll wind up with two entries. Also, message MUST start with '{' and
 *	have at least one field.
 */
static char* makeJsonMessage ( const char* streamName, const char* message )
{
	int len = ::strlen ( message );
	if ( streamName )
	{
		len += ::strlen ( kPartition );
		len += ::strlen ( streamName );
		len += 6; 	// quote each and a colon and comma
	}

	char* msg = new char [ len + 1 ];
	::strcpy ( msg, "{" );
	if ( streamName )
	{
		::strcat ( msg, "\"" );
		::strcat ( msg, kPartition );
		::strcat ( msg, "\":\"" );
		::strcat ( msg, streamName );
		::strcat ( msg, "\"," );
	}
	::strcat ( msg, message + 1 );
	return msg;
}

cambriaSendResponse* cambriaClient::send ( const char* streamName, const char* message )
{
	return send ( streamName, &message, 1 );
}

static bool replace ( std::string& str, const std::string& from, const std::string& to )
{
	size_t start_pos = str.find ( from );
	if(start_pos == std::string::npos)
		return false;
	str.replace(start_pos, from.length(), to);
	return true;
}

static void readResponse ( int s, std::string& response )
{
	char buffer [ 4096 ];

	ssize_t n = 0;
	while ( ( n = ::read ( s, buffer, 4095 ) ) > 0 )
	{
		buffer[n] = '\0';
		response += buffer;
	}
}

static int openSocket ( std::string& host, int port, int& ss )
{
	TRACE( "connecting to %s\n", host.c_str() );

	struct hostent *he = ::gethostbyname ( host.c_str() );
	if ( !he )
	{
		TRACE("no host entry\n");
		return CAMBRIA_NO_HOST;
	}

	if ( he->h_addrtype != AF_INET )
	{
		TRACE("not AF_INET\n");
		return CAMBRIA_NO_HOST;
	}

	int s = ::socket ( AF_INET, SOCK_STREAM, 0 );
	if ( s == -1 )
	{
		TRACE("no socket available\n");
		return CAMBRIA_CANT_CONNECT;
	}

	struct sockaddr_in servaddr;
	::memset ( &servaddr, 0, sizeof(servaddr) );

	::memcpy ( &servaddr.sin_addr, he->h_addr_list[0], he->h_length );
	servaddr.sin_family = AF_INET;
	servaddr.sin_port = ::htons ( port );

	if ( ::connect ( s, (struct sockaddr *)&servaddr, sizeof(servaddr) ) )
	{
		TRACE("couldn't connect\n");
		return CAMBRIA_CANT_CONNECT;
	}

	ss = s;
	return 0;
}

cambriaSendResponse* cambriaClient::send ( const char* streamName, const char** msgs, unsigned int count )
{
	TRACE ( "Sending %d messages.", count );

	cambriaSendResponse* result = new cambriaSendResponse ();
	result->statusCode = 0;
	result->statusMessage = NULL;
	result->responseBody = NULL;

	TRACE( "building message body\n" );

	std::string body;
	if ( !buildMessageBody ( streamName, msgs, count, body ) )
	{
		result->statusCode = CAMBRIA_UNRECOGNIZED_FORMAT;
		return result;
	}

	int s = -1;
	int err = ::openSocket ( fHost, fPort, s );
	if ( err > 0 )
	{
		result->statusCode = err;
		return result;
	}

	// construct path
	std::string path = "/cambriaApiServer/v1/event/";
	path += fTopic;

	// send post prefix
	char line[4096];
	::sprintf ( line,
		"POST %s HTTP/1.0\r\n"
		"Host: %s\r\n"
		"Content-Type: %s\r\n"
		"Content-Length: %d\r\n"
		"\r\n",
		path.c_str(), fHost.c_str(), fFormat.c_str(), body.length() );
	write ( s, line );

	// send the body
	write ( s, body.c_str() );

	TRACE ( "\n" );
	TRACE ( "send complete, reading reply\n" );

	// receive the response
	std::string response;
	readResponse ( s, response );
	::close ( s );

	// parse the header and body: split header and body on first occurrence of \r\n\r\n
	result->statusCode = CAMBRIA_BAD_RESPONSE;

	size_t headerBreak = response.find ( "\r\n\r\n" );
    if ( headerBreak != std::string::npos )
    {
		std::string responseBody = response.substr ( headerBreak + 4 );
		result->responseBody = new char [ responseBody.length() + 1 ];
		::strcpy ( result->responseBody, responseBody.c_str() );

		// all we need from the header for now is the status line
		std::string headerPart = response.substr ( 0, headerBreak + 2 );
		
		size_t newline = headerPart.find ( '\r' );
		if ( newline != std::string::npos )
		{
			std::string statusLine = headerPart.substr ( 0, newline );

			size_t firstSpace = statusLine.find ( ' ' );
			if ( firstSpace != std::string::npos )
			{
				size_t secondSpace = statusLine.find ( ' ', firstSpace + 1 );
				if ( secondSpace != std::string::npos )
				{
					result->statusCode = ::atoi ( statusLine.substr ( firstSpace + 1, secondSpace - firstSpace + 1 ).c_str() );
					std::string statusMessage = statusLine.substr ( secondSpace + 1 );
					result->statusMessage = new char [ statusMessage.length() + 1 ];
					::strcpy ( result->statusMessage, statusMessage.c_str() );
				}
			}
		}
	}
	return result;
}

void cambriaClient::write ( int socket, const char* str )
{
	int len = str ? ::strlen ( str ) : 0;
	::write ( socket, str, len );

	// elaborate tracing nonsense...
	std::string trace ( "> " );
	trace += str;
	while ( replace ( trace, "\r\n", "\\r\\n\n> " ) );

	TRACE ( "%s", trace.c_str() );
}

bool cambriaClient::buildMessageBody ( const char* streamName, const char** msgs, unsigned int count, std::string& buffer )
{
	if ( fFormat == CAMBRIA_NATIVE_FORMAT )
	{
		int snLen = ::strlen ( streamName );
		for ( unsigned int i=0; i<count; i++ )
		{
			const char* msg = msgs[i];

			std::ostringstream s;
			s << snLen << '.' << ::strlen(msg) << '.' << streamName << msg;
			buffer.append ( s.str() );
		}
	}
	else if ( fFormat == CAMBRIA_JSON_FORMAT )
	{
		buffer.append ( "[" );
		for ( unsigned int i=0; i<count; i++ )
		{
			if ( i>0 )
			{
				buffer.append ( "," );
			}
			const char* msg = msgs[i];
			char* jsonMsg = ::makeJsonMessage ( streamName, msg );
			buffer.append ( jsonMsg );
			delete jsonMsg;	// FIXME: allocating memory here just to delete it
		}
		buffer.append ( "]" );
	}
	else
	{
		return false;
	}
	return true;
}

// read the next string into value, and return the end pos, or 0 on error
static int readNextJsonString ( const std::string& body, int startPos, std::string& value )
{
	value = "";

	if ( startPos >= body.length () )
	{
		return 0;
	}

	// skip a comma
	int current = startPos;
	if ( body[current] == ',' ) current++;

	if ( current >= body.length() || body[current] != '"' )
	{
		return 0;
	}
	current++;

	// walk the string for the closing quote (FIXME: unicode support)
	bool esc = false;
	int hex = 0;
	while ( ( body[current] != '"' || esc ) && current < body.length() )
	{
		if ( hex > 0 )
		{
			hex--;
			if ( hex == 0 )
			{
				// presumably read a unicode escape. this code isn't
				// equipped for multibyte or unicode, so just skip it
				value += '?';
			}
		}
		else if ( esc )
		{
			esc = false;
			switch ( body[current] )
			{
				case '"':
				case '\\':
				case '/':
					value += body[current];
					break;

				case 'b': value += '\b'; break;
				case 'f': value += '\f'; break;
				case 'n': value += '\n'; break;
				case 'r': value += '\r'; break;
				case 't': value += '\t'; break;

				case 'u': hex=4; break;
			}
		}
		else
		{
			esc = body[current] == '\\';
			if ( !esc ) value += body[current];
		}
		current++;
	}

	return current + 1;
}

static void readGetBody ( std::string& body, cambriaGetResponse& response )
{
	TRACE("response %s\n", body.c_str() );	
	
	if ( body.length() < 2 || body[0] != '[' || body[body.length()-1] != ']' )
	{
		response.statusCode = CAMBRIA_BAD_RESPONSE;
	}

	std::list<char*> msgs;
	std::string val;
	int current = 1;
	while ( ( current = readNextJsonString ( body, current, val ) ) > 0 )
	{
		char* msg = new char [ val.length() + 1 ];
		::strcpy ( msg, val.c_str() );
		msgs.push_back ( msg );
	}
	
	// now build a response
	response.messageCount = msgs.size();
	response.messageSet = new char* [ msgs.size() ];
	int index = 0;
	for ( std::list<char*>::iterator it = msgs.begin(); it != msgs.end(); it++ )
	{
		response.messageSet [ index++ ] = *it;
	}
}

cambriaGetResponse* cambriaClient::get ( int timeoutMs, int limit )
{
	cambriaGetResponse* result = new cambriaGetResponse ();
	result->statusCode = 0;
	result->statusMessage = NULL;
	result->messageCount = 0;
	result->messageSet = new char* [ 1 ];

	int s = -1;
	int err = ::openSocket ( fHost, fPort, s );
	if ( err > 0 )
	{
		result->statusCode = err;
		return result;
	}
	
	// construct path
	std::string path = "/cambriaApiServer/v1/event/";
	path += fTopic;

	bool haveAdds = false;
	std::ostringstream adds;
	if ( timeoutMs > -1 )
	{
		adds << "timeout=" << timeoutMs;
		haveAdds = true;
	}
	if ( limit > -1 )
	{
		if ( haveAdds )
		{
			adds << "&";
		}
		adds << "limit=" << limit;
		haveAdds = true;
	}
	if ( haveAdds )
	{
		path += "?";
		path += adds.str();
	}

	// send post prefix
	char line[4096];
	::sprintf ( line,
		"GET %s HTTP/1.0\r\n"
		"Host: %s\r\n"
		"\r\n",
		path.c_str(), fHost.c_str() );
	write ( s, line );

	TRACE ( "\n" );
	TRACE ( "request sent; reading reply\n" );

	// receive the response (FIXME: would be nice to stream rather than load it all)
	std::string response;
	readResponse ( s, response );
	::close ( s );

	// parse the header and body: split header and body on first occurrence of \r\n\r\n
	result->statusCode = CAMBRIA_BAD_RESPONSE;

	size_t headerBreak = response.find ( "\r\n\r\n" );
    if ( headerBreak != std::string::npos )
    {
		// get the header line
		std::string headerPart = response.substr ( 0, headerBreak + 2 );

		size_t newline = headerPart.find ( '\r' );
		if ( newline != std::string::npos )
		{
			std::string statusLine = headerPart.substr ( 0, newline );

			size_t firstSpace = statusLine.find ( ' ' );
			if ( firstSpace != std::string::npos )
			{
				size_t secondSpace = statusLine.find ( ' ', firstSpace + 1 );
				if ( secondSpace != std::string::npos )
				{
					result->statusCode = ::atoi ( statusLine.substr ( firstSpace + 1, secondSpace - firstSpace + 1 ).c_str() );
					std::string statusMessage = statusLine.substr ( secondSpace + 1 );
					result->statusMessage = new char [ statusMessage.length() + 1 ];
					::strcpy ( result->statusMessage, statusMessage.c_str() );
				}
			}
		}

		if ( result->statusCode < 300 )
		{
			std::string responseBody = response.substr ( headerBreak + 4 );
			readGetBody ( responseBody, *result );
		}
	}
	return result;
}


///////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////

CAMBRIA_CLIENT cambriaCreateClient ( const char* host, int port, const char* topic, const char* format )
{
	cambriaClient* cc = new cambriaClient ( host, port, topic, format );
	return toOpaque(cc);
}

void cambriaDestroyClient ( CAMBRIA_CLIENT client )
{
	delete fromOpaque ( client );
}

cambriaSendResponse* cambriaSendMessage ( CAMBRIA_CLIENT client, const char* streamName, const char* message )
{
	cambriaClient* c = fromOpaque ( client );
	return c->send ( streamName, message );
}

cambriaSendResponse* cambriaSendMessages ( CAMBRIA_CLIENT client, const char* streamName, const char** messages, unsigned int count )
{
	cambriaClient* c = fromOpaque ( client );
	return c->send ( streamName, messages, count );
}

cambriaGetResponse* cambriaGetMessages ( CAMBRIA_CLIENT client, unsigned long timeoutMs, unsigned int limit )
{
	cambriaClient* c = fromOpaque ( client );
	return c->get ( timeoutMs, limit );
}

void cambriaDestroySendResponse ( CAMBRIA_CLIENT client, const cambriaSendResponse* response )
{
	if ( response )
	{
		delete response->statusMessage;
		delete response->responseBody;
		delete response;
	}
}

void cambriaDestroyGetResponse ( CAMBRIA_CLIENT client, const cambriaGetResponse* response )
{
	if ( response )
	{
		delete response->statusMessage;
		for ( int i=0; i<response->messageCount; i++ )
		{
			delete response->messageSet[i];
		}
		delete response;
	}
}

int cambriaSimpleSend ( const char* host, int port, const char* topic, const char* streamName, const char* msg )
{
	return cambriaSimpleSendMultiple ( host, port, topic, streamName, &msg, 1 );
}

int cambriaSimpleSendMultiple ( const char* host, int port, const char* topic, const char* streamName, const char** messages, unsigned int msgCount )
{
	int count = 0;

	const CAMBRIA_CLIENT cc = ::cambriaCreateClient ( host, port, topic, CAMBRIA_NATIVE_FORMAT );
	if ( cc )
	{
		const cambriaSendResponse* response = ::cambriaSendMessages ( cc, streamName, messages, msgCount );
		if ( response && response->statusCode < 300 )
		{
			count = msgCount;
		}
		::cambriaDestroySendResponse ( cc, response );
		::cambriaDestroyClient ( cc );
	}

	return count;
}

////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////

const unsigned int kMaxTraceBuffer = 2048;

static void writeTraceString ( const char* msg )
{
	::fprintf ( stdout, "%s", msg );
	::fflush ( stdout );	// because we want output before core dumping :-)
}

void traceOutput ( const char* format, ... )
{
	char buffer [ kMaxTraceBuffer ];
	::memset ( buffer, '\0', kMaxTraceBuffer * sizeof ( char ) );

	va_list list;
	va_start ( list, format );
	::vsprintf ( buffer, format, list );
	writeTraceString ( buffer );
	va_end ( list );
}