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()
{
    // 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());
    }
 
    // 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()
{
    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));
        }
    }
 
    eventAdd(SocketEvent::TYPE_CONNECTED);
}
 
void AsyncSocket::sendData(string data)
{
    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 1122	Rev 1124
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	// If connection is already up then we should not connect again	46	// If connection is already up then we should not connect again
47	if (!isConnected())	47	if (!isConnected())
48	{	48	{
49	// If we have choosen to not use automatic reconnect do not start reconnect loop	49	// If we have choosen to not use automatic reconnect do not start reconnect loop
50	if (myReconnectTimeout == 0)	50	if (myReconnectTimeout == 0)
51	{	51	{
52	// Connect to somewhere	52	// Connect to somewhere
53	connect();	53	connect();
54	}	54	}
55	else	55	else
56	{	56	{
57	// Start the reconnect loop	57	// Start the reconnect loop
58	reconnectLoop();	58	reconnectLoop();
59	}	59	}
60	}	60	}
61		61
62	bool loop = true;	62	bool loop = true;
63	try	63	try
64	{	64	{
65	while (loop)	65	while (loop)
66	{	66	{
67	// If we have triggered a forced reconnect do it here	67	// If we have triggered a forced reconnect do it here
68	if (myForceReconnect)	68	if (myForceReconnect)
69	{	69	{
70	myForceReconnect = false;	70	myForceReconnect = false;
71	reconnectLoop();	71	reconnectLoop();
72	}	72	}
73		73
74	// Receive data	74	// Receive data
75	loop = receiveData();	75	loop = receiveData();
76	}	76	}
77	}	77	}
78	catch (SocketException *e)	78	catch (SocketException *e)
79	{	79	{
-		80	cout << "DEBUG: socket got an exception: " << e->getDescription() << endl;
80	// Something bad happend and we can not continue	81	// Something bad happend and we can not continue
81	eventAdd(SocketEvent::TYPE_CONNECTION_DIED, e->getDescription());	82	eventAdd(SocketEvent::TYPE_CONNECTION_DIED, e->getDescription());
82	}	83	}
83		84
84	// Clean up socket if we would want to restart	85	// Clean up socket if we would want to restart
85	silentClose();	86	silentClose();
86	}	87	}
87		88
88	bool AsyncSocket::receiveData()	89	bool AsyncSocket::receiveData()
89	{	90	{
90	char buffer[MAXBUFFER + 1];	91	char buffer[MAXBUFFER + 1];
91	memset(buffer, 0, MAXBUFFER + 1);	92	memset(buffer, 0, MAXBUFFER + 1);
92		93
-		94	//cout << "recv start" << endl;
93	int status = ::recv(mySocket, buffer, MAXBUFFER, 0);	95	int status = ::recv(mySocket, buffer, MAXBUFFER, 0);
-		96	//cout << "recv end" << endl;
94		97
95	if (status == -1)	98	if (status == -1)
96	{	99	{
97	switch (errno)	100	switch (errno)
98	{	101	{
99	case EAGAIN:	102	case EAGAIN:
100	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.");	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.");
101		104
102	case EBADF:	105	case EBADF:
103	throw new SocketException("The argument s is an invalid descriptor.");	106	throw new SocketException("The argument s is an invalid descriptor.");
104		107
105	case ECONNREFUSED:	108	case ECONNREFUSED:
106	throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");	109	throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
107		110
108	case EFAULT:	111	case EFAULT:
109	throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");	112	throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
110		113
111	case EINTR:	114	case EINTR:
112	throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");	115	throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
113		116
114	case EINVAL:	117	case EINVAL:
115	throw new SocketException("Invalid argument passed.");	118	throw new SocketException("Invalid argument passed.");
116		119
117	case ENOMEM:	120	case ENOMEM:
118	throw new SocketException("Could not allocate memory for recvmsg().");	121	throw new SocketException("Could not allocate memory for recvmsg().");
119		122
120	case ENOTCONN:	123	case ENOTCONN:
121	throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");	124	throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");
122		125
123	case ENOTSOCK:	126	case ENOTSOCK:
124	throw new SocketException("The argument s does not refer to a socket.");	127	throw new SocketException("The argument s does not refer to a socket.");
125		128
126	case ECONNRESET:	129	case ECONNRESET:
127	eventAdd(SocketEvent::TYPE_CONNECTION_RESET);	130	eventAdd(SocketEvent::TYPE_CONNECTION_RESET);
128		131
129	// If we have automatic reconnect we want to start it now	132	// If we have automatic reconnect we want to start it now
130	if (myReconnectTimeout == 0)	133	if (myReconnectTimeout == 0)
131	{	134	{
132	reconnectLoop();	135	reconnectLoop();
133	}	136	}
134	else	137	else
135	{	138	{
136	// Otherwise we would like to end the main loop	139	// Otherwise we would like to end the main loop
137	return false;	140	return false;
138	}	141	}
139	break;	142	break;
140		143
141	default:	144	default:
142	throw new SocketException("Unknow exception: " + itos(errno));	145	throw new SocketException("Unknow exception: " + itos(errno));
143	break;	146	break;
144	}	147	}
145	}	148	}
146	else if (status == 0)	149	else if (status == 0)
147	{	150	{
148	// Remote host have done a normal shutdown	151	// Remote host have done a normal shutdown
149	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);	152	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
150		153
151	if (myReconnectTimeout > 0)	154	if (myReconnectTimeout > 0)
152	{	155	{
153	reconnectLoop();	156	reconnectLoop();
154	}	157	}
155	else	158	else
156	{	159	{
157	return false;	160	return false;
158	}	161	}
159	}	162	}
160	else if (status > 0)	163	else if (status > 0)
161	{	164	{
162	// We have received data	165	// We have received data
163	eventAdd(SocketEvent::TYPE_DATA, buffer);	166	eventAdd(SocketEvent::TYPE_DATA, buffer);
164	}	167	}
165		168
166	return true;	169	return true;
167	}	170	}
168		171
169	void AsyncSocket::silentClose()	172	void AsyncSocket::silentClose()
170	{	173	{
171	if (mySocket != -1)	174	if (mySocket != -1)
172	{	175	{
173	::close(mySocket);	176	::close(mySocket);
174	mySocket = -1;	177	mySocket = -1;
175	}	178	}
176	}	179	}
177		180
178	void AsyncSocket::close()	181	void AsyncSocket::close()
179	{	182	{
180	silentClose();	183	silentClose();
181	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);	184	eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
182	}	185	}
183		186
184	void AsyncSocket::reconnectLoop()	187	void AsyncSocket::reconnectLoop()
185	{	188	{
186	while (true)	189	while (true)
187	{	190	{
188	try	191	try
189	{	192	{
190	connect();	193	connect();
191	return;	194	return;
192	}	195	}
193	catch (SocketException *e)	196	catch (SocketException *e)
194	{	197	{
195	eventAdd(SocketEvent::TYPE_CONNECTION_FAILED, e->getDescription());	198	eventAdd(SocketEvent::TYPE_CONNECTION_FAILED, e->getDescription());
196	eventAdd(SocketEvent::TYPE_WAITING_RECONNECT);	199	eventAdd(SocketEvent::TYPE_WAITING_RECONNECT);
197	sleep(myReconnectTimeout);	200	sleep(myReconnectTimeout);
198	}	201	}
199	}	202	}
200	}	203	}
201		204
202	void AsyncSocket::create()	205	void AsyncSocket::create()
203	{	206	{
204	silentClose();	207	silentClose();
205		208
206	mySocket = ::socket(AF_INET, SOCK_STREAM, 0);	209	mySocket = ::socket(AF_INET, SOCK_STREAM, 0);
207		210
208	int on = 1;	211	int on = 1;
209	int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));	212	int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
210	if (status == -1)	213	if (status == -1)
211	{	214	{
212	silentClose();	215	silentClose();
213	throw new SocketException("Create:Reuseaddress: " + itos(errno));	216	throw new SocketException("Create:Reuseaddress: " + itos(errno));
214	}	217	}
215	}	218	}
216		219
217	void AsyncSocket::startListen()	220	void AsyncSocket::startListen()
218	{	221	{
219	create();	222	create();
220		223
221	myAddressStruct.sin_family = AF_INET;	224	myAddressStruct.sin_family = AF_INET;
222	myAddressStruct.sin_addr.s_addr = INADDR_ANY;	225	myAddressStruct.sin_addr.s_addr = INADDR_ANY;
223	myAddressStruct.sin_port = htons(myPort);	226	myAddressStruct.sin_port = htons(myPort);
224		227
225	int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));	228	int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));
226		229
227	if (status == -1)	230	if (status == -1)
228	{	231	{
229	silentClose();	232	silentClose();
230	switch (errno)	233	switch (errno)
231	{	234	{
232	case EACCES:	235	case EACCES:
233	throw new SocketException("The address is protected, and the user is not the superuser.");	236	throw new SocketException("The address is protected, and the user is not the superuser.");
234		237
235	case EADDRINUSE:	238	case EADDRINUSE:
236	throw new SocketException("The given address is already in use.");	239	throw new SocketException("The given address is already in use.");
237		240
238	case EBADF:	241	case EBADF:
239	throw new SocketException("sockfd is not a valid descriptor.");	242	throw new SocketException("sockfd is not a valid descriptor.");
240		243
241	case EINVAL:	244	case EINVAL:
242	throw new SocketException("The socket is already bound to an address.");	245	throw new SocketException("The socket is already bound to an address.");
243		246
244	case ENOTSOCK:	247	case ENOTSOCK:
245	throw new SocketException("sockfd is a descriptor for a file, not a socket.");	248	throw new SocketException("sockfd is a descriptor for a file, not a socket.");
246		249
247	//case EACCES:	250	//case EACCES:
248	//throw new SocketException("Search permission is denied on a component of the path prefix. (See also path_resolution(7).)");	251	//throw new SocketException("Search permission is denied on a component of the path prefix. (See also path_resolution(7).)");
249		252
250	case EADDRNOTAVAIL:	253	case EADDRNOTAVAIL:
251	throw new SocketException("A nonexistent interface was requested or the requested address was not local.");	254	throw new SocketException("A nonexistent interface was requested or the requested address was not local.");
252		255
253	case EFAULT:	256	case EFAULT:
254	throw new SocketException("addr points outside the user's accessible address space.");	257	throw new SocketException("addr points outside the user's accessible address space.");
255		258
256	//case EINVAL:	259	//case EINVAL:
257	//throw new SocketException("The addrlen is wrong, or the socket was not in the AF_UNIX family.");	260	//throw new SocketException("The addrlen is wrong, or the socket was not in the AF_UNIX family.");
258		261
259	case ELOOP:	262	case ELOOP:
260	throw new SocketException("Too many symbolic links were encountered in resolving addr.");	263	throw new SocketException("Too many symbolic links were encountered in resolving addr.");
261		264
262	case ENAMETOOLONG:	265	case ENAMETOOLONG:
263	throw new SocketException("addr is too long.");	266	throw new SocketException("addr is too long.");
264		267
265	case ENOENT:	268	case ENOENT:
266	throw new SocketException("The file does not exist.");	269	throw new SocketException("The file does not exist.");
267		270
268	case ENOMEM:	271	case ENOMEM:
269	throw new SocketException("Insufficient kernel memory was available.");	272	throw new SocketException("Insufficient kernel memory was available.");
270		273
271	case ENOTDIR:	274	case ENOTDIR:
272	throw new SocketException("A component of the path prefix is not a directory.");	275	throw new SocketException("A component of the path prefix is not a directory.");
273		276
274	case EROFS:	277	case EROFS:
275	throw new SocketException("The socket inode would reside on a read-only file system.");	278	throw new SocketException("The socket inode would reside on a read-only file system.");
276		279
277	default:	280	default:
278	throw new SocketException("Unknow exception: " + itos(errno));	281	throw new SocketException("Unknow exception: " + itos(errno));
279	break;	282	break;
280	}	283	}
281	}	284	}
282		285
283	status = ::listen(mySocket, MAXCONNECTIONS);	286	status = ::listen(mySocket, MAXCONNECTIONS);
284		287
285	if (status == -1)	288	if (status == -1)
286	{	289	{
287	silentClose();	290	silentClose();
288	switch (errno)	291	switch (errno)
289	{	292	{
290	case EADDRINUSE:	293	case EADDRINUSE:
291	throw new SocketException("Another socket is already listening on the same port.");	294	throw new SocketException("Another socket is already listening on the same port.");
292		295
293	case EBADF:	296	case EBADF:
294	throw new SocketException("The argument sockfd is not a valid descriptor.");	297	throw new SocketException("The argument sockfd is not a valid descriptor.");
295		298
296	case ENOTSOCK:	299	case ENOTSOCK:
297	throw new SocketException("The argument sockfd is not a socket.");	300	throw new SocketException("The argument sockfd is not a socket.");
298		301
299	case EOPNOTSUPP:	302	case EOPNOTSUPP:
300	throw new SocketException("The socket is not of a type that supports the listen() operation.");	303	throw new SocketException("The socket is not of a type that supports the listen() operation.");
301		304
302	default:	305	default:
303	throw new SocketException("Unknow exception: " + itos(errno));	306	throw new SocketException("Unknow exception: " + itos(errno));
304	break;	307	break;
305	}	308	}
306	}	309	}
307	}	310	}
308		311
309	bool AsyncSocket::accept(AsyncSocket* newSocket)	312	bool AsyncSocket::accept(AsyncSocket* newSocket)
310	{	313	{
311	int addr_length = sizeof(myAddressStruct);	314	int addr_length = sizeof(myAddressStruct);
312	int socket = ::accept(mySocket, (sockaddr)&myAddressStruct, (socklen_t)&addr_length);	315	int socket = ::accept(mySocket, (sockaddr)&myAddressStruct, (socklen_t)&addr_length);
313		316
314	if (socket > 0)	317	if (socket > 0)
315	{	318	{
316	newSocket->setSocket(socket);	319	newSocket->setSocket(socket);
317	return true;	320	return true;
318	}	321	}
319		322
320	return false;	323	return false;
321	}	324	}
322		325
323	void AsyncSocket::connect()	326	void AsyncSocket::connect()
324	{	327	{
325	eventAdd(SocketEvent::TYPE_CONNECTING);	328	eventAdd(SocketEvent::TYPE_CONNECTING);
326		329
327	create();	330	create();
328		331
329	memset(&myAddressStruct, 0, sizeof(myAddressStruct));	332	memset(&myAddressStruct, 0, sizeof(myAddressStruct));
330		333
331	myAddressStruct.sin_family = AF_INET;	334	myAddressStruct.sin_family = AF_INET;
332	myAddressStruct.sin_port = htons(myPort);	335	myAddressStruct.sin_port = htons(myPort);
333		336
334	struct hostent *hptr = gethostbyname(myAddress.c_str());	337	struct hostent *hptr = gethostbyname(myAddress.c_str());
335	if (hptr == NULL)	338	if (hptr == NULL)
336	{	339	{
337	silentClose();	340	silentClose();
338	throw new SocketException("Connect: Could not resolv ip address");	341	throw new SocketException("Connect: Could not resolv ip address");
339	}	342	}
340		343
341	memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);	344	memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);
342		345
343	int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));	346	int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
344		347
345	if (status == -1)	348	if (status == -1)
346	{	349	{
347	silentClose();	350	silentClose();
348	switch (errno)	351	switch (errno)
349	{	352	{
350	case EACCES:	353	case EACCES:
351	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).)");	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).)");
352		355
353	case EPERM:	356	case EPERM:
354	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.");	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.");
355		358
356	case EADDRINUSE:	359	case EADDRINUSE:
357	throw new SocketException("Local address is already in use.");	360	throw new SocketException("Local address is already in use.");
358		361
359	case EAFNOSUPPORT:	362	case EAFNOSUPPORT:
360	throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");	363	throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");
361		364
362	case EAGAIN:	365	case EAGAIN:
363	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.");	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.");
364		367
365	case EALREADY:	368	case EALREADY:
366	throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");	369	throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");
367		370
368	case EBADF:	371	case EBADF:
369	throw new SocketException("The file descriptor is not a valid index in the descriptor table.");	372	throw new SocketException("The file descriptor is not a valid index in the descriptor table.");
370		373
371	case ECONNREFUSED:	374	case ECONNREFUSED:
372	throw new SocketException("No-one listening on the remote address.");	375	throw new SocketException("No-one listening on the remote address.");
373		376
374	case EFAULT:	377	case EFAULT:
375	throw new SocketException("The socket structure address is outside the user's address space.");	378	throw new SocketException("The socket structure address is outside the user's address space.");
376		379
377	case EINPROGRESS:	380	case EINPROGRESS:
378	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).");	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).");
379		382
380	case EINTR:	383	case EINTR:
381	throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");	384	throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");
382		385
383	case EISCONN:	386	case EISCONN:
384	throw new SocketException("The socket is already connected.");	387	throw new SocketException("The socket is already connected.");
385		388
386	case ENETUNREACH:	389	case ENETUNREACH:
387	throw new SocketException("Network is unreachable.");	390	throw new SocketException("Network is unreachable.");
388		391
389	case ENOTSOCK:	392	case ENOTSOCK:
390	throw new SocketException("The file descriptor is not associated with a socket.");	393	throw new SocketException("The file descriptor is not associated with a socket.");
391		394
392	case ETIMEDOUT:	395	case ETIMEDOUT:
393	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.");	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.");
394		397
395	default:	398	default:
396	throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));	399	throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
397	}	400	}
398	}	401	}
399		402
400	eventAdd(SocketEvent::TYPE_CONNECTED);	403	eventAdd(SocketEvent::TYPE_CONNECTED);
401	}	404	}
402		405
403	void AsyncSocket::sendData(string data)	406	void AsyncSocket::sendData(string data)
404	{	407	{
405	mySendMutex.lock();	408	mySendMutex.lock();
406		409
407	int status = ::send(mySocket, data.c_str(), data.size(), 0);	410	int status = ::send(mySocket, data.c_str(), data.size(), MSG_NOSIGNAL);
408	//Logger::getInstance().add("Sent: \"" + data + "\" status was " + itos(status) + "\n");	411	//Logger::getInstance().add("Sent: \"" + data + "\" status was " + itos(status) + "\n");
409	mySendMutex.unlock();	412	mySendMutex.unlock();
410		413
411	//Logger &log = Logger::getInstance();	414	//Logger &log = Logger::getInstance();
412	//log.add("Sent: \"" + data + "\" status was " + itos(status) + "\n");	415	//log.add("Sent: \"" + data + "\" status was " + itos(status) + "\n");
413		416
414	if (status == -1)	417	if (status == -1)
415	{	418	{
416	switch (errno)	419	switch (errno)
417	{	420	{
418	case EACCES:	421	case EACCES:
419	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).)");	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).)");
420		423
421	case EAGAIN:	424	case EAGAIN:
422	throw new SocketException("The socket is marked non-blocking and the requested operation would block.");	425	throw new SocketException("The socket is marked non-blocking and the requested operation would block.");
423		426
424	case EBADF:	427	case EBADF:
425	throw new SocketException("An invalid descriptor was specified.");	428	throw new SocketException("An invalid descriptor was specified.");
426		429
427	case ECONNRESET:	430	case ECONNRESET:
428	throw new SocketException("Connection reset by peer.");	431	throw new SocketException("Connection reset by peer.");
429		432
430	case EDESTADDRREQ:	433	case EDESTADDRREQ:
431	throw new SocketException("The socket is not connection-mode, and no peer address is set.");	434	throw new SocketException("The socket is not connection-mode, and no peer address is set.");
432		435
433	case EFAULT:	436	case EFAULT:
434	throw new SocketException("An invalid user space address was specified for an argument.");	437	throw new SocketException("An invalid user space address was specified for an argument.");
435		438
436	case EINTR:	439	case EINTR:
437	throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");	440	throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");
438		441
439	case EINVAL:	442	case EINVAL:
440	throw new SocketException("1Invalid argument passed.");	443	throw new SocketException("1Invalid argument passed.");
441		444
442	case EISCONN:	445	case EISCONN:
443	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.)");	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.)");
444		447
445	case EMSGSIZE:	448	case EMSGSIZE:
446	throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");	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.");
447		450
448	case ENOBUFS:	451	case ENOBUFS:
449	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.)");	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.)");
450		453
451	case ENOMEM:	454	case ENOMEM:
452	throw new SocketException("No memory available.");	455	throw new SocketException("No memory available.");
453		456
454	case ENOTCONN:	457	case ENOTCONN:
455	throw new SocketException("The socket is not connected, and no target has been given.");	458	throw new SocketException("The socket is not connected, and no target has been given.");
456		459
457	case ENOTSOCK:	460	case ENOTSOCK:
458	throw new SocketException("The argument s is not a socket.");	461	throw new SocketException("The argument s is not a socket.");
459		462
460	case EOPNOTSUPP:	463	case EOPNOTSUPP:
461	throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");	464	throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");
462		465
463	case EPIPE:	466	case EPIPE:
-		467	close();
-		468	stop();
464	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.");	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.");
-		470	break;
465		471
466	default:	472	default:
467	throw new SocketException("Unknow exception: " + itos(errno));	473	throw new SocketException("Unknow exception: " + itos(errno));
468	}	474	}
469	}	475	}
470	}	476	}
471		477
472	void AsyncSocket::eventAdd(unsigned int eventType)	478	void AsyncSocket::eventAdd(unsigned int eventType)
473	{	479	{
474	eventAdd(eventType, "");	480	eventAdd(eventType, "");
475	}	481	}
476		482
477	void AsyncSocket::eventAdd(unsigned int eventType, string eventData)	483	void AsyncSocket::eventAdd(unsigned int eventType, string eventData)
478	{	484	{
479	SocketEvent socketEvent(eventType, eventData);	485	SocketEvent socketEvent(eventType, eventData);
480		486
481	if (myEventCallback == NULL)	487	if (myEventCallback == NULL)
482	{	488	{
483	myEventQueue.push(socketEvent);	489	myEventQueue.push(socketEvent);
484	myEventSemaphore.broadcast();	490	myEventSemaphore.broadcast();
485	}	491	}
486	else	492	else
487	{	493	{
488	myEventCallback->handleEvent(myId, socketEvent);	494	myEventCallback->handleEvent(myId, socketEvent);
489	}	495	}
490	}	496	}
491		497
492	void AsyncSocket::eventSetCallback(SocketEventCallback* eventCallback)	498	void AsyncSocket::eventSetCallback(SocketEventCallback* eventCallback)
493	{	499	{
494	myEventCallback = eventCallback;	500	myEventCallback = eventCallback;

Subversion Repositories HomeAutomation

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