You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2007/04/30 13:49:57 UTC
svn commit: r533710 - in
/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network:
ServerSocket.cpp SocketError.cpp SocketError.h SocketFactory.cpp
SocketFactory.h SocketInputStream.cpp TcpSocket.cpp TcpSocket.h
Author: tabish
Date: Mon Apr 30 04:49:56 2007
New Revision: 533710
URL: http://svn.apache.org/viewvc?view=rev&rev=533710
Log:
Merging in fixes for 2.0.1
Modified:
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/ServerSocket.cpp
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.cpp
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.h
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.cpp
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.h
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketInputStream.cpp
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.cpp
activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.h
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/ServerSocket.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/ServerSocket.cpp?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/ServerSocket.cpp (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/ServerSocket.cpp Mon Apr 30 04:49:56 2007
@@ -199,8 +199,18 @@
int temp_len = sizeof( sockaddr_in );
#endif
- SocketHandle ss_socket_handle =
- ::accept( socketHandle, reinterpret_cast<struct sockaddr*>(&temp), &temp_len );
+ SocketHandle ss_socket_handle = NULL;
+
+ // Loop to ignore any signal interruptions that occur during the operation.
+ do {
+
+ ss_socket_handle = ::accept( socketHandle,
+ reinterpret_cast<struct sockaddr*>(&temp),
+ &temp_len );
+
+ } while( ss_socket_handle < 0 &&
+ SocketError::getErrorCode() == SocketError::INTERRUPTED );
+
if( ss_socket_handle < 0 ) {
throw SocketException( __FILE__, __LINE__,
"ServerSocket::accept- %s", SocketError::getErrorString().c_str() );
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.cpp?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.cpp (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.cpp Mon Apr 30 04:49:56 2007
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
#include "SocketError.h"
#include <activemq/util/Config.h>
@@ -10,15 +27,36 @@
using namespace activemq;
using namespace activemq::network;
+// Platform constants.
+#if defined(HAVE_WINSOCK2_H)
+ const int SocketError::INTERRUPTED = WSAEINTR;
+#else
+ const int SocketError::INTERRUPTED = EINTR;
+#endif
+
+////////////////////////////////////////////////////////////////////////////////
+int SocketError::getErrorCode() {
+
+ #if defined(HAVE_WINSOCK2_H)
+
+ return ::WSAGetLastError();
+
+ #else
+
+ return errno;
+
+ #endif
+}
+
////////////////////////////////////////////////////////////////////////////////
std::string SocketError::getErrorString() {
std::string returnValue;
- #if defined(HAVE_WINSOCK2_H)
+ // Get the error code.
+ int errorCode = getErrorCode();
- // If the socket was temporarily unavailable - just try again.
- int errorCode = ::WSAGetLastError();
+ #if defined(HAVE_WINSOCK2_H)
// Create the error string.
static const int errorStringSize = 512;
@@ -37,7 +75,7 @@
#else
// Create the error string.
- returnValue = ::strerror(errno);
+ returnValue = ::strerror(errorCode);
#endif
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.h?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.h (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketError.h Mon Apr 30 04:49:56 2007
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
#ifndef ACTIVEMQ_NETWORK_SOCKETERROR_H_
#define ACTIVEMQ_NETWORK_SOCKETERROR_H_
@@ -6,9 +23,28 @@
namespace activemq{
namespace network{
- class SocketError {
+ /**
+ * Static utility class to simplify handling of error codes
+ * for socket operations.
+ */
+ class SocketError {
+ public:
+
+ /**
+ * Indicates that a socket operation was interrupted by a signal.
+ */
+ static const int INTERRUPTED;
+
public:
+ /**
+ * Gets the last error appropriate for the platform.
+ */
+ static int getErrorCode();
+
+ /**
+ * Gets the string description for the last error.
+ */
static std::string getErrorString();
};
}}
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.cpp?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.cpp (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.cpp Mon Apr 30 04:49:56 2007
@@ -27,73 +27,77 @@
using namespace activemq::exceptions;
////////////////////////////////////////////////////////////////////////////////
-Socket* SocketFactory::createSocket(const Properties& properties)
- throw ( SocketException )
+Socket* SocketFactory::createSocket(
+ const std::string& uri,
+ const Properties& properties)
+ throw ( SocketException )
{
try
{
- const char* uri = properties.getProperty( "uri" );
- if( uri == NULL )
+ // Ensure something is actually passed in for the URI
+ if( uri == "" )
{
- throw SocketException( __FILE__, __LINE__,
+ throw SocketException( __FILE__, __LINE__,
"SocketTransport::start() - uri not provided" );
}
string dummy = uri;
-
+
// Extract the port.
std::size_t portIx = dummy.find( ':' );
if( portIx == string::npos )
{
- throw SocketException( __FILE__, __LINE__,
- "SocketTransport::start() - uri malformed - port not specified: %s", uri);
+ throw SocketException( __FILE__, __LINE__,
+ "SocketTransport::start() - uri malformed - port not specified: %s", uri.c_str() );
}
string host = dummy.substr( 0, portIx );
string portString = dummy.substr( portIx + 1 );
int port;
if( sscanf( portString.c_str(), "%d", &port) != 1 )
{
- throw SocketException( __FILE__, __LINE__,
- "SocketTransport::start() - unable to extract port from uri: %s", uri);
+ throw SocketException( __FILE__, __LINE__,
+ "SocketTransport::start() - unable to extract port from uri: %s", uri.c_str() );
}
-
+
// Get the read buffer size.
int inputBufferSize = 10000;
- dummy = properties.getProperty( "inputBufferSize", "10000" );
+ dummy = properties.getProperty( "inputBufferSize", "10000" );
sscanf( dummy.c_str(), "%d", &inputBufferSize );
-
+
// Get the write buffer size.
int outputBufferSize = 10000;
- dummy = properties.getProperty( "outputBufferSize", "10000" );
+ dummy = properties.getProperty( "outputBufferSize", "10000" );
sscanf( dummy.c_str(), "%d", &outputBufferSize );
-
+
// Get the linger flag.
int soLinger = 0;
- dummy = properties.getProperty( "soLinger", "0" );
- sscanf( dummy.c_str(), "%d", &soLinger );
-
+ dummy = properties.getProperty( "soLinger", "0" );
+ sscanf( dummy.c_str(), "%d", &soLinger );
+
// Get the keepAlive flag.
- bool soKeepAlive =
- properties.getProperty( "soKeepAlive", "false" ) == "true";
-
+ bool soKeepAlive =
+ properties.getProperty( "soKeepAlive", "false" ) == "true";
+
// Get the socket receive buffer size.
int soReceiveBufferSize = -1;
- dummy = properties.getProperty( "soReceiveBufferSize", "-1" );
+ dummy = properties.getProperty( "soReceiveBufferSize", "-1" );
sscanf( dummy.c_str(), "%d", &soReceiveBufferSize );
-
+
// Get the socket send buffer size.
int soSendBufferSize = -1;
- dummy = properties.getProperty( "soSendBufferSize", "-1" );
+ dummy = properties.getProperty( "soSendBufferSize", "-1" );
sscanf( dummy.c_str(), "%d", &soSendBufferSize );
-
+
+ // Get the socket TCP_NODELAY flag.
+ bool tcpNoDelay =
+ properties.getProperty( "tcpNoDelay", "true" ) == "true";
+
// Now that we have all the elements that we wanted - let's do it!
// Create a TCP Socket and then Wrap it in a buffered socket
// so that users get the benefit of buffered reads and writes.
// The buffered socket will own the TcpSocket instance, and will
// clean it up when it is cleaned up.
TcpSocket* tcpSocket = new TcpSocket();
- /*BufferedSocket* bufferedSocket =
- new BufferedSocket(tcpSocket, inputBufferSize, outputBufferSize);*/
try
{
@@ -103,6 +107,7 @@
// Set the socket options.
tcpSocket->setSoLinger( soLinger );
tcpSocket->setKeepAlive( soKeepAlive );
+ tcpSocket->setTcpNoDelay( tcpNoDelay );
if( soReceiveBufferSize > 0 ){
tcpSocket->setReceiveBufferSize( soReceiveBufferSize );
@@ -111,6 +116,7 @@
if( soSendBufferSize > 0 ){
tcpSocket->setSendBufferSize( soSendBufferSize );
}
+
}
catch ( SocketException& ex )
{
@@ -118,7 +124,7 @@
try{
delete tcpSocket;
} catch( SocketException& ex2 ){ /* Absorb */ }
-
+
throw ex;
}
@@ -126,5 +132,5 @@
}
AMQ_CATCH_RETHROW( SocketException )
AMQ_CATCH_EXCEPTION_CONVERT( ActiveMQException, SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.h?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.h (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketFactory.h Mon Apr 30 04:49:56 2007
@@ -24,7 +24,7 @@
namespace network{
class Socket;
-
+
/**
* Socket Factory implementation for use in Creating Sockets
* <p>
@@ -32,7 +32,6 @@
* Property Options: <p>
* Name Value <p>
* ------------------------------------- <p>
- * uri The uri for the transport connection. Must be provided.<p>
* inputBufferSize size in bytes of the buffered input stream buffer. Defaults to 10000.<p>
* outputBufferSize size in bytes of the buffered output stream buffer. Defaults to 10000.<p>
* soLinger linger time for the socket (in microseconds). Defaults to 0.<p>
@@ -40,24 +39,26 @@
* soReceiveBufferSize The size of the socket receive buffer (in bytes). Defaults to 2MB.<p>
* soSendBufferSize The size of the socket send buffer (in bytes). Defaults to 2MB.<p>
* soTimeout The timeout of socket IO operations (in microseconds). Defaults to 10000<p>
- *
+ *
* @see <code>Socket</code>
*/
class SocketFactory
{
public:
- virtual ~SocketFactory();
-
+ virtual ~SocketFactory();
+
/**
* Creates and returns a Socket dervied Object based on the values
* defined in the Properties Object that is passed in.
+ * @param the URI for the Socket Connection.
* @param properties a IProperties pointer.
* @throws SocketException.
*/
- static Socket* createSocket( const util::Properties& properties )
+ static Socket* createSocket( const std::string& uri,
+ const util::Properties& properties )
throw ( SocketException );
-
+
};
}}
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketInputStream.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketInputStream.cpp?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketInputStream.cpp (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/SocketInputStream.cpp Mon Apr 30 04:49:56 2007
@@ -134,14 +134,23 @@
std::size_t SocketInputStream::read( unsigned char* buffer,
std::size_t bufferSize ) throw (IOException)
{
- int len = ::recv(socket, (char*)buffer, (int)bufferSize, 0);
+ int len = 0;
- // Check for a closed socket.
- if( len == 0 ){
- throw IOException( __FILE__, __LINE__,
- "activemq::io::SocketInputStream::read - The connection is broken" );
- }
+ // Loop to ignore any signal interruptions that occur during the read.
+ do {
+
+ // Read data from the socket.
+ len = ::recv(socket, (char*)buffer, (int)bufferSize, 0);
+ // Check for a closed socket.
+ if( len == 0 ){
+ throw IOException( __FILE__, __LINE__,
+ "activemq::io::SocketInputStream::read - The connection is broken" );
+ }
+
+ } while( len == -1 &&
+ SocketError::getErrorCode() == SocketError::INTERRUPTED );
+
// Check for error.
if( len == -1 ){
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.cpp?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.cpp (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.cpp Mon Apr 30 04:49:56 2007
@@ -16,13 +16,13 @@
*/
#include <activemq/util/Config.h>
-#if defined(HAVE_WINSOCK2_H)
+#if defined(HAVE_WINSOCK2_H)
#include <Winsock2.h>
- #include <Ws2tcpip.h>
+ #include <Ws2tcpip.h>
#include <sys/stat.h>
#define stat _stat
#else
- #include <unistd.h>
+ #include <unistd.h>
#include <netdb.h>
#include <fcntl.h>
#include <sys/file.h>
@@ -30,9 +30,10 @@
#include <netinet/in.h>
#include <arpa/inet.h>
#include <string.h>
+ #include <netinet/tcp.h>
#endif
-#ifndef SHUT_RDWR
+#ifndef SHUT_RDWR
#define SHUT_RDWR 2 // Winsock2 doesn't seem to define this
#endif
@@ -57,7 +58,7 @@
TcpSocket::StaticSocketInitializer::StaticSocketInitializer() {
socketInitError = NULL;
const WORD version_needed = MAKEWORD(2,2); // lo-order byte: major version
- WSAData temp;
+ WSAData temp;
if( WSAStartup( version_needed, &temp ) ){
clear();
socketInitError = new SocketException ( __FILE__, __LINE__,
@@ -68,14 +69,14 @@
clear();
WSACleanup();
}
-
+
// Create static instance of the socket initializer.
TcpSocket::StaticSocketInitializer TcpSocket::staticSocketInitializer;
-
+
#endif
////////////////////////////////////////////////////////////////////////////////
-TcpSocket::TcpSocket() throw (SocketException)
+TcpSocket::TcpSocket() throw (SocketException)
:
socketHandle( INVALID_SOCKET_HANDLE ),
inputStream( NULL ),
@@ -83,7 +84,7 @@
{
try {
-
+
#if defined(HAVE_WINSOCK2_H)
if( staticSocketInitializer.getSocketInitError() != NULL ) {
throw *staticSocketInitializer.getSocketInitError();
@@ -102,19 +103,19 @@
outputStream( NULL )
{
try {
-
+
#if defined(HAVE_WINSOCK2_H)
if( staticSocketInitializer.getSocketInitError() != NULL ) {
throw *staticSocketInitializer.getSocketInitError();
}
#endif
-
+
this->socketHandle = socketHandle;
this->inputStream = new SocketInputStream( socketHandle );
this->outputStream = new SocketOutputStream( socketHandle );
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
@@ -138,67 +139,67 @@
void TcpSocket::connect(const char* host, int port) throw ( SocketException )
{
try{
-
+
if( isConnected() ) {
- throw SocketException( __FILE__, __LINE__,
+ throw SocketException( __FILE__, __LINE__,
"Socket::connect - Socket already connected. host: %s, port: %d", host, port );
}
-
+
// Create the socket.
checkResult( (int)(socketHandle = ::socket(AF_INET, SOCK_STREAM, 0)) );
-
+
// Check port value.
if (port <= 0 || port > 65535) {
close();
- throw SocketException ( __FILE__, __LINE__,
+ throw SocketException ( __FILE__, __LINE__,
"Socket::connect- Port out of range: %d", port );
}
-
+
#ifdef SO_NOSIGPIPE // Don't want to get a SIGPIPE on FreeBSD and Mac OS X
-
+
int optval = 1;
checkResult( ::setsockopt( socketHandle, SOL_SOCKET, SO_NOSIGPIPE, (char*)&optval, sizeof(optval)) );
-
+
#endif
-
+
sockaddr_in target_addr;
target_addr.sin_family = AF_INET;
target_addr.sin_port = htons( ( short ) port );
target_addr.sin_addr.s_addr = 0; // To be set later down...
memset( &target_addr.sin_zero, 0, sizeof( target_addr.sin_zero ) );
-
+
// Resolve name
-#if defined(HAVE_STRUCT_ADDRINFO)
+#if defined(HAVE_STRUCT_ADDRINFO)
addrinfo hints;
memset(&hints, 0, sizeof(addrinfo));
hints.ai_family = PF_INET;
struct addrinfo *res_ptr = NULL;
-
+
checkResult( ::getaddrinfo( host, NULL, &hints, &res_ptr ) );
-
+
assert(res_ptr->ai_addr->sa_family == AF_INET);
- // Porting: On both 32bit and 64 bit systems that we compile to soo far, sin_addr
+ // Porting: On both 32bit and 64 bit systems that we compile to soo far, sin_addr
// is a 32 bit value, not an unsigned long.
assert( sizeof( ( ( sockaddr_in* )res_ptr->ai_addr )->sin_addr.s_addr ) == 4 );
target_addr.sin_addr.s_addr = ( ( sockaddr_in* )res_ptr->ai_addr )->sin_addr.s_addr;
freeaddrinfo( res_ptr );
#else
- struct ::hostent *he = ::gethostbyname(host);
- if( he == NULL ) {
+ struct ::hostent *he = ::gethostbyname(host);
+ if( he == NULL ) {
throw SocketException( __FILE__, __LINE__, "Failed to resolve hostname" );
- }
- target_addr.sin_addr.s_addr = *((in_addr_t *)he->h_addr);
+ }
+ target_addr.sin_addr.s_addr = *((in_addr_t *)he->h_addr);
#endif
-
+
// Attempt the connection to the server.
- checkResult( ::connect( socketHandle,
- ( const sockaddr * )&target_addr,
+ checkResult( ::connect( socketHandle,
+ ( const sockaddr * )&target_addr,
sizeof( target_addr ) ) );
-
+
// Create an input/output stream for this socket.
inputStream = new SocketInputStream( socketHandle );
outputStream = new SocketOutputStream( socketHandle );
-
+
}
catch( SocketException& ex ) {
ex.setMark( __FILE__, __LINE__);
@@ -219,42 +220,42 @@
delete inputStream;
inputStream = NULL;
}
-
+
// Destroy the output stream.
if( outputStream != NULL ){
delete outputStream;
outputStream = NULL;
}
-
+
if( isConnected() )
{
::shutdown( socketHandle, SHUT_RDWR );
-
- #if !defined(HAVE_WINSOCK2_H)
+
+ #if !defined(HAVE_WINSOCK2_H)
::close( socketHandle );
#else
::closesocket( socketHandle );
#endif
-
+
socketHandle = INVALID_SOCKET_HANDLE;
}
}
////////////////////////////////////////////////////////////////////////////////
int TcpSocket::getSoLinger() const throw( SocketException ){
-
+
try{
linger value;
socklen_t length = sizeof( value );
checkResult(::getsockopt( socketHandle, SOL_SOCKET, SO_LINGER, (char*)&value, &length ));
-
+
return value.l_onoff? value.l_linger : 0;
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
-////////////////////////////////////////////////////////////////////////////////
+////////////////////////////////////////////////////////////////////////////////
void TcpSocket::setSoLinger( int dolinger ) throw( SocketException ){
try{
@@ -264,12 +265,12 @@
checkResult(::setsockopt( socketHandle, SOL_SOCKET, SO_LINGER, (char*)&value, sizeof(value) ));
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
bool TcpSocket::getKeepAlive() const throw( SocketException ){
-
+
try{
int value;
socklen_t length = sizeof( int );
@@ -277,23 +278,23 @@
return value != 0;
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
void TcpSocket::setKeepAlive( const bool keepAlive ) throw( SocketException ){
-
+
try{
int value = keepAlive? 1 : 0;
checkResult(::setsockopt(socketHandle, SOL_SOCKET, SO_KEEPALIVE, (char*)&value, sizeof(int)) );
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
int TcpSocket::getReceiveBufferSize() const throw( SocketException ){
-
+
try{
int value;
socklen_t length = sizeof( value );
@@ -301,22 +302,22 @@
return value;
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
void TcpSocket::setReceiveBufferSize( int size ) throw( SocketException ){
-
+
try{
checkResult(::setsockopt( socketHandle, SOL_SOCKET, SO_RCVBUF, (char*)&size, sizeof(size) ));
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
bool TcpSocket::getReuseAddress() const throw( SocketException ){
-
+
try{
int value;
socklen_t length = sizeof( int );
@@ -324,23 +325,23 @@
return value != 0;
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
void TcpSocket::setReuseAddress( bool reuse ) throw( SocketException ){
-
+
try{
int value = reuse? 1 : 0;
checkResult(::setsockopt( socketHandle, SOL_SOCKET, SO_REUSEADDR, (char*)&value, sizeof(int) ));
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
int TcpSocket::getSendBufferSize() const throw( SocketException ){
-
+
try{
int value;
socklen_t length = sizeof( value );
@@ -348,24 +349,24 @@
return value;
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
void TcpSocket::setSendBufferSize( int size ) throw( SocketException ){
-
+
try{
checkResult(::setsockopt( socketHandle, SOL_SOCKET, SO_SNDBUF, (char*)&size, sizeof(size) ));
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
void TcpSocket::setSoTimeout ( const int millisecs ) throw ( SocketException )
{
try{
-
+
#if !defined(HAVE_WINSOCK2_H)
timeval timot;
timot.tv_sec = millisecs / 1000;
@@ -373,19 +374,19 @@
#else
int timot = millisecs;
#endif
-
+
checkResult(::setsockopt( socketHandle, SOL_SOCKET, SO_RCVTIMEO, (const char*) &timot, sizeof (timot) ));
checkResult(::setsockopt( socketHandle, SOL_SOCKET, SO_SNDTIMEO, (const char*) &timot, sizeof (timot) ));
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
int TcpSocket::getSoTimeout() const throw( SocketException )
{
try{
-
+
#if !defined(HAVE_WINSOCK2_H)
timeval timot;
timot.tv_sec = 0;
@@ -395,26 +396,50 @@
int timot = 0;
int size = sizeof(timot);
#endif
-
+
checkResult(::getsockopt(socketHandle, SOL_SOCKET, SO_RCVTIMEO, (char*) &timot, &size));
-
+
#if !defined(HAVE_WINSOCK2_H)
return (timot.tv_sec * 1000) + (timot.tv_usec / 1000);
#else
return timot;
#endif
-
+
}
AMQ_CATCH_RETHROW( SocketException )
- AMQ_CATCHALL_THROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
}
////////////////////////////////////////////////////////////////////////////////
+bool TcpSocket::getTcpNoDelay() const throw ( cms::CMSException ) {
+
+ try{
+ int value;
+ socklen_t length = sizeof( int );
+ checkResult(::getsockopt( socketHandle, IPPROTO_TCP, TCP_NODELAY, (char*)&value, &length ));
+ return value != 0;
+ }
+ AMQ_CATCH_RETHROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void TcpSocket::setTcpNoDelay( bool value ) throw ( cms::CMSException ) {
+
+ try{
+ int ivalue = value ? 1 : 0;
+ checkResult(::setsockopt( socketHandle, IPPROTO_TCP, TCP_NODELAY, (char*)&ivalue, sizeof(int) ));
+ }
+ AMQ_CATCH_RETHROW( SocketException )
+ AMQ_CATCHALL_THROW( SocketException )
+}
+
+////////////////////////////////////////////////////////////////////////////////
void TcpSocket::checkResult( int value ) const throw (SocketException) {
-
+
if( value < 0 ){
- throw SocketException( __FILE__, __LINE__,
+ throw SocketException( __FILE__, __LINE__,
SocketError::getErrorString().c_str() );
}
}
Modified: activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.h?view=diff&rev=533710&r1=533709&r2=533710
==============================================================================
--- activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.h (original)
+++ activemq/activemq-cpp/tags/activemq-cpp-2.0.1/src/main/activemq/network/TcpSocket.h Mon Apr 30 04:49:56 2007
@@ -26,56 +26,56 @@
namespace activemq{
namespace network{
-
+
// Forward declarations
class SocketInputStream;
class SocketOutputStream;
-
+
/**
* Platform-independent implementation of the socket interface.
*/
class TcpSocket : public Socket
- {
+ {
private:
-
+
/**
* The handle for this socket.
*/
SocketHandle socketHandle;
-
+
/**
* The input stream for reading this socket.
*/
SocketInputStream* inputStream;
-
+
/**
* The output stream for writing to this socket.
*/
SocketOutputStream* outputStream;
-
+
public:
-
- /**
+
+ /**
* Construct a non-connected socket.
* @throws SocketException thrown one windows if the static initialization
* call to WSAStartup was not successful.
*/
TcpSocket() throw (SocketException);
-
- /**
+
+ /**
* Construct a connected or bound socket based on given
* socket handle.
* @param socketHandle a socket handle to wrap in the object
*/
TcpSocket( SocketHandle socketHandle );
-
+
/**
* Destruct.
* Releases the socket handle but not
* gracefully shut down the connection.
*/
virtual ~TcpSocket();
-
+
/**
* Gets the handle for the socket.
* @return SocketHabler for this Socket, can be NULL
@@ -83,16 +83,16 @@
SocketHandle getSocketHandle () {
return socketHandle;
}
-
+
/**
- * Connects to the specified destination. Closes this socket if
+ * Connects to the specified destination. Closes this socket if
* connected to another destination.
* @param host The host of the server to connect to.
* @param port The port of the server to connect to.
* @throws IOException Thrown if a failure occurred in the connect.
*/
virtual void connect( const char* host, int port ) throw( SocketException );
-
+
/**
* Indicates whether or not this socket is connected to a destination.
* @return true if connected
@@ -100,96 +100,96 @@
virtual bool isConnected() const{
return socketHandle != INVALID_SOCKET_HANDLE;
}
-
+
/**
* Gets the InputStream for this socket.
* @return The InputStream for this socket. NULL if not connected.
*/
virtual io::InputStream* getInputStream();
-
+
/**
* Gets the OutputStream for this socket.
* @return the OutputStream for this socket. NULL if not connected.
*/
virtual io::OutputStream* getOutputStream();
-
+
/**
* Gets the linger time.
* @return The linger time in seconds.
* @throws SocketException if the operation fails.
*/
virtual int getSoLinger() const throw( SocketException );
-
+
/**
* Sets the linger time.
* @param linger The linger time in seconds. If 0, linger is off.
* @throws SocketException if the operation fails.
*/
virtual void setSoLinger( int linger ) throw( SocketException );
-
+
/**
* Gets the keep alive flag.
* @return True if keep alive is enabled.
* @throws SocketException if the operation fails.
*/
virtual bool getKeepAlive() const throw( SocketException );
-
+
/**
* Enables/disables the keep alive flag.
* @param keepAlive If true, enables the flag.
* @throws SocketException if the operation fails.
*/
virtual void setKeepAlive( bool keepAlive ) throw( SocketException );
-
+
/**
* Gets the receive buffer size.
* @return the receive buffer size in bytes.
* @throws SocketException if the operation fails.
*/
virtual int getReceiveBufferSize() const throw( SocketException );
-
+
/**
* Sets the recieve buffer size.
* @param size Number of bytes to set the receive buffer to.
* @throws SocketException if the operation fails.
*/
virtual void setReceiveBufferSize( int size ) throw( SocketException );
-
+
/**
* Gets the reuse address flag.
* @return True if the address can be reused.
* @throws SocketException if the operation fails.
*/
virtual bool getReuseAddress() const throw( SocketException );
-
+
/**
* Sets the reuse address flag.
* @param reuse If true, sets the flag.
* @throws SocketException if the operation fails.
*/
virtual void setReuseAddress( bool reuse ) throw( SocketException );
-
+
/**
* Gets the send buffer size.
* @return the size in bytes of the send buffer.
* @throws SocketException if the operation fails.
*/
virtual int getSendBufferSize() const throw( SocketException );
-
+
/**
* Sets the send buffer size.
* @param size The number of bytes to set the send buffer to.
* @throws SocketException if the operation fails.
*/
virtual void setSendBufferSize( int size ) throw( SocketException );
-
+
/**
* Gets the timeout for socket operations.
* @return The timeout in milliseconds for socket operations.
* @throws SocketException Thrown if unable to retrieve the information.
*/
virtual int getSoTimeout() const throw( SocketException );
-
+
/**
* Sets the timeout for socket operations.
* @param timeout The timeout in milliseconds for socket operations.<p>
@@ -202,24 +202,40 @@
* @throws CMSException
*/
virtual void close() throw( cms::CMSException );
-
+
+ public:
+
+ /**
+ * Gets the Status of the TCP_NODELAY param for this socket as a Bool
+ * @returns true if TCP_NODELAY is enabled
+ * @throws CMSException
+ */
+ virtual bool getTcpNoDelay() const throw ( cms::CMSException );
+
+ /**
+ * Sets the Status of the TCP_NODELAY param for this socket as a Bool
+ * @param value - true if TCP_NODELAY is to be enabled
+ * @throws CMSException
+ */
+ virtual void setTcpNoDelay( bool value ) throw ( cms::CMSException );
+
protected:
-
- #if defined(HAVE_WINSOCK2_H)
-
+
+ #if defined(HAVE_WINSOCK2_H)
+
// WINDOWS needs initialization of winsock
class StaticSocketInitializer {
private:
-
+
SocketException* socketInitError;
-
+
void clear(){
if( socketInitError != NULL ){
delete socketInitError;
}
socketInitError = NULL;
}
-
+
public:
SocketException* getSocketInitError() {
@@ -230,12 +246,12 @@
virtual ~StaticSocketInitializer();
};
-
+
static StaticSocketInitializer staticSocketInitializer;
#endif
-
+
void checkResult( int value ) const throw (SocketException);
-
+
};
}}