generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-10-23 08:31:46 +02:00
Code Issues Projects Releases Wiki Activity

Different connecters simplified

Browse Source 
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
This commit is contained in:
Martin Sustrik
2011-07-29 09:37:43 +02:00
parent f63db009a1
commit d5f3628ad0
8 changed files with 117 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
src/ipc_connecter.cpp
View File

@@ -18,32 +18,24 @@
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "ipc_connecter.hpp"
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
#include <new>
#include <string>
#include "ipc_connecter.hpp"
#include "tcp_engine.hpp"
#include "io_thread.hpp"
#include "platform.hpp"
#include "random.hpp"
#include "ip.hpp"
#include "err.hpp"
#include "ip.hpp"
#if defined ZMQ_HAVE_WINDOWS
#include "windows.hpp"
#else
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/tcp.h>
#include <netinet/in.h>
#include <netdb.h>
#include <fcntl.h>
#ifdef ZMQ_HAVE_OPENVMS
#include <ioctl.h>
#endif
#endif
#include <sys/un.h>
zmq::ipc_connecter_t::ipc_connecter_t (class io_thread_t *io_thread_,
class session_t *session_, const options_t &options_,
@@ -175,84 +167,6 @@ int zmq::ipc_connecter_t::get_new_reconnect_ivl ()
return this_interval;
}
#ifdef ZMQ_HAVE_WINDOWS
int zmq::ipc_connecter_t::set_address (const char *protocol_, const char *addr_)
{
errno = EPROTONOSUPPORT;
return -1;
}
int zmq::ipc_connecter_t::open ()
{
zmq_assert (s == retired_fd);
// Create the socket.
s = socket (addr.ss_family, SOCK_STREAM, IPPROTO_TCP);
if (s == INVALID_SOCKET) {
wsa_error_to_errno ();
return -1;
}
// Set to non-blocking mode.
unsigned long argp = 1;
int rc = ioctlsocket (s, FIONBIO, &argp);
wsa_assert (rc != SOCKET_ERROR);
// Connect to the remote peer.
rc = ::connect (s, (sockaddr*) &addr, addr_len);
// Connect was successfull immediately.
if (rc == 0)
return 0;
// Asynchronous connect was launched.
if (rc == SOCKET_ERROR && (WSAGetLastError () == WSAEINPROGRESS ||
WSAGetLastError () == WSAEWOULDBLOCK)) {
errno = EAGAIN;
return -1;
}
wsa_error_to_errno ();
return -1;
}
int zmq::ipc_connecter_t::close ()
{
zmq_assert (s != retired_fd);
int rc = closesocket (s);
wsa_assert (rc != SOCKET_ERROR);
s = retired_fd;
return 0;
}
zmq::fd_t zmq::ipc_connecter_t::connect ()
{
// Nonblocking connect have finished. Check whether an error occured.
int err = 0;
socklen_t len = sizeof err;
int rc = getsockopt (s, SOL_SOCKET, SO_ERROR, (char*) &err, &len);
zmq_assert (rc == 0);
if (err != 0) {
// Assert that the error was caused by the networking problems
// rather than 0MQ bug.
if (err == WSAECONNREFUSED || err == WSAETIMEDOUT ||
err == WSAECONNABORTED || err == WSAEHOSTUNREACH ||
err == WSAENETUNREACH || err == WSAENETDOWN)
return retired_fd;
wsa_assert_no (err);
}
// Return the newly connected socket.
fd_t result = s;
s = retired_fd;
return result;
}
#else
int zmq::ipc_connecter_t::set_address (const char *addr_)
{
return resolve_local_path (&addr, &addr_len, addr_);
@@ -263,78 +177,23 @@ int zmq::ipc_connecter_t::open ()
zmq_assert (s == retired_fd);
struct sockaddr *sa = (struct sockaddr*) &addr;
if (AF_UNIX != sa->sa_family) {
// Create the socket.
s = socket (sa->sa_family, SOCK_STREAM, IPPROTO_TCP);
if (s == -1)
return -1;
// Set to non-blocking mode.
#ifdef ZMQ_HAVE_OPENVMS
int flags = 1;
int rc = ioctl (s, FIONBIO, &flags);
errno_assert (rc != -1);
#else
int flags = fcntl (s, F_GETFL, 0);
if (flags == -1)
flags = 0;
int rc = fcntl (s, F_SETFL, flags | O_NONBLOCK);
errno_assert (rc != -1);
#endif
// Connect to the remote peer.
rc = ::connect (s, (struct sockaddr*) &addr, addr_len);
// Connect was successfull immediately.
if (rc == 0)
return 0;
// Asynchronous connect was launched.
if (rc == -1 && errno == EINPROGRESS) {
errno = EAGAIN;
return -1;
}
// Error occured.
int err = errno;
close ();
errno = err;
// Create the socket.
zmq_assert (AF_UNIX == sa->sa_family);
s = socket (AF_UNIX, SOCK_STREAM, 0);
if (s == -1)
return -1;
}
#ifndef ZMQ_HAVE_OPENVMS
else {
// Set the non-blocking flag.
unblock_socket (s);
// Create the socket.
zmq_assert (AF_UNIX == sa->sa_family);
s = socket (AF_UNIX, SOCK_STREAM, 0);
if (s == -1)
return -1;
// Connect to the remote peer.
int rc = ::connect (s, (struct sockaddr*) &addr, sizeof (sockaddr_un));
// Set the non-blocking flag.
int flag = fcntl (s, F_GETFL, 0);
if (flag == -1)
flag = 0;
int rc = fcntl (s, F_SETFL, flag | O_NONBLOCK);
errno_assert (rc != -1);
// Connect was successfull immediately.
if (rc == 0)
return 0;
// Connect to the remote peer.
rc = ::connect (s, (struct sockaddr*) &addr, sizeof (sockaddr_un));
// Connect was successfull immediately.
if (rc == 0)
return 0;
// Error occured.
int err = errno;
close ();
errno = err;
return -1;
}
#endif
zmq_assert (false);
// Forward the error.
return -1;
}
@@ -379,3 +238,4 @@ zmq::fd_t zmq::ipc_connecter_t::connect ()
}
#endif