WebSVN – HomeAutomation – Diff – /trunk/PcSoftware/Atom/src/Socket/asyncsocket.cpp


/***************************************************************************
 *  Copyright (C) December 6, 2008 by Mattias Runge               *
 *  mattias@runge.se                           *
 *  asyncsocket.cpp                      *
 *                                     *
 *  This program is free software; you can redistribute it and/or modify *
 *  it under the terms of the GNU General Public License as published by *
 *  the Free Software Foundation; either version 2 of the License, or   *
 *  (at your option) any later version.                  *
 *                                     *
 *  This program is distributed in the hope that it will be useful,    *
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of    *
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the     *
 *  GNU General Public License for more details.             *
 *                                     *
 *  You should have received a copy of the GNU General Public License   *
 *  along with this program; if not, write to the             *
 *  Free Software Foundation, Inc.,                    *
 *  59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.       *
 ***************************************************************************/
 
#include "socketeventcallback.h"
#include "../Logger/logger.h"
 
#include "asyncsocket.h"
 
int socketCount = 1;
 
AsyncSocket::AsyncSocket()
{
    myId = socketCount++;
    mySocket = -1;
    myReconnectTimeout = 0;
    myForceReconnect = false;
    myEventCallback = NULL;
}
 
AsyncSocket::~AsyncSocket()
{
    stop();
    silentClose();
}
 
void AsyncSocket::run()
{
//cout << "AsyncSocket::run - Thread started\n";
    // If connection is already up then we should not connect again
    if (!isConnected())
    {
        // If we have choosen to not use automatic reconnect do not start reconnect loop
        if (myReconnectTimeout == 0)
        {
            // Connect to somewhere
            connect();
        }
        else
        {
            // Start the reconnect loop
            reconnectLoop();
        }
    }
 
    bool loop = true;
    try
    {
        while (loop)
        {
            // If we have triggered a forced reconnect do it here
            if (myForceReconnect)
            {
                myForceReconnect = false;
                reconnectLoop();
            }
 
            // Receive data
            loop = receiveData();
        }
    }
    catch (SocketException *e)
    {
        cout << "DEBUG: socket got an exception: " << e->getDescription() << endl;
        // Something bad happend and we can not continue
        eventAdd(SocketEvent::TYPE_CONNECTION_DIED, e->getDescription());
    }
 
//cout << "AsyncSocket::run - Thread end?\n";
    // Clean up socket if we would want to restart
    silentClose();
}
 
bool AsyncSocket::receiveData()
{
    char buffer[MAXBUFFER + 1];
    memset(buffer, 0, MAXBUFFER + 1);
 
    //cout << "recv start" << endl;
    int status = ::recv(mySocket, buffer, MAXBUFFER, 0);
    //cout << "recv end" << endl;
   
    if (status == -1)
    {
        switch (errno)
        {
            case EAGAIN:
            throw new SocketException("The socket is marked non-blocking and the receive operation would block, or a receive timeout had been set and the timeout expired before data was received.");
 
            case EBADF:
            throw new SocketException("The argument s is an invalid descriptor.");
 
            case ECONNREFUSED:
            throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
 
            case EFAULT:
            throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
 
            case EINTR:
            throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
 
            case EINVAL:
            throw new SocketException("Invalid argument passed.");
 
            case ENOMEM:
            throw new SocketException("Could not allocate memory for recvmsg().");
 
            case ENOTCONN:
            throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");
 
            case ENOTSOCK:
            throw new SocketException("The argument s does not refer to a socket.");
 
            case ECONNRESET:
            eventAdd(SocketEvent::TYPE_CONNECTION_RESET);
 
            // If we have automatic reconnect we want to start it now
            if (myReconnectTimeout == 0)
            {
                reconnectLoop();
            }
            else
            {
                // Otherwise we would like to end the main loop
                return false;
            }
            break;
 
            default:
            throw new SocketException("Unknow exception: " + itos(errno));
            break;
        }
    }
    else if (status == 0)
    {
        // Remote host have done a normal shutdown
        eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
 
        if (myReconnectTimeout > 0)
        {
            reconnectLoop();
        }
        else
        {
            return false;
        }
    }
    else if (status > 0)
    {
        // We have received data
        eventAdd(SocketEvent::TYPE_DATA, buffer);
    }
 
    return true;
}
 
void AsyncSocket::silentClose()
{
    if (mySocket != -1)
    {
        ::close(mySocket);
        mySocket = -1;
    }
}
 
void AsyncSocket::close()
{
    silentClose();
    eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
}
 
void AsyncSocket::reconnectLoop()
{
    while (true)
    {
        try
        {
            connect();
            return;
        }
        catch (SocketException *e)
        {
            eventAdd(SocketEvent::TYPE_CONNECTION_FAILED, e->getDescription());
            eventAdd(SocketEvent::TYPE_WAITING_RECONNECT);
            sleep(myReconnectTimeout);
        }
    }
}
 
void AsyncSocket::create()
{
    silentClose();
 
    mySocket = ::socket(AF_INET, SOCK_STREAM, 0);
 
    int on = 1;
    int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
    if (status == -1)
    {
        silentClose();
        throw new SocketException("Create:Reuseaddress: " + itos(errno));
    }
}
 
void AsyncSocket::startListen()
{
    create();
 
    myAddressStruct.sin_family = AF_INET;
    myAddressStruct.sin_addr.s_addr = INADDR_ANY;
    myAddressStruct.sin_port = htons(myPort);
 
    int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));
 
    if (status == -1)
    {
        silentClose();
        switch (errno)
        {
            case EACCES:
            throw new SocketException("The address is protected, and the user is not the superuser.");
 
            case EADDRINUSE:
            throw new SocketException("The given address is already in use.");
 
            case EBADF:
            throw new SocketException("sockfd is not a valid descriptor.");
 
            case EINVAL:
            throw new SocketException("The socket is already bound to an address.");
 
            case ENOTSOCK:
            throw new SocketException("sockfd is a descriptor for a file, not a socket.");
 
            //case EACCES:
            //throw new SocketException("Search permission is denied on a component of the path prefix. (See also path_resolution(7).)");
 
            case EADDRNOTAVAIL:
            throw new SocketException("A nonexistent interface was requested or the requested address was not local.");
 
            case EFAULT:
            throw new SocketException("addr points outside the user's accessible address space.");
 
            //case EINVAL:
            //throw new SocketException("The addrlen is wrong, or the socket was not in the AF_UNIX family.");
 
            case ELOOP:
            throw new SocketException("Too many symbolic links were encountered in resolving addr.");
 
            case ENAMETOOLONG:
            throw new SocketException("addr is too long.");
 
            case ENOENT:
            throw new SocketException("The file does not exist.");
 
            case ENOMEM:
            throw new SocketException("Insufficient kernel memory was available.");
 
            case ENOTDIR:
            throw new SocketException("A component of the path prefix is not a directory.");
 
            case EROFS:
            throw new SocketException("The socket inode would reside on a read-only file system.");
 
            default:
            throw new SocketException("Unknow exception: " + itos(errno));
            break;
        }
    }
 
    status = ::listen(mySocket, MAXCONNECTIONS);
 
    if (status == -1)
    {
        silentClose();
        switch (errno)
        {
            case EADDRINUSE:
            throw new SocketException("Another socket is already listening on the same port.");
 
            case EBADF:
            throw new SocketException("The argument sockfd is not a valid descriptor.");
 
            case ENOTSOCK:
            throw new SocketException("The argument sockfd is not a socket.");
 
            case EOPNOTSUPP:
            throw new SocketException("The socket is not of a type that supports the listen() operation.");
 
            default:
            throw new SocketException("Unknow exception: " + itos(errno));
            break;
        }
    }
}
 
bool AsyncSocket::accept(AsyncSocket* newSocket)
{
    int addr_length = sizeof(myAddressStruct);
    int socket = ::accept(mySocket, (sockaddr*)&myAddressStruct, (socklen_t*)&addr_length);
 
    if (socket > 0)
    {
        newSocket->setSocket(socket);
        return true;
    }
 
    return false;
}
 
void AsyncSocket::connect()
{
//cout << "AsyncSocket::connect - eventAdd connecting\n";
    eventAdd(SocketEvent::TYPE_CONNECTING);
 
    create();
 
    memset(&myAddressStruct, 0, sizeof(myAddressStruct));
 
    myAddressStruct.sin_family = AF_INET;
    myAddressStruct.sin_port = htons(myPort);
 
    struct hostent *hptr = gethostbyname(myAddress.c_str());
    if (hptr == NULL)
    {
        silentClose();
        throw new SocketException("Connect: Could not resolv ip address");
    }
 
    memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);
 
    int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
 
    if (status == -1)
    {
        silentClose();
        switch (errno)
        {
            case EACCES:
            throw new SocketException("For Unix domain sockets, which are identified by pathname: Write permission is denied on the socket file, or search per- mission is denied for one of the directories in the path prefix. (See also path_resolution(7).)");
 
            case EPERM:
            throw new SocketException("The user tried to connect to a broadcast address without having the socket broadcast flag enabled or the connection request failed because of a local firewall rule.");
 
            case EADDRINUSE:
            throw new SocketException("Local address is already in use.");
 
            case EAFNOSUPPORT:
            throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");
 
            case EAGAIN:
            throw new SocketException("No more free local ports or insufficient entries in the routing cache. For AF_INET see the net.ipv4.ip_local_port_range sysctl in ip(7) on how to increase the number of local ports.");
 
            case EALREADY:
            throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");
 
            case EBADF:
            throw new SocketException("The file descriptor is not a valid index in the descriptor table.");
 
            case ECONNREFUSED:
            throw new SocketException("No-one listening on the remote address.");
 
            case EFAULT:
            throw new SocketException("The socket structure address is outside the user's address space.");
 
            case EINPROGRESS:
            throw new SocketException("The socket is non-blocking and the connection cannot be completed immediately. It is possible to select(2) or poll(2) for completion by selecting the socket for writing. After select(2) indicates writability, use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to determine whether connect() completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR is one of the usual error codes listed here, explaining the reason for the failure).");
 
            case EINTR:
            throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");
 
            case EISCONN:
            throw new SocketException("The socket is already connected.");
 
            case ENETUNREACH:
            throw new SocketException("Network is unreachable.");
 
            case ENOTSOCK:
            throw new SocketException("The file descriptor is not associated with a socket.");
 
            case ETIMEDOUT:
            throw new SocketException("Timeout while attempting connection. The server may be too busy to accept new connections. Note that for IP sockets the timeout may be very long when syncookies are enabled on the server.");
 
            default:
            throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
        }
    }
 
//cout << "AsyncSocket::connect - eventAdd connected\n";
    eventAdd(SocketEvent::TYPE_CONNECTED);
}
 
void AsyncSocket::sendData(string data)
{
//cout << "AsyncSocket::sendData - got data to send: ";
//cout << data;
//cout << "\n";
    mySendMutex.lock();
 
    int status = ::send(mySocket, data.c_str(), data.size(), MSG_NOSIGNAL);
//Logger::getInstance().add("Sent: \"" + data + "\" status was " + itos(status) + "\n");
    mySendMutex.unlock();
 
    //Logger &log = Logger::getInstance();
    //log.add("Sent: \"" + data + "\" status was " + itos(status) + "\n");
 
    if (status == -1)
    {
        switch (errno)
        {
            case EACCES:
            throw new SocketException("(For Unix domain sockets, which are identified by pathname) Write permission is denied on the destination socket file, or search permission is denied for one of the directories the path prefix. (See path_resolution(7).)");
 
            case EAGAIN:
            throw new SocketException("The socket is marked non-blocking and the requested operation would block.");
 
            case EBADF:
            throw new SocketException("An invalid descriptor was specified.");
 
            case ECONNRESET:
            throw new SocketException("Connection reset by peer.");
 
            case EDESTADDRREQ:
            throw new SocketException("The socket is not connection-mode, and no peer address is set.");
 
            case EFAULT:
            throw new SocketException("An invalid user space address was specified for an argument.");
 
            case EINTR:
            throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");
 
            case EINVAL:
            throw new SocketException("1Invalid argument passed.");
 
            case EISCONN:
            throw new SocketException("The connection-mode socket was connected already but a recipient was specified. (Now either this error is returned, or the recipient specification is ignored.)");
 
            case EMSGSIZE:
            throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");
 
            case ENOBUFS:
            throw new SocketException("The output queue for a network interface was full. This generally indicates that the interface has stopped sending, but may be caused by transient congestion. (Normally, this does not occur in Linux. Packets are just silently dropped when a device queue overflows.)");
 
            case ENOMEM:
            throw new SocketException("No memory available.");
 
            case ENOTCONN:
            throw new SocketException("The socket is not connected, and no target has been given.");
 
            case ENOTSOCK:
            throw new SocketException("The argument s is not a socket.");
 
            case EOPNOTSUPP:
            throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");
 
            case EPIPE:
            close();
            stop();
            //throw new SocketException("The local end has been shut down on a connection oriented socket. In this case the process will also receive a SIGPIPE unless MSG_NOSIGNAL is set.");
            break;
 
            default:
            throw new SocketException("Unknow exception: " + itos(errno));
        }
    }
}
 
void AsyncSocket::eventAdd(unsigned int eventType)
{
    eventAdd(eventType, "");
}
 
void AsyncSocket::eventAdd(unsigned int eventType, string eventData)
{
    SocketEvent socketEvent(eventType, eventData);
 
    if (myEventCallback == NULL)
    {
        myEventQueue.push(socketEvent);
        myEventSemaphore.broadcast();
    }
    else
    {
        myEventCallback->handleEvent(myId, socketEvent);
    }
}
 
void AsyncSocket::eventSetCallback(SocketEventCallback* eventCallback)
{
    myEventCallback = eventCallback;
}
 

Rev 1124	Rev 1203
1	/***************************************************************************	1	/***************************************************************************
2	* Copyright (C) December 6, 2008 by Mattias Runge *	2	* Copyright (C) December 6, 2008 by Mattias Runge *
3	* mattias@runge.se *	3	* mattias@runge.se *
4	* asyncsocket.cpp *	4	* asyncsocket.cpp *
5	* *	5	* *
6	* This program is free software; you can redistribute it and/or modify *	6	* This program is free software; you can redistribute it and/or modify *
7	* it under the terms of the GNU General Public License as published by *	7	* it under the terms of the GNU General Public License as published by *
8	* the Free Software Foundation; either version 2 of the License, or *	8	* the Free Software Foundation; either version 2 of the License, or *
9	* (at your option) any later version. *	9	* (at your option) any later version. *
10	* *	10	* *
11	* This program is distributed in the hope that it will be useful, *	11	* This program is distributed in the hope that it will be useful, *
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of *	12	* but WITHOUT ANY WARRANTY; without even the implied warranty of *
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *	13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
14	* GNU General Public License for more details. *	14	* GNU General Public License for more details. *
15	* *	15	* *
16	* You should have received a copy of the GNU General Public License *	16	* You should have received a copy of the GNU General Public License *
17	* along with this program; if not, write to the *	17	* along with this program; if not, write to the *
18	* Free Software Foundation, Inc., *	18	* Free Software Foundation, Inc., *
19	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *	19	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
20	***************************************************************************/	20	***************************************************************************/
21		21
22	#include "socketeventcallback.h"	22	#include "socketeventcallback.h"
23	#include "../Logger/logger.h"	23	#include "../Logger/logger.h"
24		24
25	#include "asyncsocket.h"	25	#include "asyncsocket.h"
26		26
27	int socketCount = 1;	27	int socketCount = 1;
28		28
29	AsyncSocket::AsyncSocket()	29	AsyncSocket::AsyncSocket()
30	{	30	{
31	myId = socketCount++;	31	myId = socketCount++;
32	mySocket = -1;	32	mySocket = -1;
33	myReconnectTimeout = 0;	33	myReconnectTimeout = 0;
34	myForceReconnect = false;	34	myForceReconnect = false;
35	myEventCallback = NULL;	35	myEventCallback = NULL;
36	}	36	}
37		37
38	AsyncSocket::~AsyncSocket()	38	AsyncSocket::~AsyncSocket()
39	{	39	{
40	stop();	40	stop();
41	silentClose();	41	silentClose();
42	}	42	}
43		43
44	void AsyncSocket::run()	44	void AsyncSocket::run()
45	{	45	{
-		46	//cout << "AsyncSocket::run - Thread started\n";
46	// If connection is already up then we should not connect again	47	// If connection is already up then we should not connect again
47	if (!isConnected())	48	if (!isConnected())
48	{	49	{
49	// If we have choosen to not use automatic reconnect do not start reconnect loop	50	// If we have choosen to not use automatic reconnect do not start reconnect loop
50	if (myReconnectTimeout == 0)	51	if (myReconnectTimeout == 0)
51	{	52	{
52	// Connect to somewhere	53	// Connect to somewhere
53	connect();	54	connect();
54	}	55	}
55	else	56	else
56	{	57	{
57	// Start the reconnect loop	58	// Start the reconnect loop
58	reconnectLoop();	59	reconnectLoop();
59	}	60	}
60	}	61	}
61		62
62	bool loop = true;	63	bool loop = true;
63	try	64	try
64	{	65	{
65	while (loop)	66	while (loop)
66	{	67	{
67	// If we have triggered a forced reconnect do it here	68	// If we have triggered a forced reconnect do it here
68	if (myForceReconnect)	69	if (myForceReconnect)
69	{	70	{
70	myForceReconnect = false;	71	myForceReconnect = false;
71	reconnectLoop();	72	reconnectLoop();
72	}	73	}
73		74
74	// Receive data	75	// Receive data
75	loop = receiveData();	76	loop = receiveData();
76	}	77	}
77	}	78	}
78	catch (SocketException *e)	79	catch (SocketException *e)
79	{	80	{
80	cout << "DEBUG: socket got an exception: " << e->getDescription() << endl;	81	cout << "DEBUG: socket got an exception: " << e->getDescription() << endl;
81	// Something bad happend and we can not continue	82	// Something bad happend and we can not continue
82	eventAdd(SocketEvent::TYPE_CONNECTION_DIED, e->getDescription());	83	eventAdd(SocketEvent::TYPE_CONNECTION_DIED, e->getDescription());
83	}	84	}
84		85
-		86	//cout << "AsyncSocket::run - Thread end?\n";
85	// Clean up socket if we would want to restart	87	// Clean up socket if we would want to restart
86	silentClose();	88	silentClose();
87	}	89	}
88		90
89	bool AsyncSocket::receiveData()	91	bool AsyncSocket::receiveData()
90	{	92	{
91	char buffer[MAXBUFFER + 1];	93	char buffer[MAXBUFFER + 1];
92	memset(buffer, 0, MAXBUFFER + 1);	94	memset(buffer, 0, MAXBUFFER + 1);
93		95
94	//cout << "recv start" << endl;	96	//cout << "recv start" << endl;
95	int status = ::recv(mySocket, buffer, MAXBUFFER, 0);	97	int status = ::recv(mySocket, buffer, MAXBUFFER, 0);
96	//cout << "recv end" << endl;	98	//cout << "recv end" << endl;
97		99
98	if (status == -1)	100	if (status == -1)
99	{	101	{
100	switch (errno)	102	switch (errno)
101	{	103	{
102	case EAGAIN:	104	case EAGAIN:
103	throw new SocketException("The socket is marked non-blocking and the receive operation would block, or a receive timeout had been set and the timeout expired before data was received.");	105	throw new SocketException("The socket is marked non-blocking and the receive operation would block, or a receive timeout had been set and the timeout expired before data was received.");
104		106
105	case EBADF:	107	case EBADF:
106	throw new SocketException("The argument s is an invalid descriptor.");	108	throw new SocketException("The argument s is an invalid descriptor.");
107		109
108	case ECONNREFUSED:	110	case ECONNREFUSED:
109	throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");	111	throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
110		112
111	case EFAULT:	113	case EFAULT:
112	throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");	114	throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
113		115
114	case EINTR:	116	case EINTR:
115	throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");	117	throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
116		118
117	case EINVAL:	119	case EINVAL:
118	throw new SocketException("Invalid argument passed.");	120	throw new SocketException("Invalid argument passed.");
119		121
120	case ENOMEM:	122	case ENOMEM:
121	throw new SocketException("Could not allocate memory for recvmsg().");	123	throw new SocketException("Could not allocate memory for recvmsg().");
122		124
123	case ENOTCONN:	125	case ENOTCONN:
124	throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");	126	throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");
125		127
126	case ENOTSOCK:	128	case ENOTSOCK:
127	throw new SocketException("The argument s does not refer to a socket.");	129	throw new SocketException("The argument s does not refer to a socket.");
128		130
129	case ECONNRESET:	131	case ECONNRESET:
130	eventAdd(SocketEvent::TYPE_CONNECTION_RESET);	132	eventAdd(SocketEvent::TYPE_CONNECTION_RESET);
131		133
132	// If we have automatic reconnect we want to start it now	134	// If we have automatic reconnect we want to start it now
133	if (myReconnectTimeout == 0)	135	if (myReconnectTimeout == 0)
134	{	136	{
135	reconnectLoop();	137	reconnectLoop();
136	}	138	}
137	else	139	else
138	{	140	{
139	// Otherwise we would like to end the main loop	141	// Otherwise we would like to end the main loop
140	return false;	142	return false;
141	}	143	}
142	break;	144	break;
143		145
144	default:	146	default:
145	throw new SocketException("Unknow exception: " + itos(errno));	147	throw new SocketException("Unknow exception: " + itos(errno));
146	break;	148	break;
147	}	149	}
148	}	150	}
149	else if (status == 0)	151	else if (status == 0)
150	{	152	{
151	// Remote host have done a normal shutdown	153	// Remote host have done a normal shutdown
152	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);	154	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
153		155
154	if (myReconnectTimeout > 0)	156	if (myReconnectTimeout > 0)
155	{	157	{
156	reconnectLoop();	158	reconnectLoop();
157	}	159	}
158	else	160	else
159	{	161	{
160	return false;	162	return false;
161	}	163	}
162	}	164	}
163	else if (status > 0)	165	else if (status > 0)
164	{	166	{
165	// We have received data	167	// We have received data
166	eventAdd(SocketEvent::TYPE_DATA, buffer);	168	eventAdd(SocketEvent::TYPE_DATA, buffer);
167	}	169	}
168		170
169	return true;	171	return true;
170	}	172	}
171		173
172	void AsyncSocket::silentClose()	174	void AsyncSocket::silentClose()
173	{	175	{
174	if (mySocket != -1)	176	if (mySocket != -1)
175	{	177	{
176	::close(mySocket);	178	::close(mySocket);
177	mySocket = -1;	179	mySocket = -1;
178	}	180	}
179	}	181	}
180		182
181	void AsyncSocket::close()	183	void AsyncSocket::close()
182	{	184	{
183	silentClose();	185	silentClose();
184	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);	186	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
185	}	187	}
186		188
187	void AsyncSocket::reconnectLoop()	189	void AsyncSocket::reconnectLoop()
188	{	190	{
189	while (true)	191	while (true)
190	{	192	{
191	try	193	try
192	{	194	{
193	connect();	195	connect();
194	return;	196	return;
195	}	197	}
196	catch (SocketException *e)	198	catch (SocketException *e)
197	{	199	{
198	eventAdd(SocketEvent::TYPE_CONNECTION_FAILED, e->getDescription());	200	eventAdd(SocketEvent::TYPE_CONNECTION_FAILED, e->getDescription());
199	eventAdd(SocketEvent::TYPE_WAITING_RECONNECT);	201	eventAdd(SocketEvent::TYPE_WAITING_RECONNECT);
200	sleep(myReconnectTimeout);	202	sleep(myReconnectTimeout);
201	}	203	}
202	}	204	}
203	}	205	}
204		206
205	void AsyncSocket::create()	207	void AsyncSocket::create()
206	{	208	{
207	silentClose();	209	silentClose();
208		210
209	mySocket = ::socket(AF_INET, SOCK_STREAM, 0);	211	mySocket = ::socket(AF_INET, SOCK_STREAM, 0);
210		212
211	int on = 1;	213	int on = 1;
212	int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));	214	int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
213	if (status == -1)	215	if (status == -1)
214	{	216	{
215	silentClose();	217	silentClose();
216	throw new SocketException("Create:Reuseaddress: " + itos(errno));	218	throw new SocketException("Create:Reuseaddress: " + itos(errno));
217	}	219	}
218	}	220	}
219		221
220	void AsyncSocket::startListen()	222	void AsyncSocket::startListen()
221	{	223	{
222	create();	224	create();
223		225
224	myAddressStruct.sin_family = AF_INET;	226	myAddressStruct.sin_family = AF_INET;
225	myAddressStruct.sin_addr.s_addr = INADDR_ANY;	227	myAddressStruct.sin_addr.s_addr = INADDR_ANY;
226	myAddressStruct.sin_port = htons(myPort);	228	myAddressStruct.sin_port = htons(myPort);
227		229
228	int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));	230	int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));
229		231
230	if (status == -1)	232	if (status == -1)
231	{	233	{
232	silentClose();	234	silentClose();
233	switch (errno)	235	switch (errno)
234	{	236	{
235	case EACCES:	237	case EACCES:
236	throw new SocketException("The address is protected, and the user is not the superuser.");	238	throw new SocketException("The address is protected, and the user is not the superuser.");
237		239
238	case EADDRINUSE:	240	case EADDRINUSE:
239	throw new SocketException("The given address is already in use.");	241	throw new SocketException("The given address is already in use.");
240		242
241	case EBADF:	243	case EBADF:
242	throw new SocketException("sockfd is not a valid descriptor.");	244	throw new SocketException("sockfd is not a valid descriptor.");
243		245
244	case EINVAL:	246	case EINVAL:
245	throw new SocketException("The socket is already bound to an address.");	247	throw new SocketException("The socket is already bound to an address.");
246		248
247	case ENOTSOCK:	249	case ENOTSOCK:
248	throw new SocketException("sockfd is a descriptor for a file, not a socket.");	250	throw new SocketException("sockfd is a descriptor for a file, not a socket.");
249		251
250	//case EACCES:	252	//case EACCES:
251	//throw new SocketException("Search permission is denied on a component of the path prefix. (See also path_resolution(7).)");	253	//throw new SocketException("Search permission is denied on a component of the path prefix. (See also path_resolution(7).)");
252		254
253	case EADDRNOTAVAIL:	255	case EADDRNOTAVAIL:
254	throw new SocketException("A nonexistent interface was requested or the requested address was not local.");	256	throw new SocketException("A nonexistent interface was requested or the requested address was not local.");
255		257
256	case EFAULT:	258	case EFAULT:
257	throw new SocketException("addr points outside the user's accessible address space.");	259	throw new SocketException("addr points outside the user's accessible address space.");
258		260
259	//case EINVAL:	261	//case EINVAL:
260	//throw new SocketException("The addrlen is wrong, or the socket was not in the AF_UNIX family.");	262	//throw new SocketException("The addrlen is wrong, or the socket was not in the AF_UNIX family.");
261		263
262	case ELOOP:	264	case ELOOP:
263	throw new SocketException("Too many symbolic links were encountered in resolving addr.");	265	throw new SocketException("Too many symbolic links were encountered in resolving addr.");
264		266
265	case ENAMETOOLONG:	267	case ENAMETOOLONG:
266	throw new SocketException("addr is too long.");	268	throw new SocketException("addr is too long.");
267		269
268	case ENOENT:	270	case ENOENT:
269	throw new SocketException("The file does not exist.");	271	throw new SocketException("The file does not exist.");
270		272
271	case ENOMEM:	273	case ENOMEM:
272	throw new SocketException("Insufficient kernel memory was available.");	274	throw new SocketException("Insufficient kernel memory was available.");
273		275
274	case ENOTDIR:	276	case ENOTDIR:
275	throw new SocketException("A component of the path prefix is not a directory.");	277	throw new SocketException("A component of the path prefix is not a directory.");
276		278
277	case EROFS:	279	case EROFS:
278	throw new SocketException("The socket inode would reside on a read-only file system.");	280	throw new SocketException("The socket inode would reside on a read-only file system.");
279		281
280	default:	282	default:
281	throw new SocketException("Unknow exception: " + itos(errno));	283	throw new SocketException("Unknow exception: " + itos(errno));
282	break;	284	break;
283	}	285	}
284	}	286	}
285		287
286	status = ::listen(mySocket, MAXCONNECTIONS);	288	status = ::listen(mySocket, MAXCONNECTIONS);
287		289
288	if (status == -1)	290	if (status == -1)
289	{	291	{
290	silentClose();	292	silentClose();
291	switch (errno)	293	switch (errno)
292	{	294	{
293	case EADDRINUSE:	295	case EADDRINUSE:
294	throw new SocketException("Another socket is already listening on the same port.");	296	throw new SocketException("Another socket is already listening on the same port.");
295		297
296	case EBADF:	298	case EBADF:
297	throw new SocketException("The argument sockfd is not a valid descriptor.");	299	throw new SocketException("The argument sockfd is not a valid descriptor.");
298		300
299	case ENOTSOCK:	301	case ENOTSOCK:
300	throw new SocketException("The argument sockfd is not a socket.");	302	throw new SocketException("The argument sockfd is not a socket.");
301		303
302	case EOPNOTSUPP:	304	case EOPNOTSUPP:
303	throw new SocketException("The socket is not of a type that supports the listen() operation.");	305	throw new SocketException("The socket is not of a type that supports the listen() operation.");
304		306
305	default:	307	default:
306	throw new SocketException("Unknow exception: " + itos(errno));	308	throw new SocketException("Unknow exception: " + itos(errno));
307	break;	309	break;
308	}	310	}
309	}	311	}
310	}	312	}
311		313
312	bool AsyncSocket::accept(AsyncSocket* newSocket)	314	bool AsyncSocket::accept(AsyncSocket* newSocket)
313	{	315	{
314	int addr_length = sizeof(myAddressStruct);	316	int addr_length = sizeof(myAddressStruct);
315	int socket = ::accept(mySocket, (sockaddr)&myAddressStruct, (socklen_t)&addr_length);	317	int socket = ::accept(mySocket, (sockaddr)&myAddressStruct, (socklen_t)&addr_length);
316		318
317	if (socket > 0)	319	if (socket > 0)
318	{	320	{
319	newSocket->setSocket(socket);	321	newSocket->setSocket(socket);
320	return true;	322	return true;
321	}	323	}
322		324
323	return false;	325	return false;
324	}	326	}
325		327
326	void AsyncSocket::connect()	328	void AsyncSocket::connect()
327	{	329	{
-		330	//cout << "AsyncSocket::connect - eventAdd connecting\n";
328	eventAdd(SocketEvent::TYPE_CONNECTING);	331	eventAdd(SocketEvent::TYPE_CONNECTING);
329		332
330	create();	333	create();
331		334
332	memset(&myAddressStruct, 0, sizeof(myAddressStruct));	335	memset(&myAddressStruct, 0, sizeof(myAddressStruct));
333		336
334	myAddressStruct.sin_family = AF_INET;	337	myAddressStruct.sin_family = AF_INET;
335	myAddressStruct.sin_port = htons(myPort);	338	myAddressStruct.sin_port = htons(myPort);
336		339
337	struct hostent *hptr = gethostbyname(myAddress.c_str());	340	struct hostent *hptr = gethostbyname(myAddress.c_str());
338	if (hptr == NULL)	341	if (hptr == NULL)
339	{	342	{
340	silentClose();	343	silentClose();
341	throw new SocketException("Connect: Could not resolv ip address");	344	throw new SocketException("Connect: Could not resolv ip address");
342	}	345	}
343		346
344	memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);	347	memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);
345		348
346	int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));	349	int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
347		350
348	if (status == -1)	351	if (status == -1)
349	{	352	{
350	silentClose();	353	silentClose();
351	switch (errno)	354	switch (errno)
352	{	355	{
353	case EACCES:	356	case EACCES:
354	throw new SocketException("For Unix domain sockets, which are identified by pathname: Write permission is denied on the socket file, or search per- mission is denied for one of the directories in the path prefix. (See also path_resolution(7).)");	357	throw new SocketException("For Unix domain sockets, which are identified by pathname: Write permission is denied on the socket file, or search per- mission is denied for one of the directories in the path prefix. (See also path_resolution(7).)");
355		358
356	case EPERM:	359	case EPERM:
357	throw new SocketException("The user tried to connect to a broadcast address without having the socket broadcast flag enabled or the connection request failed because of a local firewall rule.");	360	throw new SocketException("The user tried to connect to a broadcast address without having the socket broadcast flag enabled or the connection request failed because of a local firewall rule.");
358		361
359	case EADDRINUSE:	362	case EADDRINUSE:
360	throw new SocketException("Local address is already in use.");	363	throw new SocketException("Local address is already in use.");
361		364
362	case EAFNOSUPPORT:	365	case EAFNOSUPPORT:
363	throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");	366	throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");
364		367
365	case EAGAIN:	368	case EAGAIN:
366	throw new SocketException("No more free local ports or insufficient entries in the routing cache. For AF_INET see the net.ipv4.ip_local_port_range sysctl in ip(7) on how to increase the number of local ports.");	369	throw new SocketException("No more free local ports or insufficient entries in the routing cache. For AF_INET see the net.ipv4.ip_local_port_range sysctl in ip(7) on how to increase the number of local ports.");
367		370
368	case EALREADY:	371	case EALREADY:
369	throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");	372	throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");
370		373
371	case EBADF:	374	case EBADF:
372	throw new SocketException("The file descriptor is not a valid index in the descriptor table.");	375	throw new SocketException("The file descriptor is not a valid index in the descriptor table.");
373		376
374	case ECONNREFUSED:	377	case ECONNREFUSED:
375	throw new SocketException("No-one listening on the remote address.");	378	throw new SocketException("No-one listening on the remote address.");
376		379
377	case EFAULT:	380	case EFAULT:
378	throw new SocketException("The socket structure address is outside the user's address space.");	381	throw new SocketException("The socket structure address is outside the user's address space.");
379		382
380	case EINPROGRESS:	383	case EINPROGRESS:
381	throw new SocketException("The socket is non-blocking and the connection cannot be completed immediately. It is possible to select(2) or poll(2) for completion by selecting the socket for writing. After select(2) indicates writability, use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to determine whether connect() completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR is one of the usual error codes listed here, explaining the reason for the failure).");	384	throw new SocketException("The socket is non-blocking and the connection cannot be completed immediately. It is possible to select(2) or poll(2) for completion by selecting the socket for writing. After select(2) indicates writability, use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to determine whether connect() completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR is one of the usual error codes listed here, explaining the reason for the failure).");
382		385
383	case EINTR:	386	case EINTR:
384	throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");	387	throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");
385		388
386	case EISCONN:	389	case EISCONN:
387	throw new SocketException("The socket is already connected.");	390	throw new SocketException("The socket is already connected.");
388		391
389	case ENETUNREACH:	392	case ENETUNREACH:
390	throw new SocketException("Network is unreachable.");	393	throw new SocketException("Network is unreachable.");
391		394
392	case ENOTSOCK:	395	case ENOTSOCK:
393	throw new SocketException("The file descriptor is not associated with a socket.");	396	throw new SocketException("The file descriptor is not associated with a socket.");
394		397
395	case ETIMEDOUT:	398	case ETIMEDOUT:
396	throw new SocketException("Timeout while attempting connection. The server may be too busy to accept new connections. Note that for IP sockets the timeout may be very long when syncookies are enabled on the server.");	399	throw new SocketException("Timeout while attempting connection. The server may be too busy to accept new connections. Note that for IP sockets the timeout may be very long when syncookies are enabled on the server.");
397		400
398	default:	401	default:
399	throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));	402	throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
400	}	403	}
401	}	404	}
402		405
-		406	//cout << "AsyncSocket::connect - eventAdd connected\n";
403	eventAdd(SocketEvent::TYPE_CONNECTED);	407	eventAdd(SocketEvent::TYPE_CONNECTED);
404	}	408	}
405		409
406	void AsyncSocket::sendData(string data)	410	void AsyncSocket::sendData(string data)
407	{	411	{
-		412	//cout << "AsyncSocket::sendData - got data to send: ";
-		413	//cout << data;
-		414	//cout << "\n";
408	mySendMutex.lock();	415	mySendMutex.lock();
409		416
410	int status = ::send(mySocket, data.c_str(), data.size(), MSG_NOSIGNAL);	417	int status = ::send(mySocket, data.c_str(), data.size(), MSG_NOSIGNAL);
411	//Logger::getInstance().add("Sent: \"" + data + "\" status was " + itos(status) + "\n");	418	//Logger::getInstance().add("Sent: \"" + data + "\" status was " + itos(status) + "\n");
412	mySendMutex.unlock();	419	mySendMutex.unlock();
413		420
414	//Logger &log = Logger::getInstance();	421	//Logger &log = Logger::getInstance();
415	//log.add("Sent: \"" + data + "\" status was " + itos(status) + "\n");	422	//log.add("Sent: \"" + data + "\" status was " + itos(status) + "\n");
416		423
417	if (status == -1)	424	if (status == -1)
418	{	425	{
419	switch (errno)	426	switch (errno)
420	{	427	{
421	case EACCES:	428	case EACCES:
422	throw new SocketException("(For Unix domain sockets, which are identified by pathname) Write permission is denied on the destination socket file, or search permission is denied for one of the directories the path prefix. (See path_resolution(7).)");	429	throw new SocketException("(For Unix domain sockets, which are identified by pathname) Write permission is denied on the destination socket file, or search permission is denied for one of the directories the path prefix. (See path_resolution(7).)");
423		430
424	case EAGAIN:	431	case EAGAIN:
425	throw new SocketException("The socket is marked non-blocking and the requested operation would block.");	432	throw new SocketException("The socket is marked non-blocking and the requested operation would block.");
426		433
427	case EBADF:	434	case EBADF:
428	throw new SocketException("An invalid descriptor was specified.");	435	throw new SocketException("An invalid descriptor was specified.");
429		436
430	case ECONNRESET:	437	case ECONNRESET:
431	throw new SocketException("Connection reset by peer.");	438	throw new SocketException("Connection reset by peer.");
432		439
433	case EDESTADDRREQ:	440	case EDESTADDRREQ:
434	throw new SocketException("The socket is not connection-mode, and no peer address is set.");	441	throw new SocketException("The socket is not connection-mode, and no peer address is set.");
435		442
436	case EFAULT:	443	case EFAULT:
437	throw new SocketException("An invalid user space address was specified for an argument.");	444	throw new SocketException("An invalid user space address was specified for an argument.");
438		445
439	case EINTR:	446	case EINTR:
440	throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");	447	throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");
441		448
442	case EINVAL:	449	case EINVAL:
443	throw new SocketException("1Invalid argument passed.");	450	throw new SocketException("1Invalid argument passed.");
444		451
445	case EISCONN:	452	case EISCONN:
446	throw new SocketException("The connection-mode socket was connected already but a recipient was specified. (Now either this error is returned, or the recipient specification is ignored.)");	453	throw new SocketException("The connection-mode socket was connected already but a recipient was specified. (Now either this error is returned, or the recipient specification is ignored.)");
447		454
448	case EMSGSIZE:	455	case EMSGSIZE:
449	throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");	456	throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");
450		457
451	case ENOBUFS:	458	case ENOBUFS:
452	throw new SocketException("The output queue for a network interface was full. This generally indicates that the interface has stopped sending, but may be caused by transient congestion. (Normally, this does not occur in Linux. Packets are just silently dropped when a device queue overflows.)");	459	throw new SocketException("The output queue for a network interface was full. This generally indicates that the interface has stopped sending, but may be caused by transient congestion. (Normally, this does not occur in Linux. Packets are just silently dropped when a device queue overflows.)");
453		460
454	case ENOMEM:	461	case ENOMEM:
455	throw new SocketException("No memory available.");	462	throw new SocketException("No memory available.");
456		463
457	case ENOTCONN:	464	case ENOTCONN:
458	throw new SocketException("The socket is not connected, and no target has been given.");	465	throw new SocketException("The socket is not connected, and no target has been given.");
459		466
460	case ENOTSOCK:	467	case ENOTSOCK:
461	throw new SocketException("The argument s is not a socket.");	468	throw new SocketException("The argument s is not a socket.");
462		469
463	case EOPNOTSUPP:	470	case EOPNOTSUPP:
464	throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");	471	throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");
465		472
466	case EPIPE:	473	case EPIPE:
467	close();	474	close();
468	stop();	475	stop();
469	//throw new SocketException("The local end has been shut down on a connection oriented socket. In this case the process will also receive a SIGPIPE unless MSG_NOSIGNAL is set.");	476	//throw new SocketException("The local end has been shut down on a connection oriented socket. In this case the process will also receive a SIGPIPE unless MSG_NOSIGNAL is set.");
470	break;	477	break;
471		478
472	default:	479	default:
473	throw new SocketException("Unknow exception: " + itos(errno));	480	throw new SocketException("Unknow exception: " + itos(errno));
474	}	481	}
475	}	482	}
476	}	483	}
477		484
478	void AsyncSocket::eventAdd(unsigned int eventType)	485	void AsyncSocket::eventAdd(unsigned int eventType)
479	{	486	{
480	eventAdd(eventType, "");	487	eventAdd(eventType, "");
481	}	488	}
482		489
483	void AsyncSocket::eventAdd(unsigned int eventType, string eventData)	490	void AsyncSocket::eventAdd(unsigned int eventType, string eventData)
484	{	491	{
485	SocketEvent socketEvent(eventType, eventData);	492	SocketEvent socketEvent(eventType, eventData);
486		493
487	if (myEventCallback == NULL)	494	if (myEventCallback == NULL)
488	{	495	{
489	myEventQueue.push(socketEvent);	496	myEventQueue.push(socketEvent);
490	myEventSemaphore.broadcast();	497	myEventSemaphore.broadcast();
491	}	498	}
492	else	499	else
493	{	500	{

Subversion Repositories HomeAutomation

(root)/trunk/PcSoftware/Atom/src/Socket/asyncsocket.cpp – Rev 1124 → 1203