<commit_before>#include "networklayer.h"
#include "ua_transport_connection.h"

#ifdef WIN32
#pragma comment (lib,"ws2_32.lib")
#include <sys/types.h>
#include <Windows.h>
#include <ws2tcpip.h>
#define CLOSESOCKET(S) closesocket(S); \
	WSACleanup();
#define IOCTLSOCKET ioctlsocket
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/socketvar.h>
#include <unistd.h> // read, write, close
#define CLOSESOCKET(S) close(S)
#define IOCTLSOCKET ioctl
#endif

#include <stdlib.h> // exit
#include <errno.h> // errno, EINTR

#include <memory.h> // memset
#include <fcntl.h> // fcntl

NL_Description NL_Description_TcpBinary  = {
	NL_UA_ENCODING_BINARY,
	NL_CONNECTIONTYPE_TCPV4,
	NL_MAXCONNECTIONS_DEFAULT,
	{0,8192,8192,16384,1}
};

/* If we do not have multitasking, we implement a dispatcher-Pattern. All Connections
 * are collected in a list. From this list a fd_set is prepared and select then waits
 * for activities. We then iterate over the list, check if we've got some activites
 * and call the corresponding callback (reader, listener).
 */
#ifndef MULTITASKING
_Bool NL_ConnectionComparer(void *p1, void* p2) {
	NL_Connection* c1 = (NL_Connection*) p1;
	NL_Connection* c2 = (NL_Connection*) p2;

	return (c1->connectionHandle == c2->connectionHandle);

}

int NL_TCP_SetNonBlocking(int sock) {
#ifdef WIN32
	UA_Int64 iMode = 1;
	int opts = IOCTLSOCKET(sock, FIONBIO, &iMode);
	if (opts != NO_ERROR){
		printf("ioctlsocket failed with error: %ld\n", opts);
		return - 1;
	}
#else
	int opts = fcntl(sock,F_GETFL);
	if (opts < 0) {
		perror("fcntl(F_GETFL)");
		return -1;
	}
	opts = (opts | O_NONBLOCK);
	if (fcntl(sock,F_SETFL,opts) < 0) {
		perror("fcntl(F_SETFL)");
		return -1;
	}
#endif
	return 0;
}

void NL_Connection_printf(void* payload) {
  NL_Connection* c = (NL_Connection*) payload;
  printf("ListElement connectionHandle = %d\n",c->connectionHandle);
}
void NL_addHandleToSet(UA_Int32 handle, NL_data* nl) {
	FD_SET(handle, &(nl->readerHandles));
#ifdef WIN32
	int err = WSAGetLastError();
#endif
	nl->maxReaderHandle = (handle > nl->maxReaderHandle) ? handle : nl->maxReaderHandle;
}
void NL_setFdSet(void* payload) {
  NL_Connection* c = (NL_Connection*) payload;
  NL_addHandleToSet(c->connectionHandle, c->networkLayer);
}
void NL_checkFdSet(void* payload) {
  NL_Connection* c = (NL_Connection*) payload;
  if (FD_ISSET(c->connectionHandle, &(c->networkLayer->readerHandles))) {
	  c->reader((void*)c);
  }
}
UA_Int32 NL_msgLoop(NL_data* nl, struct timeval *tv, UA_Int32(*worker)(void*), void *arg, UA_Boolean *running)  {
	UA_Int32 result;
	while (*running) {
		// determine the largest handle
		nl->maxReaderHandle = 0;
		UA_list_iteratePayload(&(nl->connections),NL_setFdSet);
		DBG_VERBOSE(printf("\n------------\nUA_Stack_msgLoop - maxHandle=%d\n", nl->maxReaderHandle));

		// copy tv, some unixes do overwrite and return the remaining time
		struct timeval tmptv;
		memcpy(&tmptv,tv,sizeof(struct timeval));

		// and wait
		DBG_VERBOSE(printf("UA_Stack_msgLoop - enter select sec=%d,usec=%d\n",(UA_Int32) tmptv.tv_sec, (UA_Int32) tmptv.tv_usec));
		result = select(nl->maxReaderHandle + 1, &(nl->readerHandles), UA_NULL, UA_NULL, &tmptv);

		DBG_VERBOSE(printf("UA_Stack_msgLoop - leave select result=%d,sec=%d,usec=%d\n",result, (UA_Int32) tmptv.tv_sec, (UA_Int32) tmptv.tv_usec));
#ifdef WIN32
		if (result == -1) {
			DBG_ERR(printf("UA_Stack_msgLoop - errno = { %d, %s }\n", WSAGetLastError()));
		}
		else if (result >= 0){ // activity on listener or client ports
#else
		if (result == 0) {
			UA_Int32 err = errno;

			switch (err) {
			// handle known errors
			case EBADF:
			case EINTR:
			case EINVAL:
				// FIXME: handle errors
				DBG_ERR(printf("UA_Stack_msgLoop - errno={%d,%s}\n", errno, strerror(errno)));
				break;
			// we've got a timeout
			case EAGAIN:
			default:
				DBG_VERBOSE(printf("UA_Stack_msgLoop - errno={%d,%s}\n", errno, strerror(errno)));
				DBG_VERBOSE(printf("UA_Stack_msgLoop - call worker\n"));
				worker(arg);
				DBG_VERBOSE(printf("UA_Stack_msgLoop - return from worker\n"));
			}
		}

		else if (result > 0){ // activity on listener or client ports
#endif
			DBG_VERBOSE(printf("UA_Stack_msgLoop - activities on %d handles\n",result));
			UA_list_iteratePayload(&(nl->connections),NL_checkFdSet);
		}
		// FIXME: Seems to be a conceptional flaw to call the worker here...
		worker(arg);
	}

	return UA_SUCCESS;
}
#endif


/** the tcp reader function */
void* NL_TCP_reader(NL_Connection *c) {

	UA_ByteString readBuffer;

	TL_Buffer localBuffers;
	UA_Int32 connectionState;

	UA_TL_Connection_getLocalConfig(c->connection, &localBuffers);
	UA_alloc((void**)&(readBuffer.data),localBuffers.recvBufferSize);

	UA_TL_Connection_getState(c->connection, &connectionState);
	if (connectionState  != CONNECTIONSTATE_CLOSE) {
		DBG_VERBOSE(printf("NL_TCP_reader - enter read\n"));

#ifdef WIN32
		readBuffer.length = recv(c->connectionHandle, readBuffer.data, localBuffers.recvBufferSize, 0);
#else
		readBuffer.length = read(c->connectionHandle, readBuffer.data, localBuffers.recvBufferSize);
#endif
		DBG_VERBOSE(printf("NL_TCP_reader - leave read\n"));

		DBG_VERBOSE(printf("NL_TCP_reader - src={%*.s}, ",c->connection.remoteEndpointUrl.length,c->connection.remoteEndpointUrl.data));
		DBG(UA_ByteString_printx("NL_TCP_reader - received=",&readBuffer));

		if (readBuffer.length  > 0) {

#ifdef DEBUG
#include "ua_transport_binary_secure.h"
			UA_UInt32 pos = 0;
			UA_OPCUATcpMessageHeader header;
			UA_OPCUATcpMessageHeader_decodeBinary(&readBuffer, &pos, &header);
			pos = 24;
			if(header.messageType == UA_MESSAGETYPE_MSG)
			{
				UA_NodeId serviceRequestType;
				UA_NodeId_decodeBinary(&readBuffer, &pos,&serviceRequestType);
				UA_NodeId_printf("Service Type\n",&serviceRequestType);
			}
#endif

			TL_Process((c->connection),&readBuffer);
		} else {
			perror("ERROR reading from socket1");
			UA_TL_Connection_setState(c->connection, CONNECTIONSTATE_CLOSE);
		}
	}

	UA_TL_Connection_getState(c->connection, &connectionState);
	DBG_VERBOSE(printf("NL_TCP_reader - connectionState=%d\n",connectionState));
	if (connectionState == CONNECTIONSTATE_CLOSE) {
		// set connection's state to CONNECTIONSTATE_CLOSED and call callback to actually close
		UA_TL_Connection_close(c->connection);
#ifndef MULTITHREADING
		DBG_VERBOSE(printf("NL_TCP_reader - search element to remove\n"));
		UA_list_Element* lec = UA_list_search(&(c->networkLayer->connections),NL_ConnectionComparer,c);
		DBG_VERBOSE(printf("NL_TCP_reader - remove connection for handle=%d\n",((NL_Connection*)lec->payload)->connection.connectionHandle));
		UA_list_removeElement(lec,UA_NULL);
		DBG_VERBOSE(UA_list_iteratePayload(&(c->networkLayer->connections),NL_Connection_printf));
		UA_free(c);
#endif
	}
	UA_ByteString_deleteMembers(&readBuffer);
	return UA_NULL;
}

#ifdef MULTITHREADING
/** the tcp reader thread */
void* NL_TCP_readerThread(NL_Connection *c) {
	// just loop, NL_TCP_Reader will call the stack
	UA_Int32 connectionState;
	do {<commit_msg>found the bug in winsock #113. WSACleanup does shutdown the winsock.dll - a little bit of an overkill to call this whenever closing any socket...<commit_after>#include "networklayer.h"
#include "ua_transport_connection.h"

#ifdef WIN32
#pragma comment (lib,"ws2_32.lib")
#include <sys/types.h>
#include <Windows.h>
#include <ws2tcpip.h>
#define CLOSESOCKET(S) closesocket(S)
#define IOCTLSOCKET ioctlsocket
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/socketvar.h>
#include <unistd.h> // read, write, close
#define CLOSESOCKET(S) close(S)
#define IOCTLSOCKET ioctl
#endif

#include <stdlib.h> // exit
#include <errno.h> // errno, EINTR

#include <memory.h> // memset
#include <fcntl.h> // fcntl

NL_Description NL_Description_TcpBinary  = {
	NL_UA_ENCODING_BINARY,
	NL_CONNECTIONTYPE_TCPV4,
	NL_MAXCONNECTIONS_DEFAULT,
	{0,8192,8192,16384,1}
};

/* If we do not have multitasking, we implement a dispatcher-Pattern. All Connections
 * are collected in a list. From this list a fd_set is prepared and select then waits
 * for activities. We then iterate over the list, check if we've got some activites
 * and call the corresponding callback (reader, listener).
 */
#ifndef MULTITASKING
_Bool NL_ConnectionComparer(void *p1, void* p2) {
	NL_Connection* c1 = (NL_Connection*) p1;
	NL_Connection* c2 = (NL_Connection*) p2;

	return (c1->connectionHandle == c2->connectionHandle);

}

int NL_TCP_SetNonBlocking(int sock) {
#ifdef WIN32
	UA_Int64 iMode = 1;
	int opts = IOCTLSOCKET(sock, FIONBIO, &iMode);
	if (opts != NO_ERROR){
		printf("ioctlsocket failed with error: %ld\n", opts);
		return - 1;
	}
#else
	int opts = fcntl(sock,F_GETFL);
	if (opts < 0) {
		perror("fcntl(F_GETFL)");
		return -1;
	}
	opts = (opts | O_NONBLOCK);
	if (fcntl(sock,F_SETFL,opts) < 0) {
		perror("fcntl(F_SETFL)");
		return -1;
	}
#endif
	return 0;
}

void NL_Connection_printf(void* payload) {
  NL_Connection* c = (NL_Connection*) payload;
  printf("ListElement connectionHandle = %d\n",c->connectionHandle);
}
void NL_addHandleToSet(UA_Int32 handle, NL_data* nl) {
	FD_SET(handle, &(nl->readerHandles));
#ifdef WIN32
	// int err = WSAGetLastError();
#endif
	nl->maxReaderHandle = (handle > nl->maxReaderHandle) ? handle : nl->maxReaderHandle;
}
void NL_setFdSet(void* payload) {
  NL_Connection* c = (NL_Connection*) payload;
  NL_addHandleToSet(c->connectionHandle, c->networkLayer);
}
void NL_checkFdSet(void* payload) {
  NL_Connection* c = (NL_Connection*) payload;
  if (FD_ISSET(c->connectionHandle, &(c->networkLayer->readerHandles))) {
	  c->reader((void*)c);
  }
}




#if 0 
char _str_error[256];
char* strerror(int errno) {
	FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
		NULL, errno,
		MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
		&_str_error[0], 256, NULL);
	return &_str_error[0];
}
#endif

UA_Int32 NL_msgLoop(NL_data* nl, struct timeval *tv, UA_Int32(*worker)(void*), void *arg, UA_Boolean *running)  {
	UA_Int32 result;
	UA_Int32 err;
	while (*running) {
		// determine the largest handle
		nl->maxReaderHandle = 0;
		UA_list_iteratePayload(&(nl->connections),NL_setFdSet);
		printf("\n------------\nUA_Stack_msgLoop - maxHandle=%d\n", nl->maxReaderHandle);

		// copy tv, some unixes do overwrite and return the remaining time
		struct timeval tmptv;
		memcpy(&tmptv,tv,sizeof(struct timeval));

		// and wait
		DBG_VERBOSE(printf("UA_Stack_msgLoop - enter select sec=%d,usec=%d\n",(UA_Int32) tmptv.tv_sec, (UA_Int32) tmptv.tv_usec));
		result = select(nl->maxReaderHandle + 1, &(nl->readerHandles), UA_NULL, UA_NULL, &tmptv);
		// printf("UA_Stack_msgLoop - leave select result=%d,sec=%d,usec=%d\n",result, (UA_Int32) tmptv.tv_sec, (UA_Int32) tmptv.tv_usec);

		// handle timeout (winsock: result=0, unix: result=0,errno=0||EAGAIN) 
		// and errors (winsock: result=SOCKET_ERROR (-1), unix: result = 0)
		if (result <= 0) {
#ifdef WIN32
			err = (result == SOCKET_ERROR) ? WSAGetLastError() : 0;
#else
			err = errno;
#endif
			switch (err) {
				// handle known errors
#ifdef WIN32
			case WSANOTINITIALISED:
			case WSAEFAULT:
			case WSAENETDOWN:
			case WSAEINVAL:
			case WSAEINTR:
			case WSAEINPROGRESS:
			case WSAENOTSOCK:
#else
			case EBADF:
			case EINTR:
			case EINVAL:
#endif
				// FIXME: handle errors
				printf("UA_Stack_msgLoop - result=%d, errno={%d,%s}\n", result, errno, strerror(errno));
				break;
			// we've got a timeout
#ifndef WIN32
	        case EAGAIN:
#endif
			default:
					DBG(printf("UA_Stack_msgLoop - result=%d, errno={%d,%s}\n", result, errno, strerror(errno)));
					DBG(printf("UA_Stack_msgLoop - call worker\n"));
					worker(arg);
					DBG(printf("UA_Stack_msgLoop - return from worker\n"));
			}
		} else { // activity on listener or client ports
			printf("UA_Stack_msgLoop - activities on %d handles\n",result);
			UA_list_iteratePayload(&(nl->connections),NL_checkFdSet);
		}
		// FIXME: Seems to be a conceptional flaw to call the worker here...
		// worker(arg);
	}
#ifdef WIN32
	WSACleanup();
#endif
	return UA_SUCCESS;
}
#endif /* MULTITASKING */


/** the tcp reader function */
void* NL_TCP_reader(NL_Connection *c) {

	UA_ByteString readBuffer;

	TL_Buffer localBuffers;
	UA_Int32 connectionState;

	UA_TL_Connection_getLocalConfig(c->connection, &localBuffers);
	UA_alloc((void**)&(readBuffer.data),localBuffers.recvBufferSize);

	UA_TL_Connection_getState(c->connection, &connectionState);
	if (connectionState  != CONNECTIONSTATE_CLOSE) {
		DBG_VERBOSE(printf("NL_TCP_reader - enter read\n"));

#ifdef WIN32
		readBuffer.length = recv(c->connectionHandle, readBuffer.data, localBuffers.recvBufferSize, 0);
#else
		readBuffer.length = read(c->connectionHandle, readBuffer.data, localBuffers.recvBufferSize);
#endif
		DBG_VERBOSE(printf("NL_TCP_reader - leave read\n"));

		DBG_VERBOSE(printf("NL_TCP_reader - src={%*.s}, ",c->connection.remoteEndpointUrl.length,c->connection.remoteEndpointUrl.data));
		DBG(UA_ByteString_printx("NL_TCP_reader - received=",&readBuffer));

		if (readBuffer.length  > 0) {

#ifdef DEBUG
#include "ua_transport_binary_secure.h"
			UA_UInt32 pos = 0;
			UA_OPCUATcpMessageHeader header;
			UA_OPCUATcpMessageHeader_decodeBinary(&readBuffer, &pos, &header);
			pos = 24;
			if(header.messageType == UA_MESSAGETYPE_MSG)
			{
				UA_NodeId serviceRequestType;
				UA_NodeId_decodeBinary(&readBuffer, &pos,&serviceRequestType);
				UA_NodeId_printf("Service Type\n",&serviceRequestType);
			}
#endif

			TL_Process((c->connection),&readBuffer);
		} else {
			perror("ERROR reading from socket1");
			UA_TL_Connection_setState(c->connection, CONNECTIONSTATE_CLOSE);
		}
	}

	UA_TL_Connection_getState(c->connection, &connectionState);
	printf("NL_TCP_reader - connectionState=%d\n",connectionState);
	if (connectionState == CONNECTIONSTATE_CLOSE) {
		// set connection's state to CONNECTIONSTATE_CLOSED and call callback to actually close
		UA_TL_Connection_close(c->connection);
#ifndef MULTITHREADING
		DBG_VERBOSE(printf("NL_TCP_reader - search element to remove\n"));
		UA_list_Element* lec = UA_list_search(&(c->networkLayer->connections),NL_ConnectionComparer,c);
		DBG_VERBOSE(printf("NL_TCP_reader - remove connection for handle=%d\n",((NL_Connection*)lec->payload)->connection.connectionHandle));
		UA_list_removeElement(lec,UA_NULL);
		DBG_VERBOSE(UA_list_iteratePayload(&(c->networkLayer->connections),NL_Connection_printf));
		UA_free(c);
#endif
	}
	UA_ByteString_deleteMembers(&readBuffer);
	return UA_NULL;
}

#ifdef MULTITHREADING
/** the tcp reader thread */
void* NL_TCP_readerThread(NL_Connection *c) {
	// just loop, NL_TCP_Reader will call the stack
	UA_Int32 connectionState;
	do {
{'fraction_non_alphanumeric': 0.1360349297312048, 'fraction_numerical': 0.012757538545504161, 'mean_word_length': 5.208809826344769, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 101, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6164969', 'n_tokens_mistral': 5719, 'n_tokens_neox': 5289, 'n_words': 1117}