Subversion Repositories HomeAutomation

Rev

Rev 983 | Show entire file | Ignore whitespace | Details | Blame | Last modification | View Log | SVN | RSS feed

Rev 983 Rev 984
Line 1... Line 1...
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 "asyncsocket.h"
22
#include "asyncsocket.h"
23
 
23
 
24
int socketCount = 1;
24
int socketCount = 1;
Line 27... Line 27...
27
{
27
{
28
    myId = socketCount++;
28
    myId = socketCount++;
29
    mySocket = -1;
29
    mySocket = -1;
30
    myReconnectTimeout = 0;
30
    myReconnectTimeout = 0;
31
    myForceReconnect = false;
31
    myForceReconnect = false;
32
 
-
 
33
    Thread<AsyncSocket>();
-
 
34
}
32
}
35
 
33
 
36
AsyncSocket::~AsyncSocket()
34
AsyncSocket::~AsyncSocket()
37
{
35
{
38
    if (mySocket != -1)
-
 
39
    {
-
 
40
        ::close(mySocket);
-
 
41
    }
-
 
42
 
-
 
43
    stop();
36
    stop();
-
 
37
    silentClose();
44
}
38
}
45
 
39
 
46
void AsyncSocket::run()
40
void AsyncSocket::run()
47
{
41
{
48
    SyslogStream &slog = SyslogStream::getInstance();
42
    // If connection is already up then we should not connect again
49
 
-
 
50
    if (!isConnected())
43
    if (!isConnected())
51
    {
44
    {
-
 
45
        // If we have choosen to not use automatic reconnect do not start reconnect loop
52
        if (myReconnectTimeout == 0)
46
        if (myReconnectTimeout == 0)
53
        {
47
        {
-
 
48
            // Connect to somewhere
54
            connect();
49
            connect();
55
        }
50
        }
56
        else
51
        else
57
        {
52
        {
-
 
53
            // Start the reconnect loop
58
            reconnectLoop();
54
            reconnectLoop();
59
        }
55
        }
60
    }
56
    }
61
 
-
 
62
    char buf[MAXBUFFER + 1];
-
 
63
    string data;
-
 
64
    int status;
-
 
65
 
57
 
66
    bool loop = true;
58
    bool loop = true;
67
    try
59
    try
68
    {
60
    {
69
        while (loop)
61
        while (loop)
70
        {
62
        {
-
 
63
            // If we have triggered a forced reconnect do it here
71
            if (myForceReconnect)
64
            if (myForceReconnect)
72
            {
65
            {
-
 
66
                myForceReconnect = false;
73
                reconnectLoop();
67
                reconnectLoop();
74
            }
68
            }
-
 
69
 
-
 
70
            // Receive data
-
 
71
            loop = receiveData();
-
 
72
        }
-
 
73
    }
-
 
74
    catch (SocketException *e)
-
 
75
    {
-
 
76
        // Something bad happend and we can not continue
-
 
77
        eventAdd(SocketEvent::TYPE_CONNECTION_DIED, e->getDescription());
-
 
78
    }
-
 
79
 
-
 
80
    // Clean up socket if we would want to restart
-
 
81
    silentClose();
-
 
82
}
75
 
83
 
-
 
84
bool AsyncSocket::receiveData()
-
 
85
{
-
 
86
    char buffer[MAXBUFFER + 1];
76
            memset(buf, 0, MAXBUFFER + 1);
87
    memset(buffer, 0, MAXBUFFER + 1);
-
 
88
 
77
            status = ::recv(mySocket, buf, MAXBUFFER, 0);
89
    int status = ::recv(mySocket, buffer, MAXBUFFER, 0);
78
 
90
 
79
            if (status == -1)
91
    if (status == -1)
80
            {
92
    {
81
                switch (errno)
93
        switch (errno)
82
                {
94
        {
83
                    case EAGAIN:
95
            case EAGAIN:
84
                    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.");
96
            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.");
-
 
97
 
-
 
98
            case EBADF:
-
 
99
            throw new SocketException("The argument s is an invalid descriptor.");
-
 
100
 
-
 
101
            case ECONNREFUSED:
-
 
102
            throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
-
 
103
 
-
 
104
            case EFAULT:
-
 
105
            throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
-
 
106
 
85
                    break;
107
            case EINTR:
-
 
108
            throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
-
 
109
 
-
 
110
            case EINVAL:
-
 
111
            throw new SocketException("Invalid argument passed.");
-
 
112
 
-
 
113
            case ENOMEM:
-
 
114
            throw new SocketException("Could not allocate memory for recvmsg().");
-
 
115
 
-
 
116
            case ENOTCONN:
-
 
117
            throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and accept(2)).");
-
 
118
 
-
 
119
            case ENOTSOCK:
-
 
120
            throw new SocketException("The argument s does not refer to a socket.");
-
 
121
 
-
 
122
            case ECONNRESET:
-
 
123
            eventAdd(SocketEvent::TYPE_CONNECTION_RESET);
86
 
124
 
87
                    case EBADF:
125
            // If we have automatic reconnect we want to start it now
88
                    throw new SocketException("The argument s is an invalid descriptor.");
126
            if (myReconnectTimeout == 0)
89
 
127
            {
90
                    case ECONNREFUSED:
128
                reconnectLoop();
91
                    throw new SocketException("A remote host refused to allow the network connection (typically because it is not running the requested service).");
-
 
92
 
129
            }
93
                    case EFAULT:
130
            else
-
 
131
            {
94
                    throw new SocketException("The receive buffer pointer(s) point outside the process's address space.");
132
                // Otherwise we would like to end the main loop
-
 
133
                return false;
95
 
134
            }
96
                    case EINTR:
135
            break;
97
                    throw new SocketException("The receive was interrupted by delivery of a signal before any data were available; see signal(7).");
-
 
98
 
136
 
99
                    case EINVAL:
137
            default:
100
                    throw new SocketException("Invalid argument passed.");
138
            throw new SocketException("Unknow exception: " + itos(errno));
101
                    break;
139
            break;
102
 
140
        }
103
                    case ENOMEM:
-
 
104
                    throw new SocketException("Could not allocate memory for recvmsg().");
-
 
105
 
141
    }
106
                    case ENOTCONN:
142
    else if (status == 0)
107
                    throw new SocketException("The socket is associated with a connection-oriented protocol and has not been connected (see connect(2) and  accept(2)).");
-
 
108
 
143
    {
109
                    case ENOTSOCK:
144
        // Remote host have done a normal shutdown
110
                    throw new SocketException("The argument s does not refer to a socket.");
145
        eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
111
 
146
 
112
                    case ECONNRESET:
-
 
113
                    setEvent(ASYNCSOCKET_EVENT_RESET);
-
 
114
                    if (myReconnectTimeout > 0)
147
        if (myReconnectTimeout > 0)
115
                    {
148
        {
116
                        reconnectLoop();
149
            reconnectLoop();
117
                    }
150
        }
118
                    else
151
        else
119
                    {
152
        {
120
                        loop = false;
153
            return false;
121
                    }
-
 
122
                    break;
-
 
123
 
-
 
124
                    default:
-
 
125
                    throw new SocketException("Unknow exception: " + itos(errno));
-
 
126
                    break;
-
 
127
                }
154
        }
128
            }
155
    }
129
            else if (status == 0)
156
    else if (status > 0)
130
            {
157
    {
131
                setEvent(ASYNCSOCKET_EVENT_CLOSED);
-
 
132
                if (myReconnectTimeout > 0)
158
        // We have received data
133
                {
-
 
134
                    reconnectLoop();
159
        eventAdd(SocketEvent::TYPE_DATA, buffer);
135
                }
160
    }
136
                else
-
 
137
                {
161
 
138
                    loop = false;
162
    return true;
139
                }
-
 
140
            }
163
}
141
            else if (status > 0)
-
 
142
            {
-
 
143
                data = buf;
-
 
144
 
164
 
145
                //slog << "Received: " + data + "\n";
165
void AsyncSocket::silentClose()
146
 
166
{
147
                myInQueue.push(data);
167
    if (mySocket != -1)
-
 
168
    {
148
                setEvent(ASYNCSOCKET_EVENT_DATA);
169
        ::close(mySocket);
149
            }
170
        mySocket = -1;
150
        }
171
    }
151
    }
172
}
152
    catch (SocketException *e)
-
 
153
    {
173
 
154
        slog << "Exception: " + e->getDescription() + "\n";
-
 
155
        setEvent(ASYNCSOCKET_EVENT_DIED);
174
void AsyncSocket::close()
156
    }
-
 
157
 
175
{
158
    close();
176
    silentClose();
-
 
177
    eventAdd(SocketEvent::TYPE_CONNECTION_CLOSED);
159
}
178
}
160
 
179
 
161
void AsyncSocket::reconnectLoop()
180
void AsyncSocket::reconnectLoop()
162
{
181
{
163
    SyslogStream &slog = SyslogStream::getInstance();
-
 
164
 
-
 
165
    while (1)
182
    while (true)
166
    {
183
    {
167
        try
184
        try
168
        {
185
        {
169
            connect();
186
            connect();
170
            break;
187
            return;
171
        }
188
        }
172
        catch (SocketException *e)
189
        catch (SocketException *e)
173
        {
190
        {
174
            setEvent(ASYNCSOCKET_EVENT_CONNECT_FAILED);
191
            eventAdd(SocketEvent::TYPE_CONNECTION_FAILED, e->getDescription());
175
            setEvent(ASYNCSOCKET_EVENT_WAITING_TO_RECONNECT);
192
            eventAdd(SocketEvent::TYPE_WAITING_RECONNECT);
176
            slog << "Could not connect: " + e->getDescription() + "\n";//FIXME: Remove these
-
 
177
            slog << "Will try again in " + itos(myReconnectTimeout) + " seconds\n";
-
 
178
            sleep(myReconnectTimeout);
193
            sleep(myReconnectTimeout);
179
        }
194
        }
180
    }
195
    }
181
}
196
}
182
 
197
 
183
void AsyncSocket::create()
198
void AsyncSocket::create()
184
{
199
{
185
    close();
200
    silentClose();
186
 
201
 
187
    mySocket = ::socket(AF_INET, SOCK_STREAM, 0);
202
    mySocket = ::socket(AF_INET, SOCK_STREAM, 0);
188
 
203
 
189
    int on = 1;
204
    int on = 1;
190
    int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
205
    int status = setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on));
191
    if (status == -1)
206
    if (status == -1)
192
    {
207
    {
193
        close();
208
        silentClose();
194
        throw new SocketException("Create:Reuseaddress: " + itos(errno));
209
        throw new SocketException("Create:Reuseaddress: " + itos(errno));
195
    }
210
    }
196
}
211
}
197
 
212
 
198
void AsyncSocket::startListen()
213
void AsyncSocket::startListen()
199
{
214
{
200
    create();
215
    create();
201
 
216
 
202
    myAddressStruct.sin_family = AF_INET;
217
    myAddressStruct.sin_family = AF_INET;
203
    myAddressStruct.sin_addr.s_addr = INADDR_ANY;
218
    myAddressStruct.sin_addr.s_addr = INADDR_ANY;
204
    myAddressStruct.sin_port = htons(myPort);
219
    myAddressStruct.sin_port = htons(myPort);
205
 
220
 
206
    int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));
221
    int status = ::bind(mySocket, (struct sockaddr *)&myAddressStruct, sizeof(myAddressStruct));
207
 
222
 
208
    if (status == -1)
223
    if (status == -1)
209
    {
224
    {
210
        close();
225
        silentClose();
211
        switch (errno)
226
        switch (errno)
212
        {
227
        {
213
            case EACCES:
228
            case EACCES:
214
            throw new SocketException("The address is protected, and the user is not the superuser.");
229
            throw new SocketException("The address is protected, and the user is not the superuser.");
215
 
230
 
Line 263... Line 278...
263
 
278
 
264
    status = ::listen(mySocket, MAXCONNECTIONS);
279
    status = ::listen(mySocket, MAXCONNECTIONS);
265
 
280
 
266
    if (status == -1)
281
    if (status == -1)
267
    {
282
    {
268
        close();
283
        silentClose();
269
        switch (errno)
284
        switch (errno)
270
        {
285
        {
271
            case EADDRINUSE:
286
            case EADDRINUSE:
272
            throw new SocketException("Another socket is already listening on the same port.");
287
            throw new SocketException("Another socket is already listening on the same port.");
273
 
288
 
Line 281... Line 296...
281
            throw new SocketException("The socket is not of a type that supports the listen() operation.");
296
            throw new SocketException("The socket is not of a type that supports the listen() operation.");
282
 
297
 
283
            default:
298
            default:
284
            throw new SocketException("Unknow exception: " + itos(errno));
299
            throw new SocketException("Unknow exception: " + itos(errno));
285
            break;
300
            break;
286
        }
301
        }
287
    }
302
    }
288
}
303
}
289
 
304
 
290
bool AsyncSocket::accept(AsyncSocket* newSocket)
305
bool AsyncSocket::accept(AsyncSocket* newSocket)
291
{
306
{
292
    int addr_length = sizeof(myAddressStruct);
307
    int addr_length = sizeof(myAddressStruct);
293
    int socket = ::accept(mySocket, (sockaddr*)&myAddressStruct, (socklen_t*)&addr_length);
308
    int socket = ::accept(mySocket, (sockaddr*)&myAddressStruct, (socklen_t*)&addr_length);
294
 
309
 
295
    if (socket > 0)
310
    if (socket > 0)
296
    {
311
    {
297
        newSocket->setSocket(socket);
312
        newSocket->setSocket(socket);
298
        return true;
313
        return true;
299
    }
314
    }
300
 
315
 
301
    return false;
316
    return false;
302
}
317
}
303
 
318
 
304
void AsyncSocket::connect()
319
void AsyncSocket::connect()
305
{
320
{
306
    SyslogStream &slog = SyslogStream::getInstance();
321
    eventAdd(SocketEvent::TYPE_CONNECTING);
307
 
322
 
308
    create();
323
    create();
309
 
324
 
310
    memset(&myAddressStruct, 0, sizeof(myAddressStruct));
325
    memset(&myAddressStruct, 0, sizeof(myAddressStruct));
311
 
326
 
312
    myAddressStruct.sin_family = AF_INET;
327
    myAddressStruct.sin_family = AF_INET;
313
    myAddressStruct.sin_port = htons(myPort);
328
    myAddressStruct.sin_port = htons(myPort);
314
 
-
 
315
    //slog << "Trying to resolv " + myAddress + "\n";
-
 
316
 
329
 
317
    struct hostent *hptr = gethostbyname(myAddress.c_str());
330
    struct hostent *hptr = gethostbyname(myAddress.c_str());
318
    if (hptr == NULL)
331
    if (hptr == NULL)
319
    {
332
    {
320
        close();
333
        silentClose();
321
        throw new SocketException("Connect: Could not resolv ip address");
334
        throw new SocketException("Connect: Could not resolv ip address");
322
    }
335
    }
323
 
336
 
324
   
-
 
325
    memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);
337
    memcpy(&myAddressStruct.sin_addr, hptr->h_addr, hptr->h_length);
326
 
338
 
327
    int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
339
    int status = ::connect(mySocket, (sockaddr*)&myAddressStruct, sizeof(myAddressStruct));
328
 
340
 
329
    if (status == -1)
341
    if (status == -1)
330
    {
342
    {
331
        close();
343
        silentClose();
332
        switch (errno)
344
        switch (errno)
333
        {
345
        {
334
            case EACCES:
346
            case EACCES:
335
            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).)");
347
            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).)");
336
 
348
 
337
            case EPERM:
349
            case EPERM:
338
            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.");
350
            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.");
339
 
351
 
340
            case EADDRINUSE:
352
            case EADDRINUSE:
341
            throw new SocketException("Local address is already in use.");
353
            throw new SocketException("Local address is already in use.");
342
 
354
 
343
            case EAFNOSUPPORT:
355
            case EAFNOSUPPORT:
344
            throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");
356
            throw new SocketException("The passed address didn't have the correct address family in its sa_family field.");
345
 
357
 
346
            case EAGAIN:
358
            case EAGAIN:
347
            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.");
359
            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.");
348
 
360
 
349
            case EALREADY:
361
            case EALREADY:
350
            throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");
362
            throw new SocketException("The socket is non-blocking and a previous connection attempt has not yet been completed.");
351
 
363
 
352
            case EBADF:
364
            case EBADF:
Line 357... Line 369...
357
 
369
 
358
            case EFAULT:
370
            case EFAULT:
359
            throw new SocketException("The socket structure address is outside the user's address space.");
371
            throw new SocketException("The socket structure address is outside the user's address space.");
360
 
372
 
361
            case EINPROGRESS:
373
            case EINPROGRESS:
362
            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).");
374
            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).");
363
 
375
 
364
            case EINTR:
376
            case EINTR:
365
            throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");
377
            throw new SocketException("The system call was interrupted by a signal that was caught; see signal(7).");
366
 
378
 
367
            case EISCONN:
379
            case EISCONN:
Line 372... Line 384...
372
 
384
 
373
            case ENOTSOCK:
385
            case ENOTSOCK:
374
            throw new SocketException("The file descriptor is not associated with a socket.");
386
            throw new SocketException("The file descriptor is not associated with a socket.");
375
 
387
 
376
            case ETIMEDOUT:
388
            case ETIMEDOUT:
377
            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.");
389
            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.");
378
 
390
 
379
            default:
391
            default:
380
            throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
392
            throw new SocketException("Other errors may be generated by the underlying protocol modules. : " + itos(errno));
381
        }
393
        }
382
    }
394
    }
383
 
-
 
384
    myForceReconnect = false;
-
 
385
 
395
 
386
    setEvent(ASYNCSOCKET_EVENT_CONNECTED);
396
    eventAdd(SocketEvent::TYPE_CONNECTED);
387
}
-
 
388
 
-
 
389
void AsyncSocket::close()
-
 
390
{
-
 
391
    if (mySocket != -1)
-
 
392
    {
-
 
393
        ::close(mySocket);
-
 
394
        mySocket = -1;
-
 
395
        setEvent(ASYNCSOCKET_EVENT_CLOSED);
-
 
396
    }
-
 
397
}
-
 
398
 
-
 
399
bool AsyncSocket::isConnected()
-
 
400
{
-
 
401
    return mySocket != -1;
-
 
402
}
-
 
403
 
-
 
404
void AsyncSocket::setReconnectTimeout(unsigned int timeout)
-
 
405
{
-
 
406
    myReconnectTimeout = timeout;
-
 
407
}
-
 
408
 
-
 
409
void AsyncSocket::startEvent()
-
 
410
{
-
 
411
    myEventSemaphore.lock();
-
 
412
}
-
 
413
 
-
 
414
bool AsyncSocket::availableEvent()
-
 
415
{
-
 
416
    return (myEventQueue.size() > 0);
-
 
417
}
-
 
418
 
-
 
419
int AsyncSocket::getEvent()
-
 
420
{
-
 
421
    return myEventQueue.pop();
-
 
422
}
-
 
423
 
-
 
424
void AsyncSocket::waitForEvent()
-
 
425
{
-
 
426
    myEventSemaphore.wait();
-
 
427
}
-
 
428
 
-
 
429
void AsyncSocket::stopEvent()
-
 
430
{
-
 
431
    myEventSemaphore.unlock();
-
 
432
}
-
 
433
 
-
 
434
void AsyncSocket::setAddress(string address, int port)
-
 
435
{
-
 
436
    myAddress = address;
-
 
437
    myPort = port;
-
 
438
}
-
 
439
 
-
 
440
void AsyncSocket::setPort(int port)
-
 
441
{
-
 
442
    myPort = port;
-
 
443
}
-
 
444
 
-
 
445
void AsyncSocket::setSocket(int socket)
-
 
446
{
-
 
447
    mySocket = socket;
-
 
448
}
-
 
449
 
-
 
450
int AsyncSocket::getSocket()
-
 
451
{
-
 
452
    return mySocket;
-
 
453
}
-
 
454
 
-
 
455
bool AsyncSocket::availableData()
-
 
456
{
-
 
457
    return (myInQueue.size() > 0);
-
 
458
}
-
 
459
 
-
 
460
string AsyncSocket::getData()
-
 
461
{
-
 
462
    return myInQueue.pop();
-
 
463
}
397
}
464
 
398
 
465
bool AsyncSocket::sendData(string data)
399
void AsyncSocket::sendData(string data)
466
{
400
{
467
    //cout << "SendData on " + myAddress + ":" + itos(myPort) + ".\n";
-
 
468
    sendDataDirect(data);
-
 
469
    //myOutQueue.push(data);
-
 
470
    //mySemaphore.broadcast();
-
 
471
    return true;
-
 
472
}
-
 
473
 
-
 
474
void AsyncSocket::sendDataDirect(string data)
-
 
475
{
-
 
476
    SyslogStream &slog = SyslogStream::getInstance();
-
 
477
 
-
 
478
    //slog << "Sending: " << data << "\n";
-
 
479
 
-
 
480
    int status = ::send(mySocket, data.c_str(), data.size(), 0);
401
    int status = ::send(mySocket, data.c_str(), data.size(), 0);
481
 
-
 
482
    //slog << "Status: " << status << "\n";
-
 
483
 
402
 
484
    try
403
    if (status == -1)
485
    {
404
    {
486
        if (status == -1)
405
        switch (errno)
487
        {
406
        {
488
            switch (errno)
-
 
489
            {
-
 
490
                case EACCES:
407
            case EACCES:
491
                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).)");
408
            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).)");
492
 
409
 
493
                case EAGAIN:
410
            case EAGAIN:
494
                //slog << "The socket is marked non-blocking and the requested operation would block.\n";
411
            throw new SocketException("The socket is marked non-blocking and the requested operation would block.");
495
                break;
-
 
496
 
412
 
497
                case EBADF:
413
            case EBADF:
498
                throw new SocketException("An invalid descriptor was specified.");
414
            throw new SocketException("An invalid descriptor was specified.");
499
 
415
 
500
                case ECONNRESET:
416
            case ECONNRESET:
501
                throw new SocketException("Connection reset by peer.");
417
            throw new SocketException("Connection reset by peer.");
502
 
418
 
503
                case EDESTADDRREQ:
419
            case EDESTADDRREQ:
504
                throw new SocketException("The socket is not connection-mode, and no peer address is set.");
420
            throw new SocketException("The socket is not connection-mode, and no peer address is set.");
505
 
421
 
506
                case EFAULT:
422
            case EFAULT:
507
                throw new SocketException("An invalid user space address was specified for an argument.");
423
            throw new SocketException("An invalid user space address was specified for an argument.");
508
 
424
 
509
                case EINTR:
425
            case EINTR:
510
                throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");
426
            throw new SocketException("A signal occurred before any data was transmitted; see signal(7).");
511
 
427
 
512
                case EINVAL:
428
            case EINVAL:
513
                throw new SocketException("1Invalid argument passed.");
429
            throw new SocketException("1Invalid argument passed.");
514
 
430
 
515
                case EISCONN:
431
            case EISCONN:
516
                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.)");
432
            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.)");
517
 
433
 
518
                case EMSGSIZE:
434
            case EMSGSIZE:
519
                throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");
435
            throw new SocketException("The socket type requires that message be sent atomically, and the size of the message to be sent made this impossible.");
520
 
436
 
521
                case ENOBUFS:
437
            case ENOBUFS:
522
                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.)");
438
            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.)");
523
 
439
 
524
                case ENOMEM:
440
            case ENOMEM:
525
                throw new SocketException("No memory available.");
441
            throw new SocketException("No memory available.");
526
 
442
 
527
                case ENOTCONN:
443
            case ENOTCONN:
528
                throw new SocketException("The socket is not connected, and no target has been given.");
444
            throw new SocketException("The socket is not connected, and no target has been given.");
529
 
445
 
530
                case ENOTSOCK:
446
            case ENOTSOCK:
531
                throw new SocketException("The argument s is not a socket.");
447
            throw new SocketException("The argument s is not a socket.");
532
 
448
 
533
                case EOPNOTSUPP:
449
            case EOPNOTSUPP:
534
                throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");
450
            throw new SocketException("Some bit in the flags argument is inappropriate for the socket type.");
535
 
451
 
536
                case EPIPE:
452
            case EPIPE:
537
                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.");
453
            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.");
538
 
454
 
539
                default:
455
            default:
540
                slog << "Unknow exception: " + itos(errno) + "\n";
456
            throw new SocketException("Unknow exception: " + itos(errno));
541
                break;
-
 
542
            }
-
 
543
        }
457
        }
544
    }
458
    }
545
    catch (SocketException *e)
-
 
546
    {
-
 
547
        slog << "sendDataDirect: Exception: " + e->getDescription() + "\n";
-
 
548
    }
-
 
549
}
459
}
550
 
460
 
551
void AsyncSocket::setEvent(int event)
461
void AsyncSocket::eventAdd(unsigned int eventType)
-
 
462
{
-
 
463
    eventAdd(eventType, "");
-
 
464
}
-
 
465
 
-
 
466
void AsyncSocket::eventAdd(unsigned int eventType, string eventData)
552
{
467
{
-
 
468
    SocketEvent socketEvent(eventType, eventData);
553
    myEventQueue.push(event);
469
    myEventQueue.push(socketEvent);
554
    myEventSemaphore.broadcast();
470
    myEventSemaphore.broadcast();
555
}
471
}
556
 
-
 
557
void AsyncSocket::forceReconnect()
-
 
558
{
-
 
559
    ///FIXME: Threadsafe?
-
 
560
    myForceReconnect = true;
-
 
561
}
-
 
562
 
-