1 /*
2 Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
3
4 This file is part of libzmq, the ZeroMQ core engine in C++.
5
6 libzmq is free software; you can redistribute it and/or modify it under
7 the terms of the GNU Lesser General Public License (LGPL) as published
8 by the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
10
11 As a special exception, the Contributors give you permission to link
12 this library with independent modules to produce an executable,
13 regardless of the license terms of these independent modules, and to
14 copy and distribute the resulting executable under terms of your choice,
15 provided that you also meet, for each linked independent module, the
16 terms and conditions of the license of that module. An independent
17 module is a module which is not derived from or based on this library.
18 If you modify this library, you must extend this exception to your
19 version of the library.
20
21 libzmq is distributed in the hope that it will be useful, but WITHOUT
22 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
23 FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
24 License for more details.
25
26 You should have received a copy of the GNU Lesser General Public License
27 along with this program. If not, see <http://www.gnu.org/licenses/>.
28 */
29
30 #include "precompiled.hpp"
31 #include "ipc_connecter.hpp"
32
33 #if defined ZMQ_HAVE_IPC
34
35 #include <new>
36 #include <string>
37
38 #include "io_thread.hpp"
39 #include "random.hpp"
40 #include "err.hpp"
41 #include "ip.hpp"
42 #include "address.hpp"
43 #include "ipc_address.hpp"
44 #include "session_base.hpp"
45
46 #ifdef _MSC_VER
47 #include <afunix.h>
48 #else
49 #include <unistd.h>
50 #include <sys/types.h>
51 #include <sys/socket.h>
52 #include <sys/un.h>
53 #endif
54
ipc_connecter_t(class io_thread_t * io_thread_,class session_base_t * session_,const options_t & options_,address_t * addr_,bool delayed_start_)55 zmq::ipc_connecter_t::ipc_connecter_t (class io_thread_t *io_thread_,
56 class session_base_t *session_,
57 const options_t &options_,
58 address_t *addr_,
59 bool delayed_start_) :
60 stream_connecter_base_t (
61 io_thread_, session_, options_, addr_, delayed_start_)
62 {
63 zmq_assert (_addr->protocol == protocol_name::ipc);
64 }
65
out_event()66 void zmq::ipc_connecter_t::out_event ()
67 {
68 const fd_t fd = connect ();
69 rm_handle ();
70
71 // Handle the error condition by attempt to reconnect.
72 if (fd == retired_fd) {
73 close ();
74 add_reconnect_timer ();
75 return;
76 }
77
78 create_engine (fd, get_socket_name<ipc_address_t> (fd, socket_end_local));
79 }
80
start_connecting()81 void zmq::ipc_connecter_t::start_connecting ()
82 {
83 // Open the connecting socket.
84 const int rc = open ();
85
86 // Connect may succeed in synchronous manner.
87 if (rc == 0) {
88 _handle = add_fd (_s);
89 out_event ();
90 }
91
92 // Connection establishment may be delayed. Poll for its completion.
93 else if (rc == -1 && errno == EINPROGRESS) {
94 _handle = add_fd (_s);
95 set_pollout (_handle);
96 _socket->event_connect_delayed (
97 make_unconnected_connect_endpoint_pair (_endpoint), zmq_errno ());
98
99 // TODO, tcp_connecter_t adds a connect timer in this case; maybe this
100 // should be done here as well (and then this could be pulled up to
101 // stream_connecter_base_t).
102 }
103 //stop connecting after called zmq_disconnect
104 else if (rc == -1
105 && (options.reconnect_stop & ZMQ_RECONNECT_STOP_AFTER_DISCONNECT)
106 && errno == ECONNREFUSED && _socket->is_disconnected ()) {
107 if (_s != retired_fd)
108 close ();
109 }
110
111 // Handle any other error condition by eventual reconnect.
112 else {
113 if (_s != retired_fd)
114 close ();
115 add_reconnect_timer ();
116 }
117 }
118
open()119 int zmq::ipc_connecter_t::open ()
120 {
121 zmq_assert (_s == retired_fd);
122
123 // Create the socket.
124 _s = open_socket (AF_UNIX, SOCK_STREAM, 0);
125 if (_s == retired_fd)
126 return -1;
127
128 // Set the non-blocking flag.
129 unblock_socket (_s);
130
131 // Connect to the remote peer.
132 const int rc = ::connect (_s, _addr->resolved.ipc_addr->addr (),
133 _addr->resolved.ipc_addr->addrlen ());
134
135 // Connect was successful immediately.
136 if (rc == 0)
137 return 0;
138
139 // Translate other error codes indicating asynchronous connect has been
140 // launched to a uniform EINPROGRESS.
141 #ifdef ZMQ_HAVE_WINDOWS
142 const int last_error = WSAGetLastError ();
143 if (last_error == WSAEINPROGRESS || last_error == WSAEWOULDBLOCK)
144 errno = EINPROGRESS;
145 else
146 errno = wsa_error_to_errno (last_error);
147 #else
148 if (rc == -1 && errno == EINTR) {
149 errno = EINPROGRESS;
150 }
151 #endif
152
153 // Forward the error.
154 return -1;
155 }
156
connect()157 zmq::fd_t zmq::ipc_connecter_t::connect ()
158 {
159 // Following code should handle both Berkeley-derived socket
160 // implementations and Solaris.
161 int err = 0;
162 zmq_socklen_t len = static_cast<zmq_socklen_t> (sizeof (err));
163 const int rc = getsockopt (_s, SOL_SOCKET, SO_ERROR,
164 reinterpret_cast<char *> (&err), &len);
165 if (rc == -1) {
166 if (errno == ENOPROTOOPT)
167 errno = 0;
168 err = errno;
169 }
170 if (err != 0) {
171 // Assert if the error was caused by 0MQ bug.
172 // Networking problems are OK. No need to assert.
173 errno = err;
174 errno_assert (errno == ECONNREFUSED || errno == ECONNRESET
175 || errno == ETIMEDOUT || errno == EHOSTUNREACH
176 || errno == ENETUNREACH || errno == ENETDOWN);
177
178 return retired_fd;
179 }
180
181 const fd_t result = _s;
182 _s = retired_fd;
183 return result;
184 }
185
186 #endif
187