diff options
Diffstat (limited to 'cpp/src/Ice/uwp/TcpTransceiver.cpp')
| -rw-r--r-- | cpp/src/Ice/uwp/TcpTransceiver.cpp | 399 |
1 files changed, 399 insertions, 0 deletions
diff --git a/cpp/src/Ice/uwp/TcpTransceiver.cpp b/cpp/src/Ice/uwp/TcpTransceiver.cpp new file mode 100644 index 00000000000..abe855ced7b --- /dev/null +++ b/cpp/src/Ice/uwp/TcpTransceiver.cpp @@ -0,0 +1,399 @@ +// ********************************************************************** +// +// Copyright (c) 2003-2015 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/uwp/TcpTransceiver.h> +#include <Ice/Connection.h> +#include <Ice/ProtocolInstance.h> +#include <Ice/LoggerUtil.h> +#include <Ice/Buffer.h> +#include <Ice/LocalException.h> +#include <Ice/Properties.h> + +using namespace std; +using namespace Ice; +using namespace IceInternal; + +using namespace Platform; +using namespace Windows::Foundation; +using namespace Windows::Storage::Streams; +using namespace Windows::Networking; +using namespace Windows::Networking::Sockets; + +namespace +{ + +AsyncOperationCompletedHandler<unsigned int>^ +createAsyncOperationCompletedHandler(SocketOperationCompletedHandler^ cb, SocketOperation op, AsyncInfo& info) +{ + return ref new AsyncOperationCompletedHandler<unsigned int>( + [=,&info] (IAsyncOperation<unsigned int>^ operation, Windows::Foundation::AsyncStatus status) + { + if(status != Windows::Foundation::AsyncStatus::Completed) + { + info.count = SOCKET_ERROR; + info.error = operation->ErrorCode.Value; + } + else + { + info.count = static_cast<int>(operation->GetResults()); + } + cb(op); + }); +} + +} + +NativeInfoPtr +IceInternal::TcpTransceiver::getNativeInfo() +{ + return this; +} + +void +IceInternal::TcpTransceiver::setCompletedHandler(SocketOperationCompletedHandler^ handler) +{ + _completedHandler = handler; + _readOperationCompletedHandler = createAsyncOperationCompletedHandler(handler, SocketOperationRead, _read); + _writeOperationCompletedHandler = createAsyncOperationCompletedHandler(handler, SocketOperationWrite, _write); +} + +SocketOperation +IceInternal::TcpTransceiver::initialize(Buffer&, Buffer&) +{ + if(_state == StateNeedConnect) + { + _state = StateConnectPending; + return SocketOperationConnect; + } + else if(_state <= StateConnectPending) + { + if(_write.count == SOCKET_ERROR) + { + checkConnectErrorCode(__FILE__, __LINE__, _write.error, _connectAddr.host); + } + _state = StateConnected; + _desc = fdToString(_fd); + } + assert(_state == StateConnected); + return SocketOperationNone; +} + +SocketOperation +#ifdef ICE_CPP11_MAPPING +IceInternal::TcpTransceiver::closing(bool initiator, exception_ptr) +#else +IceInternal::TcpTransceiver::closing(bool initiator, const Ice::LocalException&) +#endif +{ + // If we are initiating the connection closure, wait for the peer + // to close the TCP/IP connection. Otherwise, close immediately. + return initiator ? SocketOperationRead : SocketOperationNone; +} + +void +IceInternal::TcpTransceiver::close() +{ + assert(_fd != INVALID_SOCKET); + + _completedHandler = nullptr; + _readOperationCompletedHandler = nullptr; + _writeOperationCompletedHandler = nullptr; + + try + { + closeSocket(_fd); + _fd = INVALID_SOCKET; + } + catch(const SocketException&) + { + _fd = INVALID_SOCKET; + throw; + } +} + +SocketOperation +IceInternal::TcpTransceiver::write(Buffer& buf) +{ + return buf.i == buf.b.end() ? SocketOperationNone : SocketOperationWrite; +} + +SocketOperation +IceInternal::TcpTransceiver::read(Buffer& buf) +{ + return buf.i == buf.b.end() ? SocketOperationNone : SocketOperationRead; +} + +bool +IceInternal::TcpTransceiver::startWrite(Buffer& buf) +{ + if(_state < StateConnected) + { + try + { + IAsyncAction^ action = safe_cast<StreamSocket^>(_fd)->ConnectAsync( + _connectAddr.host, + _connectAddr.port, + SocketProtectionLevel::PlainSocket); + + if(!checkIfErrorOrCompleted(SocketOperationConnect, action)) + { + SocketOperationCompletedHandler^ completed = _completedHandler; + action->Completed = ref new AsyncActionCompletedHandler( + [=] (IAsyncAction^ info, Windows::Foundation::AsyncStatus status) + { + if(status != Windows::Foundation::AsyncStatus::Completed) + { + _write.count = SOCKET_ERROR; + _write.error = info->ErrorCode.Value; + } + else + { + _write.count = 0; + } + completed(SocketOperationConnect); + }); + } + } + catch(Platform::Exception^ ex) + { + checkConnectErrorCode(__FILE__, __LINE__, ex->HResult, _connectAddr.host); + } + 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); + _writer->WriteBytes(ref new Array<unsigned char>(&*buf.i, packetSize)); + try + { + DataWriterStoreOperation^ operation = _writer->StoreAsync(); + if(checkIfErrorOrCompleted(SocketOperationWrite, operation)) + { + _write.count = operation->GetResults(); + } + else + { + operation->Completed = _writeOperationCompletedHandler; + } + } + catch(Platform::Exception^ ex) + { + checkErrorCode(__FILE__, __LINE__, ex->HResult); + } + return packetSize == static_cast<int>(buf.b.end() - buf.i); +} + +void +IceInternal::TcpTransceiver::finishWrite(Buffer& buf) +{ + if(_state < StateConnected) + { + if(_write.count == SOCKET_ERROR) + { + checkConnectErrorCode(__FILE__, __LINE__, _write.error, _connectAddr.host); + } + _verified = true; + return; + } + + if(_write.count == SOCKET_ERROR) + { + checkErrorCode(__FILE__, __LINE__, _write.error); + } + + buf.i += _write.count; +} + +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()); + + try + { + DataReaderLoadOperation^ operation = _reader->LoadAsync(packetSize); + if(checkIfErrorOrCompleted(SocketOperationRead, operation)) + { + _read.count = operation->GetResults(); + } + else + { + operation->Completed = _readOperationCompletedHandler; + } + } + catch(Platform::Exception^ ex) + { + checkErrorCode(__FILE__, __LINE__, ex->HResult); + } +} + +void +IceInternal::TcpTransceiver::finishRead(Buffer& buf) +{ + if(_read.count == SOCKET_ERROR) + { + checkErrorCode(__FILE__, __LINE__, _read.error); + } + else if(_read.count == 0) + { + ConnectionLostException ex(__FILE__, __LINE__); + ex.error = 0; + throw ex; + } + + try + { + auto data = ref new Platform::Array<unsigned char>(_read.count); + _reader->ReadBytes(data); + memcpy(&*buf.i, data->Data, _read.count); + } + catch(Platform::Exception^ ex) + { + checkErrorCode(__FILE__, __LINE__, ex->HResult); + } + + buf.i += _read.count; +} + +string +IceInternal::TcpTransceiver::protocol() const +{ + return _instance->protocol(); +} + +string +IceInternal::TcpTransceiver::toString() const +{ + return _desc; +} + +string +IceInternal::TcpTransceiver::toDetailedString() const +{ + return toString(); +} + +Ice::ConnectionInfoPtr +IceInternal::TcpTransceiver::getInfo() const +{ + Ice::IPConnectionInfoPtr info; + info = ICE_MAKE_SHARED(Ice::TCPConnectionInfo); + fillConnectionInfo(info); + return info; +} + +Ice::ConnectionInfoPtr +IceInternal::TcpTransceiver::getWSInfo(const Ice::HeaderDict& headers) const +{ + Ice::WSConnectionInfoPtr info = ICE_MAKE_SHARED(Ice::WSConnectionInfo); + fillConnectionInfo(info); + info->headers = headers; + return info; +} + +void +IceInternal::TcpTransceiver::checkSendSize(const Buffer&) +{ +} + + void + IceInternal::TcpTransceiver::setBufferSize(int rcvSize, int sndSize) + { + setTcpBufSize(_fd, rcvSize, sndSize, _instance); + } + +IceInternal::TcpTransceiver::TcpTransceiver(const ProtocolInstancePtr& instance, SOCKET fd, bool connected) : + NativeInfo(fd), + _instance(instance), + _state(connected ? StateConnected : StateNeedConnect), + _desc(connected ? fdToString(_fd) : string()), + _verified(false) +{ + StreamSocket^ streamSocket = safe_cast<StreamSocket^>(_fd); + _writer = ref new DataWriter(streamSocket->OutputStream); + _reader = ref new DataReader(streamSocket->InputStream); + _reader->InputStreamOptions = InputStreamOptions::Partial; + + setTcpBufSize(_fd, _instance); + + _maxSendPacketSize = streamSocket->Control->OutboundBufferSizeInBytes / 2; + if(_maxSendPacketSize < 512) + { + _maxSendPacketSize = 0; + } + + _maxReceivePacketSize = instance->properties()->getPropertyAsIntWithDefault("Ice.TCP.RcvSize", 128 * 1024); +} + +IceInternal::TcpTransceiver::~TcpTransceiver() +{ + assert(_fd == INVALID_SOCKET); +} + +void +IceInternal::TcpTransceiver::connect(const Address& addr) +{ + _connectAddr = addr; +} + +bool +IceInternal::TcpTransceiver::checkIfErrorOrCompleted(SocketOperation op, IAsyncInfo^ info, int count) +{ + // + // NOTE: It's important to only check for info->Status once as it + // might change during the checks below (the Status can be changed + // by the Windows thread pool concurrently). + // + // We consider that a canceled async status is the same as an + // error. A canceled async status can occur if there's a timeout + // and the socket is closed. + // + Windows::Foundation::AsyncStatus status = info->Status; + if(status == Windows::Foundation::AsyncStatus::Completed) + { + _completedHandler(op); + return true; + } + else if (status == Windows::Foundation::AsyncStatus::Started) + { + return false; + } + else + { + if(_state < StateConnected) + { + checkConnectErrorCode(__FILE__, __LINE__, info->ErrorCode.Value, _connectAddr.host); + } + else + { + checkErrorCode(__FILE__, __LINE__, info->ErrorCode.Value); + } + return true; // Prevent compiler warning. + } +} + +void +IceInternal::TcpTransceiver::fillConnectionInfo(const Ice::IPConnectionInfoPtr& info) const +{ + fdToAddressAndPort(_fd, info->localAddress, info->localPort, info->remoteAddress, info->remotePort); + info->rcvSize = getRecvBufferSize(_fd); + info->sndSize = getSendBufferSize(_fd); +} |