diff options
Diffstat (limited to 'cpp/src')
31 files changed, 1317 insertions, 1962 deletions
diff --git a/cpp/src/Ice/IPEndpointI.cpp b/cpp/src/Ice/IPEndpointI.cpp index 68b506fc839..acbfb356715 100644 --- a/cpp/src/Ice/IPEndpointI.cpp +++ b/cpp/src/Ice/IPEndpointI.cpp @@ -14,6 +14,7 @@ #include <Ice/PropertiesI.h> #include <Ice/LoggerUtil.h> #include <Ice/HashUtil.h> +#include <Ice/NetworkProxy.h> #include <IceUtil/MutexPtrLock.h> using namespace std; @@ -558,13 +559,16 @@ IceInternal::EndpointHostResolver::resolve(const string& host, int port, Ice::En vector<ConnectorPtr> connectors; try { + ProtocolSupport protocol = _protocol; if(networkProxy) { - networkProxy = networkProxy->resolveHost(); + networkProxy = networkProxy->resolveHost(_protocol); + if(networkProxy) + { + protocol = networkProxy->getProtocolSupport(); + } } - - connectors = endpoint->connectors(getAddresses(host, port, _protocol, selType, _preferIPv6, true), - networkProxy); + connectors = endpoint->connectors(getAddresses(host, port, protocol, selType, _preferIPv6, true), networkProxy); } catch(const Ice::LocalException& ex) { @@ -666,16 +670,22 @@ IceInternal::EndpointHostResolver::run() } NetworkProxyPtr networkProxy = _instance->networkProxy(); + ProtocolSupport protocol = _protocol; if(networkProxy) { - networkProxy = networkProxy->resolveHost(); + networkProxy = networkProxy->resolveHost(_protocol); + if(networkProxy) + { + protocol = networkProxy->getProtocolSupport(); + } } r.callback->connectors(r.endpoint->connectors(getAddresses(r.host, r.port, - _protocol, + protocol, r.selType, - _preferIPv6, true), + _preferIPv6, + true), networkProxy)); if(threadObserver) @@ -690,6 +700,10 @@ IceInternal::EndpointHostResolver::run() } catch(const Ice::LocalException& ex) { + if(threadObserver) + { + threadObserver->stateChanged(ThreadStateInUseForOther, ThreadStateIdle); + } if(r.observer) { r.observer->failed(ex.ice_name()); diff --git a/cpp/src/Ice/Instance.cpp b/cpp/src/Ice/Instance.cpp index 2bf9555f237..01c9d6d6b06 100644 --- a/cpp/src/Ice/Instance.cpp +++ b/cpp/src/Ice/Instance.cpp @@ -27,6 +27,7 @@ #include <Ice/PropertiesAdminI.h> #include <Ice/LoggerI.h> #include <Ice/Network.h> +#include <Ice/NetworkProxy.h> #include <Ice/EndpointFactoryManager.h> #include <Ice/RetryQueue.h> #include <Ice/DynamicLibrary.h> @@ -1203,21 +1204,8 @@ IceInternal::Instance::Instance(const CommunicatorPtr& communicator, const Initi _proxyFactory = new ProxyFactory(this); - string proxyHost = _initData.properties->getProperty("Ice.SOCKSProxyHost"); - int defaultIPv6 = 1; // IPv6 enabled by default. - if(!proxyHost.empty()) - { -#ifdef ICE_OS_WINRT - throw InitializationException(__FILE__, __LINE__, "SOCKS proxy not supported in WinRT"); -#else - int proxyPort = _initData.properties->getPropertyAsIntWithDefault("Ice.SOCKSProxyPort", 1080); - _networkProxy = new SOCKSNetworkProxy(proxyHost, proxyPort); - defaultIPv6 = 0; // IPv6 is not supported with SOCKS -#endif - } - bool ipv4 = _initData.properties->getPropertyAsIntWithDefault("Ice.IPv4", 1) > 0; - bool ipv6 = _initData.properties->getPropertyAsIntWithDefault("Ice.IPv6", defaultIPv6) > 0; + bool ipv6 = _initData.properties->getPropertyAsIntWithDefault("Ice.IPv6", 1) > 0; if(!ipv4 && !ipv6) { throw InitializationException(__FILE__, __LINE__, "Both IPV4 and IPv6 support cannot be disabled."); @@ -1236,12 +1224,7 @@ IceInternal::Instance::Instance(const CommunicatorPtr& communicator, const Initi } _preferIPv6 = _initData.properties->getPropertyAsInt("Ice.PreferIPv6Address") > 0; -#ifndef ICE_OS_WINRT - if(ipv6 && SOCKSNetworkProxyPtr::dynamicCast(_networkProxy)) - { - throw InitializationException(__FILE__, __LINE__, "IPv6 is not supported with SOCKS4 proxies"); - } -#endif + _networkProxy = IceInternal::createNetworkProxy(_initData.properties, _protocolSupport); _endpointFactoryManager = new EndpointFactoryManager(this); #ifndef ICE_OS_WINRT diff --git a/cpp/src/Ice/Instance.h b/cpp/src/Ice/Instance.h index 19bfda51f89..c774242dfdd 100644 --- a/cpp/src/Ice/Instance.h +++ b/cpp/src/Ice/Instance.h @@ -36,6 +36,7 @@ #include <Ice/DynamicLibraryF.h> #include <Ice/PluginF.h> #include <Ice/NetworkF.h> +#include <Ice/NetworkProxyF.h> #include <Ice/Initialize.h> #include <Ice/ImplicitContextI.h> #include <Ice/FacetMap.h> diff --git a/cpp/src/Ice/Makefile b/cpp/src/Ice/Makefile index 2d40a5578e0..91381c6ad95 100644 --- a/cpp/src/Ice/Makefile +++ b/cpp/src/Ice/Makefile @@ -95,6 +95,7 @@ OBJS = Acceptor.o \ MetricsAdminI.o \ MetricsObserverI.o \ Network.o \ + NetworkProxy.o \ Object.o \ ObjectAdapterFactory.o \ ObjectAdapterI.o \ @@ -125,6 +126,7 @@ OBJS = Acceptor.o \ SlicedData.o \ Stream.o \ StreamI.o \ + StreamSocket.o \ StringConverterPlugin.o \ TcpAcceptor.o \ TcpConnector.o \ diff --git a/cpp/src/Ice/Makefile.mak b/cpp/src/Ice/Makefile.mak index a9465365ec3..eccea2f1a8b 100644 --- a/cpp/src/Ice/Makefile.mak +++ b/cpp/src/Ice/Makefile.mak @@ -97,6 +97,7 @@ OBJS = Acceptor.obj \ MetricsAdminI.obj \ MetricsObserverI.obj \ Network.obj \ + NetworkProxy.obj \ Object.obj \ ObjectAdapterFactory.obj \ ObjectAdapterI.obj \ @@ -127,6 +128,7 @@ OBJS = Acceptor.obj \ SlicedData.obj \ Stream.obj \ StreamI.obj \ + StreamSocket.obj \ StringConverterPlugin.obj \ TcpAcceptor.obj \ TcpConnector.obj \ diff --git a/cpp/src/Ice/Network.cpp b/cpp/src/Ice/Network.cpp index 919826f6e2e..0ad8aa6cefa 100644 --- a/cpp/src/Ice/Network.cpp +++ b/cpp/src/Ice/Network.cpp @@ -19,6 +19,7 @@ #include <IceUtil/DisableWarnings.h> #include <Ice/Network.h> +#include <Ice/NetworkProxy.h> #include <IceUtil/StringUtil.h> #include <IceUtil/StringConverter.h> #include <Ice/LocalException.h> @@ -201,7 +202,6 @@ SOCKET createSocketImpl(bool udp, int family) { SOCKET fd; - if(udp) { fd = socket(family, SOCK_DGRAM, IPPROTO_UDP); @@ -786,118 +786,6 @@ IceInternal::NativeInfo::completed(SocketOperation operation) #endif -IceUtil::Shared* IceInternal::upCast(NetworkProxy* p) { return p; } - -#ifndef ICE_OS_WINRT - -IceInternal::SOCKSNetworkProxy::SOCKSNetworkProxy(const string& host, int port) : - _host(host), _port(port) -{ - assert(!host.empty()); - memset(&_address, 0, sizeof(_address)); -} - -IceInternal::SOCKSNetworkProxy::SOCKSNetworkProxy(const Address& addr) : - _port(0), _address(addr) -{ -} - -void -IceInternal::SOCKSNetworkProxy::beginWriteConnectRequest(const Address& addr, Buffer& buf) -{ - if(addr.saStorage.ss_family != AF_INET) - { - throw FeatureNotSupportedException(__FILE__, __LINE__, "SOCKS4 only supports IPv4 addresses"); - } - - // - // SOCKS connect request - // - buf.b.resize(9); - buf.i = buf.b.begin(); - Byte* dest = &buf.b[0]; - *dest++ = 0x04; // SOCKS version 4. - *dest++ = 0x01; // Command, establish a TCP/IP stream connection - - const Byte* src; - - // - // Port (already in big-endian order) - // - src = reinterpret_cast<const Byte*>(&addr.saIn.sin_port); - *dest++ = *src++; - *dest++ = *src; - - // - // IPv4 address (already in big-endian order) - // - src = reinterpret_cast<const Ice::Byte*>(&addr.saIn.sin_addr.s_addr); - *dest++ = *src++; - *dest++ = *src++; - *dest++ = *src++; - *dest++ = *src; - - *dest = 0x00; // User ID. -} - -void -IceInternal::SOCKSNetworkProxy::endWriteConnectRequest(Buffer& buf) -{ - buf.b.reset(); -} - -void -IceInternal::SOCKSNetworkProxy::beginReadConnectRequestResponse(Buffer& buf) -{ - // - // Read the SOCKS4 response whose size is 8 bytes. - // - buf.b.resize(8); - buf.i = buf.b.begin(); -} - -void -IceInternal::SOCKSNetworkProxy::endReadConnectRequestResponse(Buffer& buf) -{ - buf.i = buf.b.begin(); - - if(buf.b.end() - buf.i < 2) - { - throw UnmarshalOutOfBoundsException(__FILE__, __LINE__); - } - - const Byte* src = &(*buf.i); - const Byte b1 = *src++; - const Byte b2 = *src++; - if(b1 != 0x00 || b2 != 0x5a) - { - throw ConnectFailedException(__FILE__, __LINE__); - } - buf.b.reset(); -} - -NetworkProxyPtr -IceInternal::SOCKSNetworkProxy::resolveHost() const -{ - assert(!_host.empty()); - return new SOCKSNetworkProxy(getAddresses(_host, _port, EnableIPv4, Random, false, true)[0]); -} - -Address -IceInternal::SOCKSNetworkProxy::getAddress() const -{ - assert(_host.empty()); // Host must be resolved. - return _address; -} - -string -IceInternal::SOCKSNetworkProxy::getName() const -{ - return "SOCKS"; -} - -#endif // ICE_OS_WINRT - bool IceInternal::noMoreFds(int error) { @@ -947,7 +835,8 @@ IceInternal::getAddresses(const string& host, int port, ProtocolSupport, Ice::En // to Windows API. // addr.host = ref new HostName(ref new String( - IceUtil::stringToWstring(host, IceUtil::getProcessStringConverter()).c_str())); + IceUtil::stringToWstring(host, + IceUtil::getProcessStringConverter()).c_str())); } stringstream os; os << port; @@ -1400,12 +1289,7 @@ IceInternal::fdToRemoteAddress(SOCKET fd, Address& addr) } std::string -IceInternal::fdToString(SOCKET fd, const NetworkProxyPtr& proxy, const Address& target, -#if defined(_WIN32) - bool connected) -#else - bool /*connected*/) -#endif +IceInternal::fdToString(SOCKET fd, const NetworkProxyPtr& proxy, const Address& target) { if(fd == INVALID_SOCKET) { @@ -1414,8 +1298,11 @@ IceInternal::fdToString(SOCKET fd, const NetworkProxyPtr& proxy, const Address& ostringstream s; -#if defined(_WIN32) - if(!connected) + Address remoteAddr; + bool peerConnected = fdToRemoteAddress(fd, remoteAddr); + +#ifdef _WIN32 + if(!peerConnected) { // // The local address is only accessible with connected sockets on Windows. @@ -1423,19 +1310,12 @@ IceInternal::fdToString(SOCKET fd, const NetworkProxyPtr& proxy, const Address& s << "local address = <not available>"; } else +#endif { Address localAddr; fdToLocalAddress(fd, localAddr); s << "local address = " << addrToString(localAddr); } -#else - Address localAddr; - fdToLocalAddress(fd, localAddr); - s << "local address = " << addrToString(localAddr); -#endif - - Address remoteAddr; - bool peerConnected = fdToRemoteAddress(fd, remoteAddr); if(proxy) { diff --git a/cpp/src/Ice/Network.h b/cpp/src/Ice/Network.h index dfb07151099..179ebab9582 100644 --- a/cpp/src/Ice/Network.h +++ b/cpp/src/Ice/Network.h @@ -17,6 +17,7 @@ #include <Ice/Config.h> #include <Ice/NetworkF.h> +#include <Ice/NetworkProxyF.h> #include <Ice/PropertiesF.h> // For setTcpBufSize #include <Ice/LoggerF.h> // For setTcpBufSize #include <Ice/Protocol.h> @@ -207,73 +208,6 @@ protected: }; typedef IceUtil::Handle<NativeInfo> NativeInfoPtr; -class ICE_API NetworkProxy : virtual public IceUtil::Shared -{ -public: - - // - // Write the connection request on the connection established - // with the network proxy server. This is called right after - // the connection establishment succeeds. - // - virtual void beginWriteConnectRequest(const Address&, Buffer&) = 0; - virtual void endWriteConnectRequest(Buffer&) = 0; - - // - // Once the connection request has been sent, this is called - // to prepare and read the response from the proxy server. - // - virtual void beginReadConnectRequestResponse(Buffer&) = 0; - virtual void endReadConnectRequestResponse(Buffer&) = 0; - - // - // If the proxy host needs to be resolved, this should return - // a new NetworkProxy containing the IP address of the proxy. - // This is called from the endpoint host resolver thread, so - // it's safe if this this method blocks. - // - virtual NetworkProxyPtr resolveHost() const = 0; - - // - // Returns the IP address of the network proxy. This method - // must not block. It's only called on a network proxy object - // returned by resolveHost(). - // - virtual Address getAddress() const = 0; - - // - // Returns the name of the proxy, used for tracing purposes. - // - virtual std::string getName() const = 0; -}; - -#ifndef ICE_OS_WINRT - -class ICE_API SOCKSNetworkProxy : virtual public NetworkProxy -{ -public: - - SOCKSNetworkProxy(const std::string&, int); - SOCKSNetworkProxy(const Address&); - - virtual void beginWriteConnectRequest(const Address&, Buffer&); - virtual void endWriteConnectRequest(Buffer&); - virtual void beginReadConnectRequestResponse(Buffer&); - virtual void endReadConnectRequestResponse(Buffer&); - virtual NetworkProxyPtr resolveHost() const; - virtual Address getAddress() const; - virtual std::string getName() const; - -private: - - std::string _host; - int _port; - Address _address; -}; -typedef IceUtil::Handle<SOCKSNetworkProxy> SOCKSNetworkProxyPtr; - -#endif // ICE_OS_WINRT - ICE_API bool noMoreFds(int); ICE_API std::string errorToStringDNS(int); ICE_API std::vector<Address> getAddresses(const std::string&, int, ProtocolSupport, Ice::EndpointSelectionType, bool, @@ -290,7 +224,7 @@ ICE_API void closeSocket(SOCKET); ICE_API std::string addrToString(const Address&); ICE_API void fdToLocalAddress(SOCKET, Address&); ICE_API bool fdToRemoteAddress(SOCKET, Address&); -ICE_API std::string fdToString(SOCKET, const NetworkProxyPtr&, const Address&, bool); +ICE_API std::string fdToString(SOCKET, const NetworkProxyPtr&, const Address&); ICE_API std::string fdToString(SOCKET); ICE_API void fdToAddressAndPort(SOCKET, std::string&, int&, std::string&, int&); ICE_API void addrToAddressAndPort(const Address&, std::string&, int&); diff --git a/cpp/src/Ice/NetworkF.h b/cpp/src/Ice/NetworkF.h index 143ef4b0b82..45e18911b08 100644 --- a/cpp/src/Ice/NetworkF.h +++ b/cpp/src/Ice/NetworkF.h @@ -23,10 +23,6 @@ struct ICE_API Address; union Address; #endif -class NetworkProxy; -ICE_API IceUtil::Shared* upCast(NetworkProxy*); -typedef Handle<NetworkProxy> NetworkProxyPtr; - } #endif diff --git a/cpp/src/Ice/NetworkProxy.cpp b/cpp/src/Ice/NetworkProxy.cpp new file mode 100644 index 00000000000..c20417b018b --- /dev/null +++ b/cpp/src/Ice/NetworkProxy.cpp @@ -0,0 +1,325 @@ +// ********************************************************************** +// +// Copyright (c) 2003-2014 ZeroC, Inc. All rights reserved. +// +// This copy of Ice is licensed to you under the terms described in the +// ICE_LICENSE file included in this distribution. +// +// ********************************************************************** + +#include <Ice/NetworkProxy.h> +#include <Ice/HttpParser.h> +#include <Ice/LocalException.h> +#include <Ice/Properties.h> + +using namespace std; +using namespace IceInternal; + +IceUtil::Shared* IceInternal::upCast(NetworkProxy* p) { return p; } + +#ifndef ICE_OS_WINRT + +namespace +{ + +class SOCKSNetworkProxy : public NetworkProxy +{ +public: + + SOCKSNetworkProxy(const string&, int); + SOCKSNetworkProxy(const Address&); + + virtual void beginWrite(const Address&, Buffer&); + virtual SocketOperation endWrite(Buffer&); + virtual void beginRead(Buffer&); + virtual SocketOperation endRead(Buffer&); + virtual void finish(Buffer&, Buffer&); + virtual NetworkProxyPtr resolveHost(ProtocolSupport) const; + virtual Address getAddress() const; + virtual string getName() const; + virtual ProtocolSupport getProtocolSupport() const; + +private: + + string _host; + int _port; + Address _address; +}; + +class HTTPNetworkProxy : public NetworkProxy +{ +public: + + HTTPNetworkProxy(const string&, int); + HTTPNetworkProxy(const Address&, ProtocolSupport); + + virtual void beginWrite(const Address&, Buffer&); + virtual SocketOperation endWrite(Buffer&); + virtual void beginRead(Buffer&); + virtual SocketOperation endRead(Buffer&); + virtual void finish(Buffer&, Buffer&); + virtual NetworkProxyPtr resolveHost(ProtocolSupport) const; + virtual Address getAddress() const; + virtual string getName() const; + virtual ProtocolSupport getProtocolSupport() const; + +private: + + string _host; + int _port; + Address _address; + ProtocolSupport _protocol; +}; + +} + +SOCKSNetworkProxy::SOCKSNetworkProxy(const string& host, int port) : _host(host), _port(port) +{ + assert(!host.empty()); + memset(&_address, 0, sizeof(_address)); +} + +SOCKSNetworkProxy::SOCKSNetworkProxy(const Address& addr) : _port(0), _address(addr) +{ +} + +void +SOCKSNetworkProxy::beginWrite(const Address& addr, Buffer& buf) +{ + // + // SOCKS connect request + // + buf.b.resize(9); + buf.i = buf.b.begin(); + Ice::Byte* dest = &buf.b[0]; + *dest++ = 0x04; // SOCKS version 4. + *dest++ = 0x01; // Command, establish a TCP/IP stream connection + + const Ice::Byte* src; + + // + // Port (already in big-endian order) + // + src = reinterpret_cast<const Ice::Byte*>(&addr.saIn.sin_port); + *dest++ = *src++; + *dest++ = *src; + + // + // IPv4 address (already in big-endian order) + // + src = reinterpret_cast<const Ice::Byte*>(&addr.saIn.sin_addr.s_addr); + *dest++ = *src++; + *dest++ = *src++; + *dest++ = *src++; + *dest++ = *src; + + *dest = 0x00; // User ID. +} + +SocketOperation +SOCKSNetworkProxy::endWrite(Buffer& buf) +{ + // Once the request is sent, read the response + return buf.i != buf.b.end() ? SocketOperationWrite : SocketOperationRead; +} + +void +SOCKSNetworkProxy::beginRead(Buffer& buf) +{ + // + // Read the SOCKS4 response whose size is 8 bytes. + // + buf.b.resize(8); + buf.i = buf.b.begin(); +} + +SocketOperation +SOCKSNetworkProxy::endRead(Buffer& buf) +{ + // We're done once we read the response + return buf.i != buf.b.end() ? SocketOperationRead : SocketOperationNone; +} + +void +SOCKSNetworkProxy::finish(Buffer& readBuffer, Buffer&) +{ + readBuffer.i = readBuffer.b.begin(); + + if(readBuffer.b.end() - readBuffer.i < 2) + { + throw Ice::UnmarshalOutOfBoundsException(__FILE__, __LINE__); + } + + const Ice::Byte* src = &(*readBuffer.i); + const Ice::Byte b1 = *src++; + const Ice::Byte b2 = *src++; + if(b1 != 0x00 || b2 != 0x5a) + { + throw Ice::ConnectFailedException(__FILE__, __LINE__); + } +} + +NetworkProxyPtr +SOCKSNetworkProxy::resolveHost(ProtocolSupport protocol) const +{ + assert(!_host.empty()); + return new SOCKSNetworkProxy(getAddresses(_host, _port, protocol, Ice::Random, false, true)[0]); +} + +Address +SOCKSNetworkProxy::getAddress() const +{ + assert(_host.empty()); // Host must be resolved. + return _address; +} + +string +SOCKSNetworkProxy::getName() const +{ + return "SOCKS"; +} + +ProtocolSupport +SOCKSNetworkProxy::getProtocolSupport() const +{ + return EnableIPv4; +} + +HTTPNetworkProxy::HTTPNetworkProxy(const string& host, int port) : + _host(host), _port(port), _protocol(EnableBoth) +{ + assert(!host.empty()); + memset(&_address, 0, sizeof(_address)); +} + +HTTPNetworkProxy::HTTPNetworkProxy(const Address& addr, ProtocolSupport protocol) : + _port(0), _address(addr), _protocol(protocol) +{ +} + +void +HTTPNetworkProxy::beginWrite(const Address& addr, Buffer& buf) +{ + // + // HTTP connect request + // + ostringstream out; + out << "CONNECT " << addrToString(addr) << " HTTP/1.1\r\n" << "Host: " << addrToString(addr) << "\r\n\r\n"; + string str = out.str(); + buf.b.resize(str.size()); + memcpy(&buf.b[0], str.c_str(), str.size()); + buf.i = buf.b.begin(); +} + +SocketOperation +HTTPNetworkProxy::endWrite(Buffer& buf) +{ + // Once the request is sent, read the response + return buf.i != buf.b.end() ? SocketOperationWrite : SocketOperationRead; +} + +void +HTTPNetworkProxy::beginRead(Buffer& buf) +{ + // + // Read the Http response + // + buf.b.resize(7); // Enough space for reading at least HTTP1.1 + buf.i = buf.b.begin(); +} + +SocketOperation +HTTPNetworkProxy::endRead(Buffer& buf) +{ + // + // Check if we received the full HTTP response, if not, continue + // reading otherwise we're done. + // + const Ice::Byte* end = HttpParser().isCompleteMessage(buf.b.begin(), buf.i); + if(!end && buf.i == buf.b.end()) + { + // + // Read one more byte, we can't easily read bytes in advance + // since the transport implenentation might be be able to read + // the data from the memory instead of the socket. This is for + // instance the case with the OpenSSL transport (or we would + // have to use a buffering BIO). + // + buf.b.resize(buf.b.size() + 1); + buf.i = buf.b.begin() + buf.b.size() - 1; + return SocketOperationRead; + } + return SocketOperationNone; +} + +void +HTTPNetworkProxy::finish(Buffer& readBuffer, Buffer&) +{ + HttpParser parser; + parser.parse(readBuffer.b.begin(), readBuffer.b.end()); + if(parser.status() != 200) + { + throw Ice::ConnectFailedException(__FILE__, __LINE__); + } +} + +NetworkProxyPtr +HTTPNetworkProxy::resolveHost(ProtocolSupport protocol) const +{ + assert(!_host.empty()); + return new HTTPNetworkProxy(getAddresses(_host, _port, protocol, Ice::Random, false, true)[0], protocol); +} + +Address +HTTPNetworkProxy::getAddress() const +{ + assert(_host.empty()); // Host must be resolved. + return _address; +} + +string +HTTPNetworkProxy::getName() const +{ + return "HTTP"; +} + +ProtocolSupport +HTTPNetworkProxy::getProtocolSupport() const +{ + return _protocol; +} + +#endif + +NetworkProxyPtr +IceInternal::createNetworkProxy(const Ice::PropertiesPtr& properties, ProtocolSupport protocolSupport) +{ + string proxyHost; + + proxyHost = properties->getProperty("Ice.SOCKSProxyHost"); + if(!proxyHost.empty()) + { +#ifdef ICE_OS_WINRT + throw Ice::InitializationException(__FILE__, __LINE__, "SOCKS proxy not supported with WinRT"); +#else + if(protocolSupport == EnableIPv6) + { + throw Ice::InitializationException(__FILE__, __LINE__, "IPv6 only is not supported with SOCKS4 proxies"); + } + int proxyPort = properties->getPropertyAsIntWithDefault("Ice.SOCKSProxyPort", 1080); + return new SOCKSNetworkProxy(proxyHost, proxyPort); +#endif + } + + proxyHost = properties->getProperty("Ice.HTTPProxyHost"); + if(!proxyHost.empty()) + { +#ifdef ICE_OS_WINRT + throw Ice::InitializationException(__FILE__, __LINE__, "HTTP proxy not supported with WinRT"); +#else + return new HTTPNetworkProxy(proxyHost, properties->getPropertyAsIntWithDefault("Ice.HTTPProxyPort", 1080)); +#endif + } + + return 0; +} diff --git a/cpp/src/Ice/NetworkProxy.h b/cpp/src/Ice/NetworkProxy.h new file mode 100644 index 00000000000..c9b54f04d64 --- /dev/null +++ b/cpp/src/Ice/NetworkProxy.h @@ -0,0 +1,74 @@ +// ********************************************************************** +// +// Copyright (c) 2003-2014 ZeroC, Inc. All rights reserved. +// +// This copy of Ice is licensed to you under the terms described in the +// ICE_LICENSE file included in this distribution. +// +// ********************************************************************** + +#ifndef ICE_NETWORK_PROXY_H +#define ICE_NETWORK_PROXY_H + +#include <Ice/Network.h> + +namespace IceInternal +{ + +class ICE_API NetworkProxy : virtual public IceUtil::Shared +{ +public: + + // + // Write the connection request on the connection established + // with the network proxy server. This is called right after + // the connection establishment succeeds. + // + virtual void beginWrite(const Address&, Buffer&) = 0; + virtual SocketOperation endWrite(Buffer&) = 0; + + // + // Once the connection request has been sent, this is called + // to prepare and read the response from the proxy server. + // + virtual void beginRead(Buffer&) = 0; + virtual SocketOperation endRead(Buffer&) = 0; + + // + // This is called when the response from the proxy has been + // read. The proxy should copy the extra read data (if any) in the + // given byte vector. + // + virtual void finish(Buffer&, Buffer&) = 0; + + // + // If the proxy host needs to be resolved, this should return + // a new NetworkProxy containing the IP address of the proxy. + // This is called from the endpoint host resolver thread, so + // it's safe if this this method blocks. + // + virtual NetworkProxyPtr resolveHost(ProtocolSupport) const = 0; + + // + // Returns the IP address of the network proxy. This method + // must not block. It's only called on a network proxy object + // returned by resolveHost(). + // + virtual Address getAddress() const = 0; + + // + // Returns the name of the proxy, used for tracing purposes. + // + virtual std::string getName() const = 0; + + // + // Returns the protocols supported by the proxy. + // + virtual ProtocolSupport getProtocolSupport() const = 0; +}; + +NetworkProxyPtr createNetworkProxy(const Ice::PropertiesPtr&, ProtocolSupport); + +} + +#endif diff --git a/cpp/src/Ice/NetworkProxyF.h b/cpp/src/Ice/NetworkProxyF.h new file mode 100644 index 00000000000..39d3a444bd7 --- /dev/null +++ b/cpp/src/Ice/NetworkProxyF.h @@ -0,0 +1,26 @@ +// ********************************************************************** +// +// Copyright (c) 2003-2014 ZeroC, Inc. All rights reserved. +// +// This copy of Ice is licensed to you under the terms described in the +// ICE_LICENSE file included in this distribution. +// +// ********************************************************************** + +#ifndef ICE_NETWORK_PROXY_F_H +#define ICE_NETWORK_PROXY_F_H + +#include <IceUtil/Shared.h> + +#include <Ice/Handle.h> + +namespace IceInternal +{ + +class NetworkProxy; +ICE_API IceUtil::Shared* upCast(NetworkProxy*); +typedef Handle<NetworkProxy> NetworkProxyPtr; + +} + +#endif diff --git a/cpp/src/Ice/PropertyNames.cpp b/cpp/src/Ice/PropertyNames.cpp index 47e52f48a2f..73d11ee4f12 100644 --- a/cpp/src/Ice/PropertyNames.cpp +++ b/cpp/src/Ice/PropertyNames.cpp @@ -6,7 +6,7 @@ // ICE_LICENSE file included in this distribution. // // ********************************************************************** -// Generated by makeprops.py from file ../config/PropertyNames.xml, Mon Sep 8 11:41:52 2014 +// Generated by makeprops.py from file ../config/PropertyNames.xml, Tue Sep 9 11:19:09 2014 // IMPORTANT: Do not edit this file -- any edits made here will be lost! @@ -78,6 +78,7 @@ const IceInternal::Property IcePropsData[] = IceInternal::Property("Ice.ChangeUser", false, 0), IceInternal::Property("Ice.ClientAccessPolicyProtocol", false, 0), IceInternal::Property("Ice.Compression.Level", false, 0), + IceInternal::Property("Ice.CollectObjects", false, 0), IceInternal::Property("Ice.Config", false, 0), IceInternal::Property("Ice.ConsoleListener", false, 0), IceInternal::Property("Ice.Default.CollocationOptimized", false, 0), @@ -112,13 +113,14 @@ const IceInternal::Property IcePropsData[] = IceInternal::Property("Ice.Default.SlicedFormat", false, 0), IceInternal::Property("Ice.Default.SourceAddress", false, 0), IceInternal::Property("Ice.Default.Timeout", false, 0), - IceInternal::Property("Ice.IPv4", false, 0), - IceInternal::Property("Ice.IPv6", false, 0), IceInternal::Property("Ice.EventLog.Source", false, 0), IceInternal::Property("Ice.FactoryAssemblies", false, 0), - IceInternal::Property("Ice.CollectObjects", false, 0), + IceInternal::Property("Ice.HTTPProxyHost", false, 0), + IceInternal::Property("Ice.HTTPProxyPort", false, 0), IceInternal::Property("Ice.ImplicitContext", false, 0), IceInternal::Property("Ice.InitPlugins", false, 0), + IceInternal::Property("Ice.IPv4", false, 0), + IceInternal::Property("Ice.IPv6", false, 0), IceInternal::Property("Ice.LogFile", false, 0), IceInternal::Property("Ice.LogStdErr.Convert", false, 0), IceInternal::Property("Ice.MessageSizeMax", false, 0), diff --git a/cpp/src/Ice/PropertyNames.h b/cpp/src/Ice/PropertyNames.h index fda34651d5c..a338c69a8f5 100644 --- a/cpp/src/Ice/PropertyNames.h +++ b/cpp/src/Ice/PropertyNames.h @@ -6,7 +6,7 @@ // ICE_LICENSE file included in this distribution. // // ********************************************************************** -// Generated by makeprops.py from file ../config/PropertyNames.xml, Mon Sep 8 11:41:52 2014 +// Generated by makeprops.py from file ../config/PropertyNames.xml, Tue Sep 9 11:19:09 2014 // IMPORTANT: Do not edit this file -- any edits made here will be lost! diff --git a/cpp/src/Ice/StreamSocket.cpp b/cpp/src/Ice/StreamSocket.cpp new file mode 100644 index 00000000000..3a280e5578f --- /dev/null +++ b/cpp/src/Ice/StreamSocket.cpp @@ -0,0 +1,521 @@ +// ********************************************************************** +// +// Copyright (c) 2003-2014 ZeroC, Inc. All rights reserved. +// +// This copy of Ice is licensed to you under the terms described in the +// ICE_LICENSE file included in this distribution. +// +// ********************************************************************** + +#include <Ice/StreamSocket.h> +#include <Ice/NetworkProxy.h> +#include <Ice/ProtocolInstance.h> + +using namespace IceInternal; + +StreamSocket::StreamSocket(const ProtocolInstancePtr& instance, + const NetworkProxyPtr& proxy, + const Address& addr, + const Address& sourceAddr) : + NativeInfo(createSocket(false, proxy ? proxy->getAddress() : addr)), + _proxy(proxy), + _addr(addr), + _sourceAddr(sourceAddr), + _state(StateNeedConnect) +#ifdef ICE_USE_IOCP + , _read(SocketOperationRead), + _write(SocketOperationWrite) +#endif +{ + init(instance); +#ifndef ICE_USE_IOCP + if(doConnect(_fd, _proxy ? _proxy->getAddress() : _addr, sourceAddr)) + { + _state = StateConnected; + } +#endif + _desc = fdToString(_fd, _proxy, _addr); +} + +StreamSocket::StreamSocket(const ProtocolInstancePtr& instance, SOCKET fd) : + NativeInfo(fd), + _state(StateConnected) +#ifdef ICE_USE_IOCP + , _read(SocketOperationRead), + _write(SocketOperationWrite) +#endif +{ + init(instance); + _desc = fdToString(fd); +} + +StreamSocket::~StreamSocket() +{ + assert(_fd == INVALID_SOCKET); +} + +SocketOperation +StreamSocket::connect(Buffer& readBuffer, Buffer& writeBuffer) +{ + if(_state == StateNeedConnect) + { + _state = StateConnectPending; + return SocketOperationConnect; + } + else if(_state <= StateConnectPending) + { +#ifdef ICE_USE_IOCP + doFinishConnectAsync(_fd, _write); +#else + doFinishConnect(_fd); +#endif + _desc = fdToString(_fd, _proxy, _addr); + _state = _proxy ? StateProxyWrite : StateConnected; + } + + if(_state == StateProxyWrite) + { + _proxy->beginWrite(_addr, writeBuffer); + return IceInternal::SocketOperationWrite; + } + else if(_state == StateProxyRead) + { + _proxy->beginRead(readBuffer); + return IceInternal::SocketOperationRead; + } + else if(_state == StateProxyConnected) + { + _proxy->finish(readBuffer, writeBuffer); + + readBuffer.b.clear(); + readBuffer.i = readBuffer.b.end(); + + writeBuffer.b.clear(); + writeBuffer.i = writeBuffer.b.end(); + + _state = StateConnected; + } + + assert(_state == StateConnected); + return IceInternal::SocketOperationNone; +} + +bool +StreamSocket::isConnected() +{ + return _state == StateConnected; +} + +size_t +StreamSocket::getSendPacketSize(size_t length) +{ +#ifdef ICE_USE_IOCP + return _maxSendPacketSize > 0 ? std::min(length, _maxSendPacketSize) : length; +#else + return length; +#endif +} + +size_t +StreamSocket::getRecvPacketSize(size_t length) +{ +#ifdef ICE_USE_IOCP + return _maxRecvPacketSize > 0 ? std::min(length, _maxRecvPacketSize) : length; +#else + return length; +#endif +} + +SocketOperation +StreamSocket::read(Buffer& buf) +{ + if(_state == StateProxyRead) + { + while(true) + { + ssize_t ret = read(reinterpret_cast<char*>(&*buf.i), buf.b.end() - buf.i); + if(ret == 0) + { + return SocketOperationRead; + } + buf.i += ret; + _state = toState(_proxy->endRead(buf)); + if(_state != StateProxyRead) + { + return SocketOperationNone; + } + } + } + buf.i += read(reinterpret_cast<char*>(&*buf.i), buf.b.end() - buf.i); + return buf.i != buf.b.end() ? SocketOperationRead : SocketOperationNone; +} + +SocketOperation +StreamSocket::write(Buffer& buf) +{ + if(_state == StateProxyWrite) + { + while(true) + { + ssize_t ret = write(reinterpret_cast<const char*>(&*buf.i), buf.b.end() - buf.i); + if(ret == 0) + { + return SocketOperationWrite; + } + buf.i += ret; + _state = toState(_proxy->endWrite(buf)); + if(_state != StateProxyWrite) + { + return SocketOperationNone; + } + } + } + buf.i += write(reinterpret_cast<const char*>(&*buf.i), buf.b.end() - buf.i); + return buf.i != buf.b.end() ? SocketOperationWrite : SocketOperationNone; +} + +ssize_t +StreamSocket::read(char* buf, size_t length) +{ + assert(_fd != INVALID_SOCKET); + + size_t packetSize = length; + ssize_t read = 0; + + while(length > 0) + { +#ifdef _WIN32 + ssize_t ret = ::recv(_fd, buf, static_cast<int>(packetSize), 0); +#else + ssize_t ret = ::recv(_fd, buf, packetSize, 0); +#endif + if(ret == 0) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = 0; + throw ex; + } + else if(ret == SOCKET_ERROR) + { + if(interrupted()) + { + continue; + } + + if(noBuffers() && packetSize > 1024) + { + packetSize /= 2; + continue; + } + + if(wouldBlock()) + { + return read; + } + + if(connectionLost()) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + else + { + Ice::SocketException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + } + + buf += ret; + read += ret; + length -= ret; + + if(packetSize > length) + { + packetSize = length; + } + } + return read; +} + +ssize_t +StreamSocket::write(const char* buf, size_t length) +{ + assert(_fd != INVALID_SOCKET); + +#ifdef ICE_USE_IOCP + // + // On Windows, limiting the buffer size is important to prevent + // poor throughput performances when sending large amount of + // data. See Microsoft KB article KB823764. + // + size_t packetSize = _maxSendPacketSize > 0 ? std::min(length, _maxSendPacketSize / 2) : length; +#else + size_t packetSize = length; +#endif + + ssize_t sent = 0; + while(length > 0) + { +#ifdef _WIN32 + ssize_t ret = ::send(_fd, buf, static_cast<int>(packetSize), 0); +#else + ssize_t ret = ::send(_fd, buf, packetSize, 0); +#endif + if(ret == 0) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = 0; + throw ex; + } + else if(ret == SOCKET_ERROR) + { + if(interrupted()) + { + continue; + } + + if(noBuffers() && packetSize > 1024) + { + packetSize /= 2; + continue; + } + + if(wouldBlock()) + { + return sent; + } + + if(connectionLost()) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + else + { + Ice::SocketException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + } + + buf += ret; + sent += ret; + length -= ret; + + if(packetSize > length) + { + packetSize = length; + } + } + return sent; +} + +#ifdef ICE_USE_IOCP +AsyncInfo* +StreamSocket::getAsyncInfo(SocketOperation op) +{ + switch(op) + { + case SocketOperationRead: + return &_read; + case SocketOperationWrite: + return &_write; + default: + assert(false); + return 0; + } +} + +bool +StreamSocket::startWrite(Buffer& buf) +{ + if(_state == StateConnectPending) + { + doConnectAsync(_fd, _proxy ? _proxy->getAddress() : _addr, _sourceAddr, _write); + return false; + } + + size_t length = buf.b.end() - buf.i; + assert(length > 0); + size_t packetSize = getSendPacketSize(length); + + _write.buf.len = static_cast<DWORD>(packetSize); + _write.buf.buf = reinterpret_cast<char*>(&*buf.i); + int err = WSASend(_fd, &_write.buf, 1, &_write.count, 0, &_write, NULL); + if(err == SOCKET_ERROR) + { + if(!wouldBlock()) + { + if(connectionLost()) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + else + { + Ice::SocketException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + } + } + return packetSize == length; +} + +void +StreamSocket::finishWrite(Buffer& buf) +{ + if(_state < StateConnected && _state != StateProxyWrite) + { + return; + } + + if(static_cast<int>(_write.count) == SOCKET_ERROR) + { + WSASetLastError(_write.error); + if(connectionLost()) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + else + { + Ice::SocketException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + } + + buf.i += _write.count; + if(_state == StateProxyWrite) + { + _state = toState(_proxy->endWrite(buf)); + } +} + +void +StreamSocket::startRead(Buffer& buf) +{ + size_t length = buf.b.end() - buf.i; + assert(length > 0); + + size_t packetSize = getRecvPacketSize(length); + _read.buf.len = static_cast<DWORD>(packetSize); + _read.buf.buf = reinterpret_cast<char*>(&*buf.i); + int err = WSARecv(_fd, &_read.buf, 1, &_read.count, &_read.flags, &_read, NULL); + if(err == SOCKET_ERROR) + { + if(!wouldBlock()) + { + if(connectionLost()) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + else + { + Ice::SocketException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + } + } +} + +void +StreamSocket::finishRead(Buffer& buf) +{ + if(static_cast<int>(_read.count) == SOCKET_ERROR) + { + WSASetLastError(_read.error); + if(connectionLost()) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + else + { + Ice::SocketException ex(__FILE__, __LINE__); + ex.error = getSocketErrno(); + throw ex; + } + } + else if(_read.count == 0) + { + Ice::ConnectionLostException ex(__FILE__, __LINE__); + ex.error = 0; + throw ex; + } + + buf.i += _read.count; + + if(_state == StateProxyRead) + { + _state = toState(_proxy->endRead(buf)); + } +} +#endif + +void +StreamSocket::close() +{ + assert(_fd != INVALID_SOCKET); + try + { + closeSocket(_fd); + _fd = INVALID_SOCKET; + } + catch(const Ice::SocketException&) + { + _fd = INVALID_SOCKET; + throw; + } +} + +const std::string& +StreamSocket::toString() const +{ + return _desc; +} + +void +StreamSocket::init(const ProtocolInstancePtr& instance) +{ + setBlock(_fd, false); + setTcpBufSize(_fd, instance->properties(), instance->logger()); + +#ifdef ICE_USE_IOCP + // + // For timeouts to work properly, we need to receive or send the + // data in several chunks when using IOCP WSARecv or WSASend. + // Otherwise, we would only be notified when all the data is + // received or written. The connection timeout could easily be + // triggered when receiging or sending large messages. + // + _maxSendPacketSize = std::max(512, IceInternal::getSendBufferSize(_fd)); + _maxRecvPacketSize = std::max(512, IceInternal::getRecvBufferSize(_fd)); +#endif +} + +StreamSocket::State +StreamSocket::toState(SocketOperation operation) const +{ + switch(operation) + { + case SocketOperationRead: + return StateProxyRead; + case SocketOperationWrite: + return StateProxyWrite; + default: + return StateProxyConnected; + } +} + + diff --git a/cpp/src/Ice/StreamSocket.h b/cpp/src/Ice/StreamSocket.h new file mode 100644 index 00000000000..7a872599944 --- /dev/null +++ b/cpp/src/Ice/StreamSocket.h @@ -0,0 +1,85 @@ +// ********************************************************************** +// +// Copyright (c) 2003-2014 ZeroC, Inc. All rights reserved. +// +// This copy of Ice is licensed to you under the terms described in the +// ICE_LICENSE file included in this distribution. +// +// ********************************************************************** + +#ifndef ICE_STREAM_SOCKET_H +#define ICE_STREAM_SOCKET_H + +#include <IceUtil/Shared.h> +#include <Ice/Network.h> +#include <Ice/Buffer.h> +#include <Ice/ProtocolInstanceF.h> + +namespace IceInternal +{ + +class ICE_API StreamSocket : public NativeInfo +{ +public: + + StreamSocket(const ProtocolInstancePtr&, const NetworkProxyPtr&, const Address&, const Address&); + StreamSocket(const ProtocolInstancePtr&, SOCKET); + virtual ~StreamSocket(); + + SocketOperation connect(Buffer&, Buffer&); + bool isConnected(); + size_t getSendPacketSize(size_t); + size_t getRecvPacketSize(size_t); + + SocketOperation read(Buffer&); + SocketOperation write(Buffer&); + + ssize_t read(char*, size_t); + ssize_t write(const char*, size_t); + +#ifdef ICE_USE_IOCP + AsyncInfo* getAsyncInfo(SocketOperation); + bool startWrite(Buffer&); + void finishWrite(Buffer&); + void startRead(Buffer&); + void finishRead(Buffer&); +#endif + + void close(); + const std::string& toString() const; + +private: + + void init(const ProtocolInstancePtr&); + + enum State + { + StateNeedConnect, + StateConnectPending, + StateProxyWrite, + StateProxyRead, + StateProxyConnected, + StateConnected + }; + State toState(SocketOperation) const; + + const NetworkProxyPtr _proxy; + const Address _addr; + const Address _sourceAddr; + + State _state; + std::string _desc; + +#ifdef ICE_USE_IOCP + size_t _maxSendPacketSize; + size_t _maxRecvPacketSize; + AsyncInfo _read; + AsyncInfo _write; +#endif +}; +typedef IceUtil::Handle<StreamSocket> StreamSocketPtr; + +} + +#endif + diff --git a/cpp/src/Ice/TcpAcceptor.cpp b/cpp/src/Ice/TcpAcceptor.cpp index c77afdef2b1..10605b1c5dc 100644 --- a/cpp/src/Ice/TcpAcceptor.cpp +++ b/cpp/src/Ice/TcpAcceptor.cpp @@ -14,6 +14,7 @@ #include <Ice/LoggerUtil.h> #include <Ice/LocalException.h> #include <Ice/Properties.h> +#include <Ice/StreamSocket.h> #include <IceUtil/StringUtil.h> #ifdef ICE_USE_IOCP @@ -144,7 +145,7 @@ IceInternal::TcpAcceptor::accept() SOCKET fd = doAccept(_fd); #endif - return new TcpTransceiver(_instance, fd); + return new TcpTransceiver(_instance, new StreamSocket(_instance, fd)); } string diff --git a/cpp/src/Ice/TcpConnector.cpp b/cpp/src/Ice/TcpConnector.cpp index cf79f0dd5df..3c4deabea3f 100644 --- a/cpp/src/Ice/TcpConnector.cpp +++ b/cpp/src/Ice/TcpConnector.cpp @@ -13,6 +13,8 @@ #include <Ice/ProtocolInstance.h> #include <Ice/LoggerUtil.h> #include <Ice/Network.h> +#include <Ice/NetworkProxy.h> +#include <Ice/StreamSocket.h> #include <Ice/Exception.h> using namespace std; @@ -22,10 +24,7 @@ using namespace IceInternal; TransceiverPtr IceInternal::TcpConnector::connect() { - TransceiverPtr transceiver = new TcpTransceiver(_instance, createSocket(false, _addr), _proxy, _addr, _sourceAddr); - dynamic_cast<TcpTransceiver*>(transceiver.get())->connect(); - return transceiver; - + return new TcpTransceiver(_instance, new StreamSocket(_instance, _proxy, _addr, _sourceAddr)); } Short diff --git a/cpp/src/Ice/TcpTransceiver.cpp b/cpp/src/Ice/TcpTransceiver.cpp index 76440f5f935..9f7a04c9496 100644 --- a/cpp/src/Ice/TcpTransceiver.cpp +++ b/cpp/src/Ice/TcpTransceiver.cpp @@ -23,121 +23,13 @@ using namespace IceInternal; NativeInfoPtr IceInternal::TcpTransceiver::getNativeInfo() { - return this; + return _stream; } -#ifdef ICE_USE_IOCP -AsyncInfo* -IceInternal::TcpTransceiver::getAsyncInfo(SocketOperation status) -{ - switch(status) - { - case SocketOperationRead: - return &_read; - case SocketOperationWrite: - return &_write; - default: - assert(false); - return 0; - } -} -#endif - SocketOperation -IceInternal::TcpTransceiver::initialize(Buffer& readBuffer, Buffer& writeBuffer, bool& hasMoreData) +IceInternal::TcpTransceiver::initialize(Buffer& readBuffer, Buffer& writeBuffer, bool&) { - if(_state == StateNeedConnect) - { - _state = StateConnectPending; - return SocketOperationConnect; - } - else if(_state <= StateConnectPending) - { -#ifdef ICE_USE_IOCP - doFinishConnectAsync(_fd, _write); -#else - doFinishConnect(_fd); -#endif - - _desc = fdToString(_fd, _proxy, _addr, true); - - if(_proxy) - { - // - // Prepare the read & write buffers in advance. - // - _proxy->beginWriteConnectRequest(_addr, writeBuffer); - _proxy->beginReadConnectRequestResponse(readBuffer); - -#ifdef ICE_USE_IOCP - // - // Return SocketOperationWrite to indicate we need to start a write. - // - _state = StateProxyConnectRequest; // Send proxy connect request - return IceInternal::SocketOperationWrite; -#else - // - // Write the proxy connection message. - // - if(write(writeBuffer)) - { - // - // Write completed without blocking. - // - _proxy->endWriteConnectRequest(writeBuffer); - - // - // Try to read the response. - // - if(read(readBuffer, hasMoreData)) - { - // - // Read completed without blocking - fall through. - // - _proxy->endReadConnectRequestResponse(readBuffer); - } - else - { - // - // Return SocketOperationRead to indicate we need to complete the read. - // - _state = StateProxyConnectRequestPending; // Wait for proxy response - return SocketOperationRead; - } - } - else - { - // - // Return SocketOperationWrite to indicate we need to complete the write. - // - _state = StateProxyConnectRequest; // Send proxy connect request - return SocketOperationWrite; - } -#endif - } - - _state = StateConnected; - } - else if(_state == StateProxyConnectRequest) - { - // - // Write completed. - // - _proxy->endWriteConnectRequest(writeBuffer); - _state = StateProxyConnectRequestPending; // Wait for proxy response - return SocketOperationRead; - } - else if(_state == StateProxyConnectRequestPending) - { - // - // Read completed. - // - _proxy->endReadConnectRequestResponse(readBuffer); - _state = StateConnected; - } - - assert(_state == StateConnected); - return SocketOperationNone; + return _stream->connect(readBuffer, writeBuffer); } SocketOperation @@ -151,290 +43,44 @@ IceInternal::TcpTransceiver::closing(bool initiator, const Ice::LocalException&) void IceInternal::TcpTransceiver::close() { - assert(_fd != INVALID_SOCKET); - try - { - closeSocket(_fd); - _fd = INVALID_SOCKET; - } - catch(const SocketException&) - { - _fd = INVALID_SOCKET; - throw; - } + _stream->close(); } SocketOperation IceInternal::TcpTransceiver::write(Buffer& buf) { - if(buf.i == buf.b.end()) - { - return SocketOperationNone; - } - - // - // It's impossible for packetSize to be more than an Int. - // - int packetSize = static_cast<int>(buf.b.end() - buf.i); -#ifdef ICE_USE_IOCP - // - // Limit packet size to avoid performance problems on WIN32 - // - if(_maxSendPacketSize > 0 && packetSize > _maxSendPacketSize) - { - packetSize = _maxSendPacketSize; - } -#endif - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - - ssize_t ret = ::send(_fd, reinterpret_cast<const char*>(&*buf.i), packetSize, 0); - if(ret == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - if(ret == SOCKET_ERROR) - { - if(interrupted()) - { - continue; - } - - if(noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(wouldBlock()) - { - return SocketOperationWrite; - } - - if(connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - } - - buf.i += ret; - - if(packetSize > buf.b.end() - buf.i) - { - packetSize = static_cast<int>(buf.b.end() - buf.i); - } - } - - return SocketOperationNone; + return _stream->write(buf); } SocketOperation IceInternal::TcpTransceiver::read(Buffer& buf, bool&) { - if(buf.i == buf.b.end()) - { - return SocketOperationNone; - } - - // - // It's impossible for packetSize to be more than an Int. - // - int packetSize = static_cast<int>(buf.b.end() - buf.i); - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - ssize_t ret = ::recv(_fd, reinterpret_cast<char*>(&*buf.i), packetSize, 0); - - if(ret == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - if(ret == SOCKET_ERROR) - { - if(interrupted()) - { - continue; - } - - if(noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(wouldBlock()) - { - return SocketOperationRead; - } - - if(connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - } - - buf.i += ret; - - packetSize = static_cast<int>(buf.b.end() - buf.i); - } - return SocketOperationNone; + return _stream->read(buf); } #ifdef ICE_USE_IOCP bool IceInternal::TcpTransceiver::startWrite(Buffer& buf) { - if(_state == StateConnectPending) - { - Address addr = _proxy ? _proxy->getAddress() : _addr; - doConnectAsync(_fd, addr, _sourceAddr, _write); - return false; - } - - assert(!buf.b.empty()); - assert(buf.i != buf.b.end()); - - int packetSize = static_cast<int>(buf.b.end() - buf.i); - if(_maxSendPacketSize > 0 && packetSize > _maxSendPacketSize) - { - packetSize = _maxSendPacketSize; - } - assert(packetSize > 0); - _write.buf.len = packetSize; - _write.buf.buf = reinterpret_cast<char*>(&*buf.i); - int err = WSASend(_fd, &_write.buf, 1, &_write.count, 0, &_write, NULL); - if(err == SOCKET_ERROR) - { - if(!wouldBlock()) - { - if(connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - } - } - return packetSize == static_cast<int>(buf.b.end() - buf.i); + return _stream->startWrite(buf); } void IceInternal::TcpTransceiver::finishWrite(Buffer& buf) { - if(_state < StateConnected && _state != StateProxyConnectRequest) - { - return; - } - - if(static_cast<int>(_write.count) == SOCKET_ERROR) - { - WSASetLastError(_write.error); - if(connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - } - - buf.i += _write.count; + _stream->finishWrite(buf); } void IceInternal::TcpTransceiver::startRead(Buffer& buf) { - int packetSize = static_cast<int>(buf.b.end() - buf.i); - if(_maxReceivePacketSize > 0 && packetSize > _maxReceivePacketSize) - { - packetSize = _maxReceivePacketSize; - } - assert(!buf.b.empty() && buf.i != buf.b.end()); - - _read.buf.len = packetSize; - _read.buf.buf = reinterpret_cast<char*>(&*buf.i); - int err = WSARecv(_fd, &_read.buf, 1, &_read.count, &_read.flags, &_read, NULL); - if(err == SOCKET_ERROR) - { - if(!wouldBlock()) - { - if(connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - } - } + _stream->startRead(buf); } void IceInternal::TcpTransceiver::finishRead(Buffer& buf, bool&) { - if(static_cast<int>(_read.count) == SOCKET_ERROR) - { - WSASetLastError(_read.error); - if(connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = getSocketErrno(); - throw ex; - } - } - else if(_read.count == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - buf.i += _read.count; + _stream->finishRead(buf); } #endif @@ -447,7 +93,7 @@ IceInternal::TcpTransceiver::protocol() const string IceInternal::TcpTransceiver::toString() const { - return _desc; + return _stream->toString(); } string @@ -460,7 +106,7 @@ Ice::ConnectionInfoPtr IceInternal::TcpTransceiver::getInfo() const { Ice::TCPConnectionInfoPtr info = new Ice::TCPConnectionInfo(); - fdToAddressAndPort(_fd, info->localAddress, info->localPort, info->remoteAddress, info->remotePort); + fdToAddressAndPort(_stream->fd(), info->localAddress, info->localPort, info->remoteAddress, info->remotePort); return info; } @@ -473,104 +119,13 @@ IceInternal::TcpTransceiver::checkSendSize(const Buffer& buf, size_t messageSize } } -IceInternal::TcpTransceiver::TcpTransceiver(const ProtocolInstancePtr& instance, SOCKET fd, - const NetworkProxyPtr& proxy, const Address& addr, - const Address& sourceAddr) : - NativeInfo(fd), - _instance(instance), - _proxy(proxy), - _addr(addr), - _sourceAddr(sourceAddr), - _state(StateNeedConnect) -#ifdef ICE_USE_IOCP - , _read(SocketOperationRead), - _write(SocketOperationWrite) -#endif +IceInternal::TcpTransceiver::TcpTransceiver(const ProtocolInstancePtr& instance, const StreamSocketPtr& stream) : + _instance(instance), + _stream(stream) { - setBlock(_fd, false); - - setTcpBufSize(_fd, _instance->properties(), _instance->logger()); - -#ifdef ICE_USE_IOCP - // - // On Windows, limiting the buffer size is important to prevent - // poor throughput performances when transfering large amount of - // data. See Microsoft KB article KB823764. - // - _maxSendPacketSize = IceInternal::getSendBufferSize(fd) / 2; - if(_maxSendPacketSize < 512) - { - _maxSendPacketSize = 0; - } - - _maxReceivePacketSize = IceInternal::getRecvBufferSize(fd); - if(_maxReceivePacketSize < 512) - { - _maxReceivePacketSize = 0; - } -#endif -} - -IceInternal::TcpTransceiver::TcpTransceiver(const ProtocolInstancePtr& instance, SOCKET fd) : - NativeInfo(fd), - _instance(instance), - _state(StateConnected), - _desc(fdToString(_fd)) -#ifdef ICE_USE_IOCP - , _read(SocketOperationRead), - _write(SocketOperationWrite) -#endif -{ - setBlock(_fd, false); - - setTcpBufSize(_fd, _instance->properties(), _instance->logger()); - -#ifdef ICE_USE_IOCP - // - // On Windows, limiting the buffer size is important to prevent - // poor throughput performances when transfering large amount of - // data. See Microsoft KB article KB823764. - // - _maxSendPacketSize = IceInternal::getSendBufferSize(fd) / 2; - if(_maxSendPacketSize < 512) - { - _maxSendPacketSize = 0; - } - - _maxReceivePacketSize = IceInternal::getRecvBufferSize(fd); - if(_maxReceivePacketSize < 512) - { - _maxReceivePacketSize = 0; - } -#endif } IceInternal::TcpTransceiver::~TcpTransceiver() { - assert(_fd == INVALID_SOCKET); } -void -IceInternal::TcpTransceiver::connect() -{ -#ifndef ICE_USE_IOCP - try - { - Address addr = _proxy ? _proxy->getAddress() : _addr; - if(doConnect(_fd, addr, _sourceAddr)) - { - _state = StateConnected; - _desc = fdToString(_fd, _proxy, _addr, true); - } - else - { - _desc = fdToString(_fd, _proxy, _addr, true); - } - } - catch(...) - { - _fd = INVALID_SOCKET; - throw; - } -#endif -} diff --git a/cpp/src/Ice/TcpTransceiver.h b/cpp/src/Ice/TcpTransceiver.h index 0e1189a3d31..a3a170b85ee 100644 --- a/cpp/src/Ice/TcpTransceiver.h +++ b/cpp/src/Ice/TcpTransceiver.h @@ -13,6 +13,7 @@ #include <Ice/ProtocolInstanceF.h> #include <Ice/Transceiver.h> #include <Ice/Network.h> +#include <Ice/StreamSocket.h> namespace IceInternal { @@ -20,23 +21,11 @@ namespace IceInternal class TcpConnector; class TcpAcceptor; -class TcpTransceiver : public Transceiver, public NativeInfo +class TcpTransceiver : public Transceiver { - enum State - { - StateNeedConnect, - StateConnectPending, - StateProxyConnectRequest, - StateProxyConnectRequestPending, - StateConnected - }; - public: virtual NativeInfoPtr getNativeInfo(); -#ifdef ICE_USE_IOCP - virtual AsyncInfo* getAsyncInfo(SocketOperation); -#endif virtual SocketOperation initialize(Buffer&, Buffer&, bool&); virtual SocketOperation closing(bool, const Ice::LocalException&); @@ -57,29 +46,14 @@ public: private: - TcpTransceiver(const ProtocolInstancePtr&, SOCKET, const NetworkProxyPtr&, const Address&, const Address&); - TcpTransceiver(const ProtocolInstancePtr&, SOCKET); + TcpTransceiver(const ProtocolInstancePtr&, const StreamSocketPtr&); virtual ~TcpTransceiver(); - void connect(); - friend class TcpConnector; friend class TcpAcceptor; const ProtocolInstancePtr _instance; - const NetworkProxyPtr _proxy; - const Address _addr; - const Address _sourceAddr; - - State _state; - std::string _desc; - -#ifdef ICE_USE_IOCP - AsyncInfo _read; - AsyncInfo _write; - int _maxSendPacketSize; - int _maxReceivePacketSize; -#endif + const StreamSocketPtr _stream; }; } diff --git a/cpp/src/Ice/ThreadPool.h b/cpp/src/Ice/ThreadPool.h index b7dcbf3cf42..15c913c6c07 100644 --- a/cpp/src/Ice/ThreadPool.h +++ b/cpp/src/Ice/ThreadPool.h @@ -191,7 +191,7 @@ public: #else bool ioReady() { - return _handler->_registered & operation; + return (_handler->_registered & operation) != 0; } #endif diff --git a/cpp/src/Ice/Transceiver.cpp b/cpp/src/Ice/Transceiver.cpp index 962aefeca1d..7dc7723490c 100644 --- a/cpp/src/Ice/Transceiver.cpp +++ b/cpp/src/Ice/Transceiver.cpp @@ -18,6 +18,6 @@ IceUtil::Shared* IceInternal::upCast(Transceiver* p) { return p; } EndpointIPtr IceInternal::Transceiver::bind(const EndpointIPtr&) { - assert(false); - return 0; + assert(false); + return 0; } diff --git a/cpp/src/Ice/winrt/Makefile.mak b/cpp/src/Ice/winrt/Makefile.mak index f517f66aa4f..95965d8d271 100644 --- a/cpp/src/Ice/winrt/Makefile.mak +++ b/cpp/src/Ice/winrt/Makefile.mak @@ -62,6 +62,7 @@ OBJS = $(ARCH)\$(CONFIG)\Acceptor.obj \ $(ARCH)\$(CONFIG)\MetricsAdminI.obj \ $(ARCH)\$(CONFIG)\MetricsObserverI.obj \ $(ARCH)\$(CONFIG)\Network.obj \ + $(ARCH)\$(CONFIG)\NetworkProxy.obj \ $(ARCH)\$(CONFIG)\ObjectAdapterFactory.obj \ $(ARCH)\$(CONFIG)\ObjectAdapterI.obj \ $(ARCH)\$(CONFIG)\ObjectAdapter.obj \ diff --git a/cpp/src/IceSSL/AcceptorI.cpp b/cpp/src/IceSSL/AcceptorI.cpp index f55f1b2b2f4..2cb3f19515b 100644 --- a/cpp/src/IceSSL/AcceptorI.cpp +++ b/cpp/src/IceSSL/AcceptorI.cpp @@ -22,6 +22,7 @@ #include <Ice/LocalException.h> #include <Ice/LoggerUtil.h> #include <Ice/Properties.h> +#include <Ice/StreamSocket.h> #include <IceUtil/StringUtil.h> #ifdef ICE_USE_IOCP @@ -41,13 +42,8 @@ IceSSL::AcceptorI::getNativeInfo() #ifdef ICE_USE_IOCP IceInternal::AsyncInfo* -# ifndef NDEBUG -IceSSL::AcceptorI::getAsyncInfo(IceInternal::SocketOperation status) -# else IceSSL::AcceptorI::getAsyncInfo(IceInternal::SocketOperation) -# endif { - assert(status == IceInternal::SocketOperationRead); return &_info; } #endif @@ -167,7 +163,7 @@ IceSSL::AcceptorI::accept() // SSL handshaking is performed in TransceiverI::initialize, since // accept must not block. // - return new TransceiverI(_instance, fd, _adapterName); + return new TransceiverI(_instance, new IceInternal::StreamSocket(_instance, fd), "", _adapterName); } string diff --git a/cpp/src/IceSSL/ConnectorI.cpp b/cpp/src/IceSSL/ConnectorI.cpp index e7914663088..a31b25fee01 100644 --- a/cpp/src/IceSSL/ConnectorI.cpp +++ b/cpp/src/IceSSL/ConnectorI.cpp @@ -18,7 +18,9 @@ #include <IceSSL/Util.h> #include <Ice/Communicator.h> #include <Ice/LocalException.h> +#include <Ice/StreamSocket.h> #include <Ice/LoggerUtil.h> +#include <Ice/NetworkProxy.h> using namespace std; using namespace Ice; @@ -36,7 +38,9 @@ IceSSL::ConnectorI::connect() ex.reason = "IceSSL: plug-in is not initialized"; throw ex; } - return new TransceiverI(_instance, IceInternal::createSocket(false, _addr), _proxy, _host, _addr, _sourceAddr); + + IceInternal::StreamSocketPtr stream = new IceInternal::StreamSocket(_instance, _proxy, _addr, _sourceAddr); + return new TransceiverI(_instance, stream, _host, ""); } Short diff --git a/cpp/src/IceSSL/OpenSSLTransceiverI.cpp b/cpp/src/IceSSL/OpenSSLTransceiverI.cpp index 8df471c87eb..22f2412a845 100644 --- a/cpp/src/IceSSL/OpenSSLTransceiverI.cpp +++ b/cpp/src/IceSSL/OpenSSLTransceiverI.cpp @@ -68,98 +68,24 @@ Init init; IceInternal::NativeInfoPtr IceSSL::TransceiverI::getNativeInfo() { - return this; + return _stream; } IceInternal::SocketOperation IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::Buffer& writeBuffer, bool&) { - if(_state == StateNeedConnect) + IceInternal::SocketOperation status = _stream->connect(readBuffer, writeBuffer); + if(status != IceInternal::SocketOperationNone) { - _state = StateConnectPending; - return IceInternal::SocketOperationConnect; + return status; } - else if(_state <= StateConnectPending) - { - IceInternal::doFinishConnect(_fd); - - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - - if(_proxy) - { - // - // Prepare the read & write buffers in advance. - // - _proxy->beginWriteConnectRequest(_addr, writeBuffer); - _proxy->beginReadConnectRequestResponse(readBuffer); - - // - // Write the proxy connection message using TCP. - // - if(writeRaw(writeBuffer)) - { - // - // Write completed without blocking. - // - _proxy->endWriteConnectRequest(writeBuffer); - - // - // Try to read the response using TCP. - // - if(readRaw(readBuffer)) - { - // - // Read completed without blocking - fall through. - // - _proxy->endReadConnectRequestResponse(readBuffer); - } - else - { - // - // Return SocketOperationRead to indicate we need to complete the read. - // - _state = StateProxyConnectRequestPending; // Wait for proxy response - return IceInternal::SocketOperationRead; - } - } - else - { - // - // Return SocketOperationWrite to indicate we need to complete the write. - // - _state = StateProxyConnectRequest; // Send proxy connect request - return IceInternal::SocketOperationWrite; - } - } - - _state = StateConnected; - } - else if(_state == StateProxyConnectRequest) - { - // - // Write completed. - // - _proxy->endWriteConnectRequest(writeBuffer); - _state = StateProxyConnectRequestPending; // Wait for proxy response - return IceInternal::SocketOperationRead; - } - else if(_state == StateProxyConnectRequestPending) - { - // - // Read completed. - // - _proxy->endReadConnectRequestResponse(readBuffer); - _state = StateConnected; - } - - assert(_state == StateConnected); if(!_ssl) { // // This static_cast is necessary due to 64bit windows. There SOCKET is a non-int type. // - BIO* bio = BIO_new_socket(static_cast<int>(_fd), 0); + BIO* bio = BIO_new_socket(static_cast<int>(_stream->fd()), 0); if(!bio) { SecurityException ex(__FILE__, __LINE__); @@ -205,8 +131,10 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B switch(SSL_get_error(_ssl, ret)) { case SSL_ERROR_NONE: + { assert(SSL_is_init_finished(_ssl)); break; + } case SSL_ERROR_ZERO_RETURN: { ConnectionLostException ex(__FILE__, __LINE__); @@ -266,7 +194,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B { IceInternal::Address remoteAddr; string desc = "<not available>"; - if(IceInternal::fdToRemoteAddress(_fd, remoteAddr)) + if(IceInternal::fdToRemoteAddress(_stream->fd(), remoteAddr)) { desc = IceInternal::addrToString(remoteAddr); } @@ -308,8 +236,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B throw ex; } } - _engine->verifyPeer(_fd, _host, getNativeConnectionInfo()); - _state = StateHandshakeComplete; + _engine->verifyPeer(_stream->fd(), _host, getNativeConnectionInfo()); if(_engine->securityTraceLevel() >= 1) { @@ -348,12 +275,6 @@ IceSSL::TransceiverI::closing(bool initiator, const Ice::LocalException&) void IceSSL::TransceiverI::close() { - if(_state == StateHandshakeComplete && _instance->traceLevel() >= 1) - { - Trace out(_instance->logger(), _instance->traceCategory()); - out << "closing " << _instance->protocol() << " connection\n" << toString(); - } - if(_ssl) { int err = SSL_shutdown(_ssl); @@ -370,28 +291,15 @@ IceSSL::TransceiverI::close() _ssl = 0; } - assert(_fd != INVALID_SOCKET); - try - { - IceInternal::closeSocket(_fd); - _fd = INVALID_SOCKET; - } - catch(const SocketException&) - { - _fd = INVALID_SOCKET; - throw; - } + _stream->close(); } IceInternal::SocketOperation IceSSL::TransceiverI::write(IceInternal::Buffer& buf) { - if(_state == StateProxyConnectRequest) + if(!_stream->isConnected()) { - // - // We need to write the proxy message, but we have to use TCP and not SSL. - // - return writeRaw(buf) ? IceInternal::SocketOperationNone : IceInternal::SocketOperationWrite; + return _stream->write(buf); } if(buf.i == buf.b.end()) @@ -406,8 +314,6 @@ IceSSL::TransceiverI::write(IceInternal::Buffer& buf) while(buf.i != buf.b.end()) { ERR_clear_error(); // Clear any spurious errors. - assert(_fd != INVALID_SOCKET); - int ret = SSL_write(_ssl, reinterpret_cast<const void*>(&*buf.i), packetSize); if(ret <= 0) { @@ -479,12 +385,6 @@ IceSSL::TransceiverI::write(IceInternal::Buffer& buf) } } - if(_instance->traceLevel() >= 3) - { - Trace out(_instance->logger(), _instance->traceCategory()); - out << "sent " << ret << " of " << packetSize << " bytes via " << protocol() << "\n" << toString(); - } - buf.i += ret; if(packetSize > buf.b.end() - buf.i) @@ -499,6 +399,11 @@ IceSSL::TransceiverI::write(IceInternal::Buffer& buf) IceInternal::SocketOperation IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) { + if(!_stream->isConnected()) + { + return _stream->read(buf); + } + // // Note: we don't set the hasMoreData flag in this implementation. // We assume that OpenSSL doesn't read more SSL records than @@ -506,14 +411,6 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) // Ice messages in individual SSL records. // - if(_state == StateProxyConnectRequestPending) - { - // - // We need to read the proxy reply, but we have to use TCP and not SSL. - // - return readRaw(buf) ? IceInternal::SocketOperationNone : IceInternal::SocketOperationRead; - } - if(buf.i == buf.b.end()) { return IceInternal::SocketOperationNone; @@ -526,7 +423,6 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) while(buf.i != buf.b.end()) { ERR_clear_error(); // Clear any spurious errors. - assert(_fd != INVALID_SOCKET); int ret = SSL_read(_ssl, reinterpret_cast<void*>(&*buf.i), packetSize); if(ret <= 0) { @@ -539,17 +435,6 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) } case SSL_ERROR_ZERO_RETURN: { - // - // If the connection is lost when reading data, we shut - // down the write end of the socket. This helps to unblock - // threads that are stuck in send() or select() while - // sending data. Note: I don't really understand why - // send() or select() sometimes don't detect a connection - // loss. Therefore this helper to make them detect it. - // - //assert(_fd != INVALID_SOCKET); - //shutdownSocketReadWrite(_fd); - ConnectionLostException ex(__FILE__, __LINE__); ex.error = 0; throw ex; @@ -586,14 +471,6 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) if(IceInternal::connectionLost()) { - // - // See the commment above about shutting down the - // socket if the connection is lost while reading - // data. - // - //assert(_fd != INVALID_SOCKET); - //shutdownSocketReadWrite(_fd); - ConnectionLostException ex(__FILE__, __LINE__); ex.error = IceInternal::getSocketErrno(); throw ex; @@ -644,12 +521,6 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) } } - if(_instance->traceLevel() >= 3) - { - Trace out(_instance->logger(), _instance->traceCategory()); - out << "received " << ret << " of " << packetSize << " bytes via " << protocol() << "\n" << toString(); - } - buf.i += ret; if(packetSize > buf.b.end() - buf.i) @@ -670,7 +541,7 @@ IceSSL::TransceiverI::protocol() const string IceSSL::TransceiverI::toString() const { - return _desc; + return _stream->toString(); } string @@ -694,64 +565,28 @@ IceSSL::TransceiverI::checkSendSize(const IceInternal::Buffer& buf, size_t messa } } -IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, SOCKET fd, const IceInternal::NetworkProxyPtr& proxy, - const string& host, const IceInternal::Address& addr, - const IceInternal::Address& sourceAddr) : - IceInternal::NativeInfo(fd), +IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, const IceInternal::StreamSocketPtr& stream, + const string& host, const std::string& adapterName) : _instance(instance), _engine(OpenSSLEnginePtr::dynamicCast(instance->engine())), - _proxy(proxy), _host(host), - _addr(addr), - _sourceAddr(sourceAddr), - _incoming(false), - _ssl(0), - _state(StateNeedConnect) -{ - IceInternal::setBlock(fd, false); - IceInternal::setTcpBufSize(fd, _instance->properties(), _instance->logger()); - - IceInternal::Address connectAddr = proxy ? proxy->getAddress() : addr; - if(IceInternal::doConnect(_fd, connectAddr, _sourceAddr)) - { - _state = StateConnected; - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - if(_instance->traceLevel() >= 1) - { - Trace out(_instance->logger(), _instance->traceCategory()); - out << _instance->protocol() << " connection established\n" << _desc; - } - } - else - { - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - } -} - -IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, SOCKET fd, const string& adapterName) : - IceInternal::NativeInfo(fd), - _instance(instance), - _engine(OpenSSLEnginePtr::dynamicCast(instance->engine())), _adapterName(adapterName), - _incoming(true), - _ssl(0), - _state(StateConnected), - _desc(IceInternal::fdToString(fd)) + _incoming(_host.empty()), + _stream(stream), + _ssl(0) { - IceInternal::setBlock(fd, false); - IceInternal::setTcpBufSize(fd, _instance->properties(), _instance->logger()); } IceSSL::TransceiverI::~TransceiverI() { - assert(_fd == INVALID_SOCKET); } NativeConnectionInfoPtr IceSSL::TransceiverI::getNativeConnectionInfo() const { NativeConnectionInfoPtr info = new NativeConnectionInfo(); - IceInternal::fdToAddressAndPort(_fd, info->localAddress, info->localPort, info->remoteAddress, info->remotePort); + IceInternal::fdToAddressAndPort(_stream->fd(), info->localAddress, info->localPort, info->remoteAddress, + info->remotePort); if(_ssl != 0) { @@ -797,145 +632,4 @@ IceSSL::TransceiverI::getNativeConnectionInfo() const return info; } -bool -IceSSL::TransceiverI::writeRaw(IceInternal::Buffer& buf) -{ - // - // It's impossible for packetSize to be more than an Int. - // - int packetSize = static_cast<int>(buf.b.end() - buf.i); -#ifdef ICE_USE_IOCP - // - // Limit packet size to avoid performance problems on WIN32 - // - if(_maxSendPacketSize > 0 && packetSize > _maxSendPacketSize) - { - packetSize = _maxSendPacketSize; - } -#endif - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - - ssize_t ret = ::send(_fd, reinterpret_cast<const char*>(&*buf.i), packetSize, 0); - if(ret == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - return false; - } - - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - } - - if(_instance->traceLevel() >= 3) - { - Trace out(_instance->logger(), _instance->traceCategory()); - out << "sent " << ret << " of " << packetSize << " bytes via " << protocol() << "\n" << toString(); - } - - buf.i += ret; - - if(packetSize > buf.b.end() - buf.i) - { - packetSize = static_cast<int>(buf.b.end() - buf.i); - } - } - - return true; -} - -bool -IceSSL::TransceiverI::readRaw(IceInternal::Buffer& buf) -{ - // - // It's impossible for packetSize to be more than an Int. - // - int packetSize = static_cast<int>(buf.b.end() - buf.i); - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - ssize_t ret = ::recv(_fd, reinterpret_cast<char*>(&*buf.i), packetSize, 0); - - if(ret == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - return false; - } - - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - } - - if(_instance->traceLevel() >= 3) - { - Trace out(_instance->logger(), _instance->traceCategory()); - out << "received " << ret << " of " << packetSize << " bytes via " << protocol() << "\n" << toString(); - } - - buf.i += ret; - - packetSize = static_cast<int>(buf.b.end() - buf.i); - } - - return true; -} #endif diff --git a/cpp/src/IceSSL/OpenSSLTransceiverI.h b/cpp/src/IceSSL/OpenSSLTransceiverI.h index 3925944c590..b9e5fb6761d 100644 --- a/cpp/src/IceSSL/OpenSSLTransceiverI.h +++ b/cpp/src/IceSSL/OpenSSLTransceiverI.h @@ -17,6 +17,7 @@ #include <Ice/Transceiver.h> #include <Ice/Network.h> +#include <Ice/StreamSocket.h> #ifdef ICE_USE_OPENSSL @@ -29,36 +30,17 @@ namespace IceSSL class ConnectorI; class AcceptorI; -class TransceiverI : public IceInternal::Transceiver, public IceInternal::NativeInfo +class TransceiverI : public IceInternal::Transceiver { - enum State - { - StateNeedConnect, - StateConnectPending, - StateProxyConnectRequest, - StateProxyConnectRequestPending, - StateConnected, - StateHandshakeComplete - }; - public: virtual IceInternal::NativeInfoPtr getNativeInfo(); -#ifdef ICE_USE_IOCP - virtual IceInternal::AsyncInfo* getAsyncInfo(IceInternal::SocketOperation); -#endif virtual IceInternal::SocketOperation initialize(IceInternal::Buffer&, IceInternal::Buffer&, bool&); virtual IceInternal::SocketOperation closing(bool, const Ice::LocalException&); virtual void close(); virtual IceInternal::SocketOperation write(IceInternal::Buffer&); virtual IceInternal::SocketOperation read(IceInternal::Buffer&, bool&); -#ifdef ICE_USE_IOCP - virtual bool startWrite(IceInternal::Buffer&); - virtual void finishWrite(IceInternal::Buffer&); - virtual void startRead(IceInternal::Buffer&); - virtual void finishRead(IceInternal::Buffer&); -#endif virtual std::string protocol() const; virtual std::string toString() const; virtual std::string toDetailedString() const; @@ -67,54 +49,22 @@ public: private: - TransceiverI(const InstancePtr&, SOCKET, const IceInternal::NetworkProxyPtr&, const std::string&, - const IceInternal::Address&, const IceInternal::Address&); - TransceiverI(const InstancePtr&, SOCKET, const std::string&); + TransceiverI(const InstancePtr&, const IceInternal::StreamSocketPtr&, const std::string&, const std::string&); virtual ~TransceiverI(); virtual NativeConnectionInfoPtr getNativeConnectionInfo() const; -#ifdef ICE_USE_IOCP - bool receive(); - bool send(); - int writeAsync(char*, int); - int readAsync(char*, int); -#endif - - bool writeRaw(IceInternal::Buffer&); - bool readRaw(IceInternal::Buffer&); - friend class ConnectorI; friend class AcceptorI; const InstancePtr _instance; const OpenSSLEnginePtr _engine; - - const IceInternal::NetworkProxyPtr _proxy; const std::string _host; - const IceInternal::Address _addr; - const IceInternal::Address _sourceAddr; - const std::string _adapterName; const bool _incoming; + const IceInternal::StreamSocketPtr _stream; SSL* _ssl; - - State _state; - std::string _desc; -#ifdef ICE_USE_IOCP - int _maxSendPacketSize; - int _maxReceivePacketSize; - BIO* _iocpBio; - IceInternal::AsyncInfo _read; - IceInternal::AsyncInfo _write; - std::vector<char> _writeBuffer; - std::vector<char>::iterator _writeI; - std::vector<char> _readBuffer; - std::vector<char>::iterator _readI; - int _sentBytes; - int _sentPacketSize; -#endif }; typedef IceUtil::Handle<TransceiverI> TransceiverIPtr; diff --git a/cpp/src/IceSSL/PluginI.cpp b/cpp/src/IceSSL/PluginI.cpp index 9f92e227729..e55f50275b9 100644 --- a/cpp/src/IceSSL/PluginI.cpp +++ b/cpp/src/IceSSL/PluginI.cpp @@ -11,13 +11,11 @@ #include <IceSSL/Instance.h> #include <IceSSL/SSLEngine.h> #include <IceSSL/EndpointI.h> -#include <IceSSL/EndpointInfo.h> #include <Ice/WSEndpoint.h> #include <Ice/ProtocolPluginFacade.h> #include <Ice/ProtocolInstance.h> #include <Ice/LocalException.h> -#include <Ice/Object.h> using namespace std; using namespace Ice; @@ -53,15 +51,15 @@ IceSSL::PluginI::PluginI(const Ice::CommunicatorPtr& communicator) IceInternal::ProtocolPluginFacadePtr facade = IceInternal::getProtocolPluginFacade(communicator); // - // Register the endpoint factory. We have to do this now, rather than - // in initialize, because the communicator may need to interpret - // proxies before the plug-in is fully initialized. + // Register the endpoint factory. We have to do this now, rather + // than in initialize, because the communicator may need to + // interpret proxies before the plug-in is fully initialized. // IceInternal::EndpointFactoryPtr sslFactory = new EndpointFactoryI(new Instance(_engine, EndpointType, "ssl")); facade->addEndpointFactory(sslFactory); - IceInternal::ProtocolInstancePtr wssInstance = new IceInternal::ProtocolInstance(communicator, WSSEndpointType, "wss"); - facade->addEndpointFactory(new IceInternal::WSEndpointFactory(wssInstance, sslFactory->clone(wssInstance))); + IceInternal::ProtocolInstancePtr wss = new IceInternal::ProtocolInstance(communicator, WSSEndpointType, "wss"); + facade->addEndpointFactory(new IceInternal::WSEndpointFactory(wss, sslFactory->clone(wss))); } void diff --git a/cpp/src/IceSSL/SChannelTransceiverI.cpp b/cpp/src/IceSSL/SChannelTransceiverI.cpp index 8b3867d571e..108a130af12 100644 --- a/cpp/src/IceSSL/SChannelTransceiverI.cpp +++ b/cpp/src/IceSSL/SChannelTransceiverI.cpp @@ -208,26 +208,9 @@ getSecBufferWithType(const SecBufferDesc& desc, ULONG bufferType) IceInternal::NativeInfoPtr IceSSL::TransceiverI::getNativeInfo() { - return this; + return _stream; } -#ifdef ICE_USE_IOCP -IceInternal::AsyncInfo* -IceSSL::TransceiverI::getAsyncInfo(IceInternal::SocketOperation status) -{ - switch(status) - { - case IceInternal::SocketOperationRead: - return &_read; - case IceInternal::SocketOperationWrite: - return &_write; - default: - assert(false); - return 0; - } -} -#endif - IceInternal::SocketOperation IceSSL::TransceiverI::sslHandshake() { @@ -248,15 +231,18 @@ IceSSL::TransceiverI::sslHandshake() SECURITY_STATUS err = SEC_E_OK; DWORD ctxFlags = 0; - - while(true) + if(_state == StateHandshakeNotStarted) { - if(_state == StateConnected) + _readBuffer.b.resize(2048); + _readBuffer.i = _readBuffer.b.begin(); + _credentials = _engine->newCredentialsHandle(_incoming); + _credentialsInitialized = true; + + if(!_incoming) { - assert(!_incoming); SecBuffer outBuffer = { 0, SECBUFFER_TOKEN, 0 }; SecBufferDesc outBufferDesc = { SECBUFFER_VERSION, 1, &outBuffer }; - + err = InitializeSecurityContext(&_credentials, 0, const_cast<char *>(_host.c_str()), flags, 0, 0, 0, 0, &_ssl, &outBufferDesc, &ctxFlags, 0); _sslInitialized = true; @@ -265,7 +251,7 @@ IceSSL::TransceiverI::sslHandshake() throw SecurityException(__FILE__, __LINE__, "IceSSL: handshake failure:\n" + IceUtilInternal::lastErrorToString()); } - + // // Copy the data to the write buffer // @@ -273,10 +259,17 @@ IceSSL::TransceiverI::sslHandshake() _writeBuffer.i = _writeBuffer.b.begin(); memcpy(_writeBuffer.i, outBuffer.pvBuffer, outBuffer.cbBuffer); FreeContextBuffer(outBuffer.pvBuffer); - + _state = StateHandshakeWriteContinue; } + else + { + _state = StateHandshakeReadContinue; + } + } + while(true) + { if(_state == StateHandshakeReadContinue) { // If read buffer is empty, try to read some data. @@ -637,64 +630,13 @@ IceSSL::TransceiverI::encryptMessage(IceInternal::Buffer& buffer) IceInternal::SocketOperation IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::Buffer& writeBuffer, bool& hasMoreData) { - if(_state == StateNeedConnect) - { - _state = StateConnectPending; - return IceInternal::SocketOperationConnect; - } - else if(_state <= StateConnectPending) - { - IceInternal::doFinishConnectAsync(_fd, _write); - - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - - if(_proxy) - { - // - // Prepare the read & write buffers in advance. - // - _proxy->beginWriteConnectRequest(_addr, writeBuffer); - _proxy->beginReadConnectRequestResponse(readBuffer); - - // - // Return SocketOperationWrite to indicate we need to start a write. - // - _state = StateProxyConnectRequest; // Send proxy connect request - return IceInternal::SocketOperationWrite; - } - - _state = StateConnected; - } - else if(_state == StateProxyConnectRequest) - { - // - // Write completed. - // - _proxy->endWriteConnectRequest(writeBuffer); - _state = StateProxyConnectRequestPending; // Wait for proxy response - return IceInternal::SocketOperationRead; - } - else if(_state == StateProxyConnectRequestPending) - { - // - // Read completed. - // - _proxy->endReadConnectRequestResponse(readBuffer); - _state = StateConnected; - } - - assert(_state >= StateConnected && _state <= StateHandshakeWriteContinue); - - if(!_credentialsInitialized) + IceInternal::SocketOperation op = _stream->connect(readBuffer, writeBuffer); + if(op != IceInternal::SocketOperationNone) { - _readBuffer.b.resize(2048); - _readBuffer.i = _readBuffer.b.begin(); - - _credentials = _engine->newCredentialsHandle(_incoming); - _credentialsInitialized = true; + return op; } - - IceInternal::SocketOperation op = sslHandshake(); + + op = sslHandshake(); if(op != IceInternal::SocketOperationNone) { return op; @@ -776,7 +718,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B } } } - _engine->verifyPeer(_fd, _host, getNativeConnectionInfo()); + _engine->verifyPeer(_stream->fd(), _host, getNativeConnectionInfo()); _state = StateHandshakeComplete; if(_instance->engine()->securityTraceLevel() >= 1) @@ -805,7 +747,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B << "\nkey exchange = " << sslKeyExchangeAlgorithm << "\nprotocol = " << sslProtocolName << "\n"; } - out << IceInternal::fdToString(_fd); + out << toString(); } hasMoreData = !_readUnprocessed.b.empty() || _readBuffer.i != _readBuffer.b.begin(); return IceInternal::SocketOperationNone; @@ -832,22 +774,17 @@ IceSSL::TransceiverI::close() FreeCredentialsHandle(&_credentials); } - assert(_fd != INVALID_SOCKET); - try - { - IceInternal::closeSocket(_fd); - _fd = INVALID_SOCKET; - } - catch(const SocketException&) - { - _fd = INVALID_SOCKET; - throw; - } + _stream->close(); } IceInternal::SocketOperation IceSSL::TransceiverI::write(IceInternal::Buffer& buf) { + if(!_stream->isConnected()) + { + return _stream->write(buf); + } + if(buf.i == buf.b.end()) { return IceInternal::SocketOperationNone; @@ -878,6 +815,11 @@ IceSSL::TransceiverI::write(IceInternal::Buffer& buf) IceInternal::SocketOperation IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool& hasMoreData) { + if(!_stream->isConnected()) + { + return _stream->read(buf); + } + if(buf.i == buf.b.end()) { return IceInternal::SocketOperationNone; @@ -913,173 +855,74 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool& hasMoreData) bool IceSSL::TransceiverI::startWrite(IceInternal::Buffer& buffer) { - if(_state == StateConnectPending) + if(!_stream->isConnected()) { - IceInternal::Address addr = _proxy ? _proxy->getAddress() : _addr; - doConnectAsync(_fd, addr, _sourceAddr, _write); - return false; + return _stream->startWrite(buffer); } - IceInternal::Buffer& buf = _state == StateProxyConnectRequest ? buffer : _writeBuffer; - if(_state == StateHandshakeComplete && _bufferedW == 0) { assert(_writeBuffer.i == _writeBuffer.b.end()); - _bufferedW = encryptMessage(buf); + _bufferedW = encryptMessage(buffer); } - assert(buf.i != buf.b.end()); - int packetSize = static_cast<int>(buf.b.end() - buf.i); - if(_maxSendPacketSize > 0 && packetSize > _maxSendPacketSize) - { - packetSize = _maxSendPacketSize; - } - assert(packetSize > 0); - _write.buf.len = static_cast<DWORD>(packetSize); - _write.buf.buf = reinterpret_cast<char*>(buf.i); - int err = WSASend(_fd, &_write.buf, 1, &_write.count, 0, &_write, NULL); - if(err == SOCKET_ERROR) - { - if(!IceInternal::wouldBlock()) - { - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - } - } - return packetSize == static_cast<int>(buf.b.end() - buf.i); + return _stream->startWrite(_writeBuffer); } void IceSSL::TransceiverI::finishWrite(IceInternal::Buffer& buf) { - if(_state < StateConnected && _state != StateProxyConnectRequest) + if(!_stream->isConnected()) { + _stream->finishWrite(buf); return; } - if(static_cast<int>(_write.count) == SOCKET_ERROR) + _stream->finishWrite(_writeBuffer); + if(_writeBuffer.i != _writeBuffer.b.end()) { - WSASetLastError(_write.error); - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } + return; // We're not finished yet with writing the write buffer. } - if(_state == StateProxyConnectRequest) + if(_state == StateHandshakeComplete) { - buf.i += _write.count; - } - else - { - _writeBuffer.i += _write.count; - if(_writeBuffer.i == _writeBuffer.b.end()) - { - buf.i += _bufferedW; - _bufferedW = 0; - } + buf.i += _bufferedW; + _bufferedW = 0; } } void IceSSL::TransceiverI::startRead(IceInternal::Buffer& buffer) { - IceInternal::Buffer& buf = _state == StateProxyConnectRequest ? buffer : _readBuffer; - - int packetSize = static_cast<int>(buf.b.end() - buf.i); - if(_maxReceivePacketSize > 0 && packetSize > _maxReceivePacketSize) - { - packetSize = _maxReceivePacketSize; - } - assert(!buf.b.empty() && buf.i != buf.b.end()); - - _read.buf.len = static_cast<DWORD>(packetSize); - _read.buf.buf = reinterpret_cast<char*>(buf.i); - - int err = WSARecv(_fd, &_read.buf, 1, &_read.count, &_read.flags, &_read, NULL); - if(err == SOCKET_ERROR) + if(!_stream->isConnected()) { - if(!IceInternal::wouldBlock()) - { - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - } + _stream->startRead(buffer); + return; } + _stream->startRead(_readBuffer); } void IceSSL::TransceiverI::finishRead(IceInternal::Buffer& buf, bool& hasMoreData) { - if(static_cast<int>(_read.count) == SOCKET_ERROR) - { - WSASetLastError(_read.error); - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - } - else if(_read.count == 0) + if(!_stream->isConnected()) { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; + _stream->finishRead(buf); + return; } - if(_state == StateProxyConnectRequest) + _stream->finishRead(_readBuffer); + if(_state == StateHandshakeComplete) { - buf.i += _read.count; - } - else - { - _readBuffer.i += _read.count; - if(_state == StateHandshakeComplete) + size_t decrypted = decryptMessage(buf); + if(decrypted > 0) { - size_t decrypted = decryptMessage(buf); - if(decrypted > 0) - { - buf.i += decrypted; - hasMoreData = !_readUnprocessed.b.empty() || _readBuffer.i != _readBuffer.b.begin(); - } - else - { - hasMoreData = false; - } + buf.i += decrypted; + hasMoreData = !_readUnprocessed.b.empty() || _readBuffer.i != _readBuffer.b.begin(); + } + else + { + hasMoreData = false; } } } @@ -1094,7 +937,7 @@ IceSSL::TransceiverI::protocol() const string IceSSL::TransceiverI::toString() const { - return _desc; + return _stream->toString(); } string @@ -1118,113 +961,36 @@ IceSSL::TransceiverI::checkSendSize(const IceInternal::Buffer& buf, size_t messa } } -IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, SOCKET fd, const IceInternal::NetworkProxyPtr& proxy, - const string& host, const IceInternal::Address& addr, - const IceInternal::Address& sourceAddr) : - IceInternal::NativeInfo(fd), +IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, + const IceInternal::StreamSocketPtr& stream, + const string& host, + const string& adapterName) : _instance(instance), _engine(SChannelEnginePtr::dynamicCast(instance->engine())), - _proxy(proxy), _host(host), - _addr(addr), - _sourceAddr(sourceAddr), - _incoming(false), - _state(StateNeedConnect), - _writeBuffer(0), - _bufferedW(0), - _readBuffer(0), - _readUnprocessed(0), - _sslInitialized(false), - _credentialsInitialized(false) -#ifdef ICE_USE_IOCP - , _read(IceInternal::SocketOperationRead), - _write(IceInternal::SocketOperationWrite) -#endif -{ - IceInternal::setBlock(fd, false); - IceInternal::setTcpBufSize(fd, _instance->properties(), _instance->logger()); - - // - // On Windows, limiting the buffer size is important to prevent - // poor throughput performances when transfering large amount of - // data. See Microsoft KB article KB823764. - // - _maxSendPacketSize = IceInternal::getSendBufferSize(_fd) / 2; - if(_maxSendPacketSize < 512) - { - _maxSendPacketSize = 0; - } - - _maxReceivePacketSize = IceInternal::getRecvBufferSize(_fd); - if(_maxReceivePacketSize < 512) - { - _maxReceivePacketSize = 0; - } - -#ifndef ICE_USE_IOCP - IceInternal::Address connectAddr = proxy ? proxy->getAddress() : addr; - if(IceInternal::doConnect(_fd, connectAddr, _sourceAddr)) - { - _state = StateConnected; - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - } - else - { - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - } -#endif -} - -IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, SOCKET fd, const string& adapterName) : - IceInternal::NativeInfo(fd), - _instance(instance), - _engine(SChannelEnginePtr::dynamicCast(instance->engine())), _adapterName(adapterName), - _incoming(true), - _state(StateHandshakeReadContinue), - _desc(IceInternal::fdToString(fd)), + _incoming(host.empty()), + _stream(stream), + _state(StateHandshakeNotStarted), _writeBuffer(0), _bufferedW(0), _readBuffer(0), _readUnprocessed(0), _sslInitialized(false), _credentialsInitialized(false) -#ifdef ICE_USE_IOCP - , _read(IceInternal::SocketOperationRead), - _write(IceInternal::SocketOperationWrite) -#endif { - IceInternal::setBlock(fd, false); - IceInternal::setTcpBufSize(fd, _instance->properties(), _instance->logger()); - - // - // On Windows, limiting the buffer size is important to prevent - // poor throughput performances when transfering large amount of - // data. See Microsoft KB article KB823764. - // - _maxSendPacketSize = IceInternal::getSendBufferSize(_fd) / 2; - if(_maxSendPacketSize < 512) - { - _maxSendPacketSize = 0; - } - - _maxReceivePacketSize = IceInternal::getRecvBufferSize(_fd); - if(_maxReceivePacketSize < 512) - { - _maxReceivePacketSize = 0; - } } IceSSL::TransceiverI::~TransceiverI() { - assert(_fd == INVALID_SOCKET); } NativeConnectionInfoPtr IceSSL::TransceiverI::getNativeConnectionInfo() const { NativeConnectionInfoPtr info = new NativeConnectionInfo(); - IceInternal::fdToAddressAndPort(_fd, info->localAddress, info->localPort, info->remoteAddress, info->remotePort); + IceInternal::fdToAddressAndPort(_stream->fd(), info->localAddress, info->localPort, info->remoteAddress, + info->remotePort); if(_sslInitialized) { @@ -1282,121 +1048,16 @@ IceSSL::TransceiverI::getNativeConnectionInfo() const bool IceSSL::TransceiverI::writeRaw(IceInternal::Buffer& buf) { - int packetSize = static_cast<int>(buf.b.end() - buf.i); -#ifdef ICE_USE_IOCP - // - // Limit packet size to avoid performance problems on WIN32 - // - if(_maxSendPacketSize > 0 && packetSize > _maxSendPacketSize) - { - packetSize = _maxSendPacketSize; - } -#endif - - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - - int ret = ::send(_fd, reinterpret_cast<const char*>(buf.i), packetSize, 0); - if(ret == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - return false; - } - - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - } - buf.i += ret; - if(packetSize > static_cast<int>(buf.b.end() - buf.i)) - { - packetSize = static_cast<int>(buf.b.end() - buf.i); - } - } - return true; + ssize_t ret = _stream->write(reinterpret_cast<const char*>(&*buf.i), buf.b.end() - buf.i); + buf.i += ret; + return buf.i == buf.b.end(); } bool IceSSL::TransceiverI::readRaw(IceInternal::Buffer& buf) { - assert(buf.i != buf.b.end()); - int packetSize = static_cast<int>(buf.b.end() - buf.i); - Byte* i = buf.i; - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - ssize_t ret = ::recv(_fd, reinterpret_cast<char*>(buf.i), packetSize, 0); - - if(ret == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - return buf.i != i; - } - - if(IceInternal::connectionLost()) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - else - { - SocketException ex(__FILE__, __LINE__); - ex.error = IceInternal::getSocketErrno(); - throw ex; - } - } - buf.i += ret; - packetSize = static_cast<int>(buf.b.end() - buf.i); - } - return buf.i != i; + ssize_t ret = _stream->read(reinterpret_cast<char*>(&*buf.i), buf.b.end() - buf.i); + buf.i += ret; + return ret > 0; } #endif diff --git a/cpp/src/IceSSL/SChannelTransceiverI.h b/cpp/src/IceSSL/SChannelTransceiverI.h index 53f5d2765ae..be9b1f89af0 100644 --- a/cpp/src/IceSSL/SChannelTransceiverI.h +++ b/cpp/src/IceSSL/SChannelTransceiverI.h @@ -18,6 +18,7 @@ #include <Ice/Transceiver.h> #include <Ice/Network.h> #include <Ice/Buffer.h> +#include <Ice/StreamSocket.h> #ifdef ICE_USE_SCHANNEL @@ -41,34 +42,23 @@ namespace IceSSL class ConnectorI; class AcceptorI; -class TransceiverI : public IceInternal::Transceiver, public IceInternal::NativeInfo +class TransceiverI : public IceInternal::Transceiver { - enum State - { - StateNeedConnect, - StateConnectPending, - StateProxyConnectRequest, - StateProxyConnectRequestPending, - StateConnected, - StateHandshakeReadContinue, - StateHandshakeWriteContinue, - StateHandshakeComplete - }; - public: virtual IceInternal::NativeInfoPtr getNativeInfo(); - virtual IceInternal::AsyncInfo* getAsyncInfo(IceInternal::SocketOperation); virtual IceInternal::SocketOperation initialize(IceInternal::Buffer&, IceInternal::Buffer&, bool&); virtual IceInternal::SocketOperation closing(bool, const Ice::LocalException&); virtual void close(); virtual IceInternal::SocketOperation write(IceInternal::Buffer&); virtual IceInternal::SocketOperation read(IceInternal::Buffer&, bool&); +#ifdef ICE_USE_IOCP virtual bool startWrite(IceInternal::Buffer&); virtual void finishWrite(IceInternal::Buffer&); virtual void startRead(IceInternal::Buffer&); virtual void finishRead(IceInternal::Buffer&, bool&); +#endif virtual std::string protocol() const; virtual std::string toString() const; virtual std::string toDetailedString() const; @@ -77,9 +67,7 @@ public: private: - TransceiverI(const InstancePtr&, SOCKET, const IceInternal::NetworkProxyPtr&, const std::string&, - const IceInternal::Address&, const IceInternal::Address&); - TransceiverI(const InstancePtr&, SOCKET, const std::string&); + TransceiverI(const InstancePtr&, const IceInternal::StreamSocketPtr&, const std::string&, const std::string&); virtual ~TransceiverI(); virtual NativeConnectionInfoPtr getNativeConnectionInfo() const; @@ -95,19 +83,21 @@ private: friend class ConnectorI; friend class AcceptorI; + enum State + { + StateHandshakeNotStarted, + StateHandshakeReadContinue, + StateHandshakeWriteContinue, + StateHandshakeComplete + }; + const InstancePtr _instance; const SChannelEnginePtr _engine; - - const IceInternal::NetworkProxyPtr _proxy; const std::string _host; - const IceInternal::Address _addr; - const IceInternal::Address _sourceAddr; - const std::string _adapterName; const bool _incoming; - + const IceInternal::StreamSocketPtr _stream; State _state; - std::string _desc; // // Buffered encrypted data that has not been written. @@ -129,13 +119,6 @@ private: bool _sslInitialized; CredHandle _credentials; bool _credentialsInitialized; - -#ifdef ICE_USE_IOCP - IceInternal::AsyncInfo _read; - IceInternal::AsyncInfo _write; - int _maxSendPacketSize; - int _maxReceivePacketSize; -#endif SecPkgContext_StreamSizes _sizes; }; typedef IceUtil::Handle<TransceiverI> TransceiverIPtr; diff --git a/cpp/src/IceSSL/SecureTransportTransceiverI.cpp b/cpp/src/IceSSL/SecureTransportTransceiverI.cpp index 9a8534e70e5..a9058f5e9fb 100644 --- a/cpp/src/IceSSL/SecureTransportTransceiverI.cpp +++ b/cpp/src/IceSSL/SecureTransportTransceiverI.cpp @@ -170,91 +170,18 @@ checkTrustResult(SecTrustRef trust, const SecureTransportEnginePtr& engine, cons IceInternal::NativeInfoPtr IceSSL::TransceiverI::getNativeInfo() { - return this; + return _stream; } IceInternal::SocketOperation IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::Buffer& writeBuffer, bool&) { - if(_state == StateNeedConnect) + IceInternal::SocketOperation status = _stream->connect(readBuffer, writeBuffer); + if(status != IceInternal::SocketOperationNone) { - _state = StateConnectPending; - return IceInternal::SocketOperationConnect; - } - else if(_state <= StateConnectPending) - { - IceInternal::doFinishConnect(_fd); - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - - if(_proxy) - { - // - // Prepare the read & write buffers in advance. - // - _proxy->beginWriteConnectRequest(_addr, writeBuffer); - _proxy->beginReadConnectRequestResponse(readBuffer); - - // - // Write the proxy connection message using TCP. - // - if(writeRaw(writeBuffer)) - { - // - // Write completed without blocking. - // - _proxy->endWriteConnectRequest(writeBuffer); - - // - // Try to read the response using TCP. - // - if(readRaw(readBuffer)) - { - // - // Read completed without blocking - fall through. - // - _proxy->endReadConnectRequestResponse(readBuffer); - } - else - { - // - // Return SocketOperationRead to indicate we need to complete the read. - // - _state = StateProxyConnectRequestPending; // Wait for proxy response - return IceInternal::SocketOperationRead; - } - } - else - { - // - // Return SocketOperationWrite to indicate we need to complete the write. - // - _state = StateProxyConnectRequest; // Send proxy connect request - return IceInternal::SocketOperationWrite; - } - } - - _state = StateConnected; - } - else if(_state == StateProxyConnectRequest) - { - // - // Write completed. - // - _proxy->endWriteConnectRequest(writeBuffer); - _state = StateProxyConnectRequestPending; // Wait for proxy response - return IceInternal::SocketOperationRead; - } - else if(_state == StateProxyConnectRequestPending) - { - // - // Read completed. - // - _proxy->endReadConnectRequestResponse(readBuffer); - _state = StateConnected; + return status; } - assert(_state == StateConnected); - OSStatus err = 0; if(!_ssl) { @@ -270,7 +197,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B if((err = SSLSetConnection(_ssl, reinterpret_cast<SSLConnectionRef>(this)))) { - throw SecurityException(__FILE__, __LINE__, "IceSSL: setting SSL connection failed\n" + + throw SecurityException(__FILE__, __LINE__, "IceSSL: setting SSL connection failed\n" + errorToString(err)); } } @@ -311,7 +238,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B IceInternal::Address remoteAddr; string desc = "<not available>"; - if(IceInternal::fdToRemoteAddress(_fd, remoteAddr)) + if(IceInternal::fdToRemoteAddress(_stream->fd(), remoteAddr)) { desc = IceInternal::addrToString(remoteAddr); } @@ -320,8 +247,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B << "remote address = " << desc << "\n" << errorToString(err); throw ProtocolException(__FILE__, __LINE__, os.str()); } - _engine->verifyPeer(_fd, _host, getNativeConnectionInfo()); - _state = StateHandshakeComplete; + _engine->verifyPeer(_stream->fd(), _host, getNativeConnectionInfo()); if(_instance->engine()->securityTraceLevel() >= 1) { @@ -346,7 +272,7 @@ IceSSL::TransceiverI::initialize(IceInternal::Buffer& readBuffer, IceInternal::B out << "cipher = " << sslCipherName << "\n"; out << "protocol = " << sslProtocolName << "\n"; } - out << IceInternal::fdToString(_fd); + out << toString(); } return IceInternal::SocketOperationNone; @@ -376,28 +302,15 @@ IceSSL::TransceiverI::close() _ssl = 0; } - assert(_fd != INVALID_SOCKET); - try - { - IceInternal::closeSocket(_fd); - _fd = INVALID_SOCKET; - } - catch(const SocketException&) - { - _fd = INVALID_SOCKET; - throw; - } + _stream->close(); } IceInternal::SocketOperation IceSSL::TransceiverI::write(IceInternal::Buffer& buf) { - if(_state == StateProxyConnectRequest) + if(!_stream->isConnected()) { - // - // We need to write the proxy message, but we have to use TCP and not SSL. - // - return writeRaw(buf) ? IceInternal::SocketOperationNone : IceInternal::SocketOperationWrite; + return _stream->write(buf); } if(buf.i == buf.b.end()) @@ -408,11 +321,9 @@ IceSSL::TransceiverI::write(IceInternal::Buffer& buf) // // It's impossible for packetSize to be more than an Int. // - size_t packetSize = buf.b.end() - buf.i; - packetSize = std::min(packetSize, _maxSendPacketSize); + size_t packetSize = std::min(static_cast<size_t>(buf.b.end() - buf.i), _maxSendPacketSize); while(buf.i != buf.b.end()) { - assert(_fd != INVALID_SOCKET); size_t processed = 0; OSStatus err = _buffered ? SSLWrite(_ssl, 0, 0, &processed) : SSLWrite(_ssl, reinterpret_cast<const void*>(buf.i), packetSize, &processed); @@ -440,7 +351,7 @@ IceSSL::TransceiverI::write(IceInternal::Buffer& buf) // if(err <= -9800 && err >= -9849) { - throw ProtocolException(__FILE__, __LINE__, "IceSSL: error during read:\n" + errorToString(err)); + throw ProtocolException(__FILE__, __LINE__, "IceSSL: error during write:\n" + errorToString(err)); } errno = err; @@ -476,6 +387,11 @@ IceSSL::TransceiverI::write(IceInternal::Buffer& buf) IceInternal::SocketOperation IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) { + if(!_stream->isConnected()) + { + return _stream->read(buf); + } + // // Note: we don't set the hasMoreData flag in this implementation. // We assume that SecureTransport doesn't read more SSL records @@ -483,28 +399,16 @@ IceSSL::TransceiverI::read(IceInternal::Buffer& buf, bool&) // sends Ice messages in individual SSL records. // - if(_state == StateProxyConnectRequestPending) - { - // - // We need to read the proxy reply, but we have to use TCP and not SSL. - // - return readRaw(buf) ? IceInternal::SocketOperationNone : IceInternal::SocketOperationRead; - } - if(buf.i == buf.b.end()) { return IceInternal::SocketOperationNone; } - size_t packetSize = buf.b.end() - buf.i; - packetSize = std::min(packetSize, _maxReceivePacketSize); - + size_t packetSize = std::min(static_cast<size_t>(buf.b.end() - buf.i), _maxRecvPacketSize); while(buf.i != buf.b.end()) { - assert(_fd != INVALID_SOCKET); size_t processed = 0; OSStatus err = SSLRead(_ssl, reinterpret_cast<void*>(buf.i), packetSize, &processed); - if(err) { if(err == errSSLWouldBlock) @@ -563,7 +467,7 @@ IceSSL::TransceiverI::protocol() const string IceSSL::TransceiverI::toString() const { - return _desc; + return _stream->toString(); } string @@ -587,79 +491,38 @@ IceSSL::TransceiverI::checkSendSize(const IceInternal::Buffer& buf, size_t messa } } -IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, SOCKET fd, const IceInternal::NetworkProxyPtr& proxy, - const string& host, const IceInternal::Address& addr, - const IceInternal::Address& sourceAddr) : - IceInternal::NativeInfo(fd), +IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, + const IceInternal::StreamSocketPtr& stream, + const string& host, + const string& adapterName) : _instance(instance), _engine(SecureTransportEnginePtr::dynamicCast(instance->engine())), - _proxy(proxy), _host(host), - _addr(addr), - _sourceAddr(sourceAddr), - _incoming(false), - _ssl(0), - _trust(0), - _buffered(0), - _state(StateNeedConnect) -{ - assert(_engine); - IceInternal::setBlock(fd, false); - IceInternal::setTcpBufSize(fd, _instance->properties(), _instance->logger()); - - IceInternal::Address connectAddr = proxy ? proxy->getAddress() : addr; - if(IceInternal::doConnect(_fd, connectAddr, _sourceAddr)) - { - _state = StateConnected; - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - } - else - { - _desc = IceInternal::fdToString(_fd, _proxy, _addr, true); - } - - // - // Limit the size of packets passed to SSLWrite/SSLRead to avoid - // blocking and holding too much memory. - // - _maxSendPacketSize = std::max(512, IceInternal::getSendBufferSize(fd)); - _maxReceivePacketSize = std::max(512, IceInternal::getRecvBufferSize(fd)); -} - -IceSSL::TransceiverI::TransceiverI(const InstancePtr& instance, SOCKET fd, const string& adapterName) : - IceInternal::NativeInfo(fd), - _instance(instance), - _engine(SecureTransportEnginePtr::dynamicCast(instance->engine())), _adapterName(adapterName), - _incoming(true), + _incoming(host.empty()), + _stream(stream), _ssl(0), _trust(0), - _buffered(0), - _state(StateConnected), - _desc(IceInternal::fdToString(fd)) + _buffered(0) { - assert(_engine); - IceInternal::setBlock(fd, false); - IceInternal::setTcpBufSize(fd, _instance->properties(), _instance->logger()); - // // Limit the size of packets passed to SSLWrite/SSLRead to avoid // blocking and holding too much memory. // - _maxSendPacketSize = std::max(512, IceInternal::getSendBufferSize(fd)); - _maxReceivePacketSize = std::max(512, IceInternal::getRecvBufferSize(fd)); + _maxSendPacketSize = std::max(512, IceInternal::getSendBufferSize(_stream->fd())); + _maxRecvPacketSize = std::max(512, IceInternal::getRecvBufferSize(_stream->fd())); } IceSSL::TransceiverI::~TransceiverI() { - assert(_fd == INVALID_SOCKET); } NativeConnectionInfoPtr IceSSL::TransceiverI::getNativeConnectionInfo() const { NativeConnectionInfoPtr info = new NativeConnectionInfo(); - IceInternal::fdToAddressAndPort(_fd, info->localAddress, info->localPort, info->remoteAddress, info->remotePort); + IceInternal::fdToAddressAndPort(_stream->fd(), info->localAddress, info->localPort, info->remoteAddress, + info->remotePort); if(_ssl) { @@ -683,162 +546,34 @@ IceSSL::TransceiverI::getNativeConnectionInfo() const return info; } -bool -IceSSL::TransceiverI::writeRaw(IceInternal::Buffer& buf) -{ - int packetSize = buf.b.end() - buf.i; - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - - ssize_t ret = ::send(_fd, reinterpret_cast<const char*>(&*buf.i), packetSize, 0); - if(ret == 0) - { - ConnectionLostException ex(__FILE__, __LINE__); - ex.error = 0; - throw ex; - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - return false; - } - - if(IceInternal::connectionLost()) - { - throw ConnectionLostException(__FILE__, __LINE__, IceInternal::getSocketErrno()); - } - else - { - throw SocketException(__FILE__, __LINE__, IceInternal::getSocketErrno()); - } - } - - buf.i += ret; - - if(packetSize > buf.b.end() - buf.i) - { - packetSize = buf.b.end() - buf.i; - } - } - - return true; -} - -bool -IceSSL::TransceiverI::readRaw(IceInternal::Buffer& buf) -{ - int packetSize = buf.b.end() - buf.i; - while(buf.i != buf.b.end()) - { - assert(_fd != INVALID_SOCKET); - ssize_t ret = ::recv(_fd, reinterpret_cast<char*>(&*buf.i), packetSize, 0); - - if(ret == 0) - { - throw ConnectionLostException(__FILE__, __LINE__, 0); - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - return false; - } - - if(IceInternal::connectionLost()) - { - throw ConnectionLostException(__FILE__, __LINE__, IceInternal::getSocketErrno()); - } - else - { - throw SocketException(__FILE__, __LINE__, IceInternal::getSocketErrno()); - } - } - - buf.i += ret; - if(packetSize > buf.b.end() - buf.i) - { - packetSize = buf.b.end() - buf.i; - } - } - - return true; -} - OSStatus IceSSL::TransceiverI::writeRaw(const char* data, size_t* length) const { _flags &= ~SSLWantWrite; - assert(_fd != INVALID_SOCKET); - - char* i = const_cast<char*>(data); - int packetSize = *length; - char* end = i + packetSize; - - while(i != end) + try { - ssize_t ret = ::send(_fd, const_cast<const char*>(i), packetSize, 0); - if(ret == 0) - { - return errSSLClosedGraceful; - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - *length = i - data; - _flags |= SSLWantWrite; - return errSSLWouldBlock; - } - return errno; - } - - i += ret; - - if(packetSize > end - i) + ssize_t ret = _stream->write(data, *length); + if(ret < *length) { - packetSize = end - i; + *length = static_cast<size_t>(ret); + _flags |= SSLWantWrite; + return errSSLWouldBlock; } } - *length = i - data; + catch(const Ice::ConnectionLostException&) + { + return errSSLClosedGraceful; + } + catch(const Ice::SocketException& ex) + { + return ex.error; + } + catch(...) + { + assert(false); + return IceInternal::getSocketErrno(); + } return noErr; } @@ -847,49 +582,29 @@ IceSSL::TransceiverI::readRaw(char* data, size_t* length) const { _flags &= ~SSLWantRead; - assert(_fd != INVALID_SOCKET); - - char* i = data; - int packetSize = *length; - char* end = i + packetSize; - while(i != end) + try { - ssize_t ret = ::recv(_fd, i, packetSize, 0); - if(ret == 0) - { - return errSSLClosedGraceful; - } - - if(ret == SOCKET_ERROR) - { - if(IceInternal::interrupted()) - { - continue; - } - - if(IceInternal::noBuffers() && packetSize > 1024) - { - packetSize /= 2; - continue; - } - - if(IceInternal::wouldBlock()) - { - *length = i - data; - _flags |= SSLWantRead; - return errSSLWouldBlock; - } - return errno; - } - - i += ret; - if(packetSize > end - i) + ssize_t ret = _stream->read(data, *length); + if(ret < *length) { - packetSize = end - i; + *length = static_cast<size_t>(ret); + _flags |= SSLWantRead; + return errSSLWouldBlock; } } - - *length = i - data; + catch(const Ice::ConnectionLostException&) + { + return errSSLClosedGraceful; + } + catch(const Ice::SocketException& ex) + { + return ex.error; + } + catch(...) + { + assert(false); + return IceInternal::getSocketErrno(); + } return noErr; } diff --git a/cpp/src/IceSSL/SecureTransportTransceiverI.h b/cpp/src/IceSSL/SecureTransportTransceiverI.h index e0221db2123..5da9e628784 100644 --- a/cpp/src/IceSSL/SecureTransportTransceiverI.h +++ b/cpp/src/IceSSL/SecureTransportTransceiverI.h @@ -17,6 +17,7 @@ #include <Ice/Transceiver.h> #include <Ice/Network.h> +#include <Ice/StreamSocket.h> #ifdef ICE_USE_SECURE_TRANSPORT @@ -29,18 +30,8 @@ namespace IceSSL class ConnectorI; class AcceptorI; -class TransceiverI : public IceInternal::Transceiver, public IceInternal::NativeInfo +class TransceiverI : public IceInternal::Transceiver { - enum State - { - StateNeedConnect, - StateConnectPending, - StateProxyConnectRequest, - StateProxyConnectRequestPending, - StateConnected, - StateHandshakeComplete - }; - public: virtual IceInternal::NativeInfoPtr getNativeInfo(); @@ -62,29 +53,20 @@ public: private: - TransceiverI(const InstancePtr&, SOCKET, const IceInternal::NetworkProxyPtr&, const std::string&, - const IceInternal::Address&, const IceInternal::Address&); - TransceiverI(const InstancePtr&, SOCKET, const std::string&); + TransceiverI(const InstancePtr&, const IceInternal::StreamSocketPtr&, const std::string&, const std::string&); virtual ~TransceiverI(); virtual NativeConnectionInfoPtr getNativeConnectionInfo() const; - bool writeRaw(IceInternal::Buffer&); - bool readRaw(IceInternal::Buffer&); - friend class ConnectorI; friend class AcceptorI; const InstancePtr _instance; const SecureTransportEnginePtr _engine; - - const IceInternal::NetworkProxyPtr _proxy; const std::string _host; - const IceInternal::Address _addr; - const IceInternal::Address _sourceAddr; - const std::string _adapterName; const bool _incoming; + const IceInternal::StreamSocketPtr _stream; SSLContextRef _ssl; SecTrustRef _trust; @@ -97,11 +79,8 @@ private: }; mutable Ice::Byte _flags; - - State _state; - std::string _desc; size_t _maxSendPacketSize; - size_t _maxReceivePacketSize; + size_t _maxRecvPacketSize; }; typedef IceUtil::Handle<TransceiverI> TransceiverIPtr; |