src/Site2SitePeer.cpp - nifi-minifi-cpp - Git at Google

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

 //! CRC tables
 std::atomic<bool> CRC32::tableInit(false);
 unsigned int CRC32::table[256];

 bool Site2SitePeer::Open()
 {
 	in_addr_t addr;
 	int sock = 0;
 	struct hostent *h;
 	const char *host;
 	uint16_t port;

 	host = this->_host.c_str();
 	port = this->_port;

 	if (strlen(host) == 0)
 		return false;

 #ifdef __MACH__
 	h = gethostbyname(host);
 #else
 	char buf[1024];
 	struct hostent he;
 	int hh_errno;
 	gethostbyname_r(host, &he, buf, sizeof(buf), &h, &hh_errno);
 #endif
 	memcpy((char *) &addr, h->h_addr_list[0], h->h_length);
 	sock = socket(AF_INET, SOCK_STREAM, 0);
 	if (sock < 0)
 	{
 		_logger->log_error("Could not create socket to hostName %s", host);
 		this->yield();
 		return false;
 	}

 #ifndef __MACH__
 	int opt = 1;
 	bool nagle_off = true;

 	if (nagle_off)
 	{
 		if (setsockopt(sock, SOL_TCP, TCP_NODELAY, (void *)&opt, sizeof(opt)) < 0)
 		{
 			_logger->log_error("setsockopt() TCP_NODELAY failed");
 			close(sock);
 			this->yield();
 			return false;
 		}
 		if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
 				(char *)&opt, sizeof(opt)) < 0)
 		{
 			_logger->log_error("setsockopt() SO_REUSEADDR failed");
 			close(sock);
 			this->yield();
 			return false;
 		}
 	}

 	int sndsize = 256*1024;
 	if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *)&sndsize, (int)sizeof(sndsize)) < 0)
 	{
 		_logger->log_error("setsockopt() SO_SNDBUF failed");
 		close(sock);
 		this->yield();
 		return false;
 	}
 	int rcvsize = 256*1024;
 	if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *)&rcvsize, (int)sizeof(rcvsize)) < 0)
 	{
 		_logger->log_error("setsockopt() SO_RCVBUF failed");
 		close(sock);
 		this->yield();
 		return false;
 	}
 #endif

 	struct sockaddr_in sa;
 	socklen_t socklen;
 	int status;

 	memset(&sa, 0, sizeof(sa));
 	sa.sin_family = AF_INET;
 	sa.sin_addr.s_addr = htonl(INADDR_ANY);
 	sa.sin_port = htons(0);
 	socklen = sizeof(sa);
 	if (bind(sock, (struct sockaddr *)&sa, socklen) < 0)
 	{
 		_logger->log_error("socket bind failed");
 		close(sock);
 		this->yield();
 		return false;
 	}

 	memset(&sa, 0, sizeof(sa));
 	sa.sin_family = AF_INET;
 	sa.sin_addr.s_addr = addr;
 	sa.sin_port = htons(port);
 	socklen = sizeof(sa);

 	status = connect(sock, (struct sockaddr *)&sa, socklen);

 	if (status < 0)
 	{
 		_logger->log_error("socket connect failed to %s %d", host, port);
 		close(sock);
 		this->yield();
 		return false;
 	}

 	_logger->log_info("Site2Site Peer socket %d connect to server %s port %d success", sock, host, port);

 	_socket = sock;

 	status = sendData((uint8_t *) MAGIC_BYTES, sizeof(MAGIC_BYTES));

 	if (status <= 0)
 	{
 		Close();
 		return false;
 	}

 	return true;
 }

 void Site2SitePeer::Close()
 {
 	if (_socket)
 	{
 		_logger->log_info("Site2Site Peer socket %d close", _socket);
 		close(_socket);
 		_socket = 0;
 	}
 }

 int Site2SitePeer::sendData(uint8_t *buf, int buflen, CRC32 *crc)
 {
 	int ret = 0, bytes = 0;

 	if (_socket <= 0)
 	{
 		// this->yield();
 		return -1;
 	}

 	while (bytes < buflen)
 	{
 		ret = send(_socket, buf+bytes, buflen-bytes, 0);
 		//check for errors
 		if (ret == -1)
 		{
 			Close();
 			// this->yield();
 			return ret;
 		}
 		bytes+=ret;
 	}

 	if (crc)
 		crc->update(buf, buflen);

 	return bytes;
 }

 int Site2SitePeer::Select(int msec)
 {
 	fd_set fds;
 	struct timeval tv;
     int retval;
     int fd = _socket;

     FD_ZERO(&fds);
     FD_SET(fd, &fds);

     tv.tv_sec = msec/1000;
     tv.tv_usec = (msec % 1000) * 1000;

     if (msec > 0)
        retval = select(fd+1, &fds, NULL, NULL, &tv);
     else
        retval = select(fd+1, &fds, NULL, NULL, NULL);

     if (retval <= 0)
       return retval;
     if (FD_ISSET(fd, &fds))
       return retval;
     else
       return 0;
 }

 int Site2SitePeer::readData(uint8_t *buf, int buflen, CRC32 *crc)
 {
 	int sendSize = buflen;
 	uint8_t *start = buf;

 	if (_socket <= 0)
 	{
 		// this->yield();
 		return -1;
 	}

 	while (buflen)
 	{
 		int status;
 		status = Select((int) _timeOut);
 		if (status <= 0)
 		{
 			Close();
 			return status;
 		}
 		status = recv(_socket, buf, buflen, 0);
 		if (status <= 0)
 		{
 			Close();
 			// this->yield();
 			return status;
 		}
 		buflen -= status;
 		buf += status;
 	}

 	if (crc)
 		crc->update(start, sendSize);

 	return sendSize;
 }

 int Site2SitePeer::writeUTF(std::string str, bool widen, CRC32 *crc)
 {
 	int strlen = str.length();
 	int utflen = 0;
 	int c, count = 0;

 	/* use charAt instead of copying String to char array */
 	for (int i = 0; i < strlen; i++) {
 		c = str.at(i);
 		if ((c >= 0x0001) && (c <= 0x007F)) {
 			utflen++;
 		} else if (c > 0x07FF) {
 			utflen += 3;
 		} else {
 			utflen += 2;
 		}
 	}

 	if (utflen > 65535)
 		return -1;

 	uint8_t *bytearr = NULL;
 	if (!widen)
 	{
 		bytearr = new uint8_t[utflen+2];
 		bytearr[count++] = (uint8_t) ((utflen >> 8) & 0xFF);
 		bytearr[count++] = (uint8_t) ((utflen >> 0) & 0xFF);
 	}
 	else
 	{
 		bytearr = new uint8_t[utflen+4];
 		bytearr[count++] = (uint8_t) ((utflen >> 24) & 0xFF);
 		bytearr[count++] = (uint8_t) ((utflen >> 16) & 0xFF);
 		bytearr[count++] = (uint8_t) ((utflen >> 8) & 0xFF);
 		bytearr[count++] = (uint8_t) ((utflen >> 0) & 0xFF);
 	}

 	int i=0;
 	for (i=0; i<strlen; i++) {
 		c = str.at(i);
 		if (!((c >= 0x0001) && (c <= 0x007F))) break;
 		bytearr[count++] = (uint8_t) c;
 	}

 	for (;i < strlen; i++){
 		c = str.at(i);
 		if ((c >= 0x0001) && (c <= 0x007F)) {
 			bytearr[count++] = (uint8_t) c;
 		} else if (c > 0x07FF) {
 			bytearr[count++] = (uint8_t) (0xE0 | ((c >> 12) & 0x0F));
 			bytearr[count++] = (uint8_t) (0x80 | ((c >>  6) & 0x3F));
 			bytearr[count++] = (uint8_t) (0x80 | ((c >>  0) & 0x3F));
 		} else {
 			bytearr[count++] = (uint8_t) (0xC0 | ((c >>  6) & 0x1F));
 			bytearr[count++] = (uint8_t) (0x80 | ((c >>  0) & 0x3F));
 		}
 	}
 	int ret;
 	if (!widen)
 	{
 		ret = sendData(bytearr, utflen+2, crc);
 	}
 	else
 	{
 		ret = sendData(bytearr, utflen+4, crc);
 	}
 	delete[] bytearr;
 	return ret;
 }

 int Site2SitePeer::readUTF(std::string &str, bool widen, CRC32 *crc)
 {
     uint16_t utflen;
     int ret;

     if (!widen)
     {
     	ret = read(utflen, crc);
     	if (ret <= 0)
     		return ret;
     }
     else
     {
     	uint32_t len;
        	ret = read(len, crc);
         if (ret <= 0)
         	return ret;
         utflen = len;
     }

     uint8_t *bytearr = NULL;
     char *chararr = NULL;
     bytearr = new uint8_t[utflen];
     chararr = new char[utflen];
     memset(chararr, 0, utflen);

     int c, char2, char3;
     int count = 0;
     int chararr_count=0;

     ret = read(bytearr, utflen, crc);
     if (ret <= 0)
     {
     	delete[] bytearr;
     	delete[] chararr;
     	return ret;
     }

     while (count < utflen) {
         c = (int) bytearr[count] & 0xff;
         if (c > 127) break;
         count++;
         chararr[chararr_count++]=(char)c;
     }

     while (count < utflen) {
         c = (int) bytearr[count] & 0xff;
         switch (c >> 4) {
             case 0: case 1: case 2: case 3: case 4: case 5: case 6: case 7:
                 /* 0xxxxxxx*/
                 count++;
                 chararr[chararr_count++]=(char)c;
                 break;
             case 12: case 13:
                 /* 110x xxxx   10xx xxxx*/
                 count += 2;
                 if (count > utflen)
                 {
                 	delete[] bytearr;
                 	delete[] chararr;
                 	return -1;
                 }
                 char2 = (int) bytearr[count-1];
                 if ((char2 & 0xC0) != 0x80)
                 {
                 	delete[] bytearr;
                 	delete[] chararr;
                 	return -1;
                 }
                 chararr[chararr_count++]=(char)(((c & 0x1F) << 6) |
                                                 (char2 & 0x3F));
                 break;
             case 14:
                 /* 1110 xxxx  10xx xxxx  10xx xxxx */
                 count += 3;
                 if (count > utflen)
                 {
                 	delete[] bytearr;
                 	delete[] chararr;
                 	return -1;
                 }
                 char2 = (int) bytearr[count-2];
                 char3 = (int) bytearr[count-1];
                 if (((char2 & 0xC0) != 0x80) || ((char3 & 0xC0) != 0x80))
                 {
                 	delete[] bytearr;
                 	delete[] chararr;
                 	return -1;
                 }
                 chararr[chararr_count++]=(char)(((c     & 0x0F) << 12) |
                                                 ((char2 & 0x3F) << 6)  |
                                                 ((char3 & 0x3F) << 0));
                 break;
             default:
             	delete[] bytearr;
             	delete[] chararr;
             	return -1;
         }
     }
     // The number of chars produced may be less than utflen
     std::string value(chararr, chararr_count);
     str = value;
     delete[] bytearr;
     delete[] chararr;
     if (!widen)
     	return (2 + utflen);
     else
     	return (4 + utflen);
 }
	/**
	* @file Site2SitePeer.cpp
	* Site2SitePeer class implementation
	*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	#include <sys/time.h>
	#include <stdio.h>
	#include <time.h>
	#include <chrono>
	#include <thread>
	#include <random>
	#include <netinet/tcp.h>
	#include <iostream>
	#include "Site2SitePeer.h"

	//! CRC tables
	std::atomic<bool> CRC32::tableInit(false);
	unsigned int CRC32::table[256];

	bool Site2SitePeer::Open()
	{
	in_addr_t addr;
	int sock = 0;
	struct hostent *h;
	const char *host;
	uint16_t port;

	host = this->_host.c_str();
	port = this->_port;

	if (strlen(host) == 0)
	return false;

	#ifdef __MACH__
	h = gethostbyname(host);
	#else
	char buf[1024];
	struct hostent he;
	int hh_errno;
	gethostbyname_r(host, &he, buf, sizeof(buf), &h, &hh_errno);
	#endif
	memcpy((char *) &addr, h->h_addr_list[0], h->h_length);
	sock = socket(AF_INET, SOCK_STREAM, 0);
	if (sock < 0)
	{
	_logger->log_error("Could not create socket to hostName %s", host);
	this->yield();
	return false;
	}

	#ifndef __MACH__
	int opt = 1;
	bool nagle_off = true;

	if (nagle_off)
	{
	if (setsockopt(sock, SOL_TCP, TCP_NODELAY, (void *)&opt, sizeof(opt)) < 0)
	{
	_logger->log_error("setsockopt() TCP_NODELAY failed");
	close(sock);
	this->yield();
	return false;
	}
	if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
	(char *)&opt, sizeof(opt)) < 0)
	{
	_logger->log_error("setsockopt() SO_REUSEADDR failed");
	close(sock);
	this->yield();
	return false;
	}
	}

	int sndsize = 256*1024;
	if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *)&sndsize, (int)sizeof(sndsize)) < 0)
	{
	_logger->log_error("setsockopt() SO_SNDBUF failed");
	close(sock);
	this->yield();
	return false;
	}
	int rcvsize = 256*1024;
	if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *)&rcvsize, (int)sizeof(rcvsize)) < 0)
	{
	_logger->log_error("setsockopt() SO_RCVBUF failed");
	close(sock);
	this->yield();
	return false;
	}
	#endif

	struct sockaddr_in sa;
	socklen_t socklen;
	int status;

	memset(&sa, 0, sizeof(sa));
	sa.sin_family = AF_INET;
	sa.sin_addr.s_addr = htonl(INADDR_ANY);
	sa.sin_port = htons(0);
	socklen = sizeof(sa);
	if (bind(sock, (struct sockaddr *)&sa, socklen) < 0)
	{
	_logger->log_error("socket bind failed");
	close(sock);
	this->yield();
	return false;
	}

	memset(&sa, 0, sizeof(sa));
	sa.sin_family = AF_INET;
	sa.sin_addr.s_addr = addr;
	sa.sin_port = htons(port);
	socklen = sizeof(sa);

	status = connect(sock, (struct sockaddr *)&sa, socklen);

	if (status < 0)
	{
	_logger->log_error("socket connect failed to %s %d", host, port);
	close(sock);
	this->yield();
	return false;
	}

	_logger->log_info("Site2Site Peer socket %d connect to server %s port %d success", sock, host, port);

	_socket = sock;

	status = sendData((uint8_t *) MAGIC_BYTES, sizeof(MAGIC_BYTES));

	if (status <= 0)
	{
	Close();
	return false;
	}

	return true;
	}

	void Site2SitePeer::Close()
	{
	if (_socket)
	{
	_logger->log_info("Site2Site Peer socket %d close", _socket);
	close(_socket);
	_socket = 0;
	}
	}

	int Site2SitePeer::sendData(uint8_t buf, int buflen, CRC32 crc)
	{
	int ret = 0, bytes = 0;

	if (_socket <= 0)
	{
	// this->yield();
	return -1;
	}

	while (bytes < buflen)
	{
	ret = send(_socket, buf+bytes, buflen-bytes, 0);
	//check for errors
	if (ret == -1)
	{
	Close();
	// this->yield();
	return ret;
	}
	bytes+=ret;
	}

	if (crc)
	crc->update(buf, buflen);

	return bytes;
	}

	int Site2SitePeer::Select(int msec)
	{
	fd_set fds;
	struct timeval tv;
	int retval;
	int fd = _socket;

	FD_ZERO(&fds);
	FD_SET(fd, &fds);

	tv.tv_sec = msec/1000;
	tv.tv_usec = (msec % 1000) * 1000;

	if (msec > 0)
	retval = select(fd+1, &fds, NULL, NULL, &tv);
	else
	retval = select(fd+1, &fds, NULL, NULL, NULL);

	if (retval <= 0)
	return retval;
	if (FD_ISSET(fd, &fds))
	return retval;
	else
	return 0;
	}

	int Site2SitePeer::readData(uint8_t buf, int buflen, CRC32 crc)
	{
	int sendSize = buflen;
	uint8_t *start = buf;

	if (_socket <= 0)
	{
	// this->yield();
	return -1;
	}

	while (buflen)
	{
	int status;
	status = Select((int) _timeOut);
	if (status <= 0)
	{
	Close();
	return status;
	}
	status = recv(_socket, buf, buflen, 0);
	if (status <= 0)
	{
	Close();
	// this->yield();
	return status;
	}
	buflen -= status;
	buf += status;
	}

	if (crc)
	crc->update(start, sendSize);

	return sendSize;
	}

	int Site2SitePeer::writeUTF(std::string str, bool widen, CRC32 *crc)
	{
	int strlen = str.length();
	int utflen = 0;
	int c, count = 0;

	/* use charAt instead of copying String to char array */
	for (int i = 0; i < strlen; i++) {
	c = str.at(i);
	if ((c >= 0x0001) && (c <= 0x007F)) {
	utflen++;
	} else if (c > 0x07FF) {
	utflen += 3;
	} else {
	utflen += 2;
	}
	}

	if (utflen > 65535)
	return -1;

	uint8_t *bytearr = NULL;
	if (!widen)
	{
	bytearr = new uint8_t[utflen+2];
	bytearr[count++] = (uint8_t) ((utflen >> 8) & 0xFF);
	bytearr[count++] = (uint8_t) ((utflen >> 0) & 0xFF);
	}
	else
	{
	bytearr = new uint8_t[utflen+4];
	bytearr[count++] = (uint8_t) ((utflen >> 24) & 0xFF);
	bytearr[count++] = (uint8_t) ((utflen >> 16) & 0xFF);
	bytearr[count++] = (uint8_t) ((utflen >> 8) & 0xFF);
	bytearr[count++] = (uint8_t) ((utflen >> 0) & 0xFF);
	}

	int i=0;
	for (i=0; i<strlen; i++) {
	c = str.at(i);
	if (!((c >= 0x0001) && (c <= 0x007F))) break;
	bytearr[count++] = (uint8_t) c;
	}

	for (;i < strlen; i++){
	c = str.at(i);
	if ((c >= 0x0001) && (c <= 0x007F)) {
	bytearr[count++] = (uint8_t) c;
	} else if (c > 0x07FF) {
	bytearr[count++] = (uint8_t) (0xE0 \| ((c >> 12) & 0x0F));
	bytearr[count++] = (uint8_t) (0x80 \| ((c >> 6) & 0x3F));
	bytearr[count++] = (uint8_t) (0x80 \| ((c >> 0) & 0x3F));
	} else {
	bytearr[count++] = (uint8_t) (0xC0 \| ((c >> 6) & 0x1F));
	bytearr[count++] = (uint8_t) (0x80 \| ((c >> 0) & 0x3F));
	}
	}
	int ret;
	if (!widen)
	{
	ret = sendData(bytearr, utflen+2, crc);
	}
	else
	{
	ret = sendData(bytearr, utflen+4, crc);
	}
	delete[] bytearr;
	return ret;
	}

	int Site2SitePeer::readUTF(std::string &str, bool widen, CRC32 *crc)
	{
	uint16_t utflen;
	int ret;

	if (!widen)
	{
	ret = read(utflen, crc);
	if (ret <= 0)
	return ret;
	}
	else
	{
	uint32_t len;
	ret = read(len, crc);
	if (ret <= 0)
	return ret;
	utflen = len;
	}

	uint8_t *bytearr = NULL;
	char *chararr = NULL;
	bytearr = new uint8_t[utflen];
	chararr = new char[utflen];
	memset(chararr, 0, utflen);

	int c, char2, char3;
	int count = 0;
	int chararr_count=0;

	ret = read(bytearr, utflen, crc);
	if (ret <= 0)
	{
	delete[] bytearr;
	delete[] chararr;
	return ret;
	}

	while (count < utflen) {
	c = (int) bytearr[count] & 0xff;
	if (c > 127) break;
	count++;
	chararr[chararr_count++]=(char)c;
	}

	while (count < utflen) {
	c = (int) bytearr[count] & 0xff;
	switch (c >> 4) {
	case 0: case 1: case 2: case 3: case 4: case 5: case 6: case 7:
	/* 0xxxxxxx*/
	count++;
	chararr[chararr_count++]=(char)c;
	break;
	case 12: case 13:
	/* 110x xxxx 10xx xxxx*/
	count += 2;
	if (count > utflen)
	{
	delete[] bytearr;
	delete[] chararr;
	return -1;
	}
	char2 = (int) bytearr[count-1];
	if ((char2 & 0xC0) != 0x80)
	{
	delete[] bytearr;
	delete[] chararr;
	return -1;
	}
	chararr[chararr_count++]=(char)(((c & 0x1F) << 6) \|
	(char2 & 0x3F));
	break;
	case 14:
	/* 1110 xxxx 10xx xxxx 10xx xxxx */
	count += 3;
	if (count > utflen)
	{
	delete[] bytearr;
	delete[] chararr;
	return -1;
	}
	char2 = (int) bytearr[count-2];
	char3 = (int) bytearr[count-1];
	if (((char2 & 0xC0) != 0x80) \|\| ((char3 & 0xC0) != 0x80))
	{
	delete[] bytearr;
	delete[] chararr;
	return -1;
	}
	chararr[chararr_count++]=(char)(((c & 0x0F) << 12) \|
	((char2 & 0x3F) << 6) \|
	((char3 & 0x3F) << 0));
	break;
	default:
	delete[] bytearr;
	delete[] chararr;
	return -1;
	}
	}
	// The number of chars produced may be less than utflen
	std::string value(chararr, chararr_count);
	str = value;
	delete[] bytearr;
	delete[] chararr;
	if (!widen)
	return (2 + utflen);
	else
	return (4 + utflen);
	}