Author: tabish
Date: Wed Apr 18 10:44:28 2007
New Revision: 530101
URL: http://svn.apache.org/viewvc?view=rev&rev=530101
Log:
http://issues.apache.org/activemq/browse/AMQCPP-101
Modified:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.h
Modified:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.cpp
URL:
http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.cpp?view=diff&rev=530101&r1=530100&r2=530101
==============================================================================
---
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.cpp
(original)
+++
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.cpp
Wed Apr 18 10:44:28 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>
@@ -32,7 +32,7 @@
#include <string.h>
#endif
-#ifndef SHUT_RDWR
+#ifndef SHUT_RDWR
#define SHUT_RDWR 2 // Winsock2 doesn't seem to define this
#endif
@@ -57,7 +57,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 +68,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 +83,7 @@
{
try {
-
+
#if defined(HAVE_WINSOCK2_H)
if( staticSocketInitializer.getSocketInitError() != NULL ) {
throw *staticSocketInitializer.getSocketInitError();
@@ -102,19 +102,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 +138,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 +219,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 +264,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 +277,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 +301,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 +324,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 +348,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 +373,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 +395,35 @@
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 ) {
+ return false;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void TcpSocket::setTcpNoDelay( bool value ) throw ( cms::CMSException ) {
+}
+
+////////////////////////////////////////////////////////////////////////////////
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/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.h
URL:
http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.h?view=diff&rev=530101&r1=530100&r2=530101
==============================================================================
---
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.h
(original)
+++
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/network/TcpSocket.h
Wed Apr 18 10:44:28 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);
-
+
};
}}
