src/main/activemq/network/SocketInputStream.cpp

/*
 * 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 <activemq/util/Config.h>

#if !defined(HAVE_WINSOCK2_H)
    #include <sys/select.h>
    #include <sys/socket.h>
#else
    #include <Winsock2.h>
#endif

#ifdef HAVE_SYS_IOCTL_H
    #define BSD_COMP /* Get FIONREAD on Solaris2. */
    #include <sys/ioctl.h>
#endif

// Pick up FIONREAD on Solaris 2.5.
#ifdef HAVE_SYS_FILIO_H
    #include <sys/filio.h>
#endif

#include <activemq/network/SocketInputStream.h>
#include <activemq/network/SocketError.h>
#include <activemq/io/IOException.h>
#include <activemq/util/Character.h>
#include <activemq/exceptions/UnsupportedOperationException.h>
#include <stdlib.h>
#include <string>
#include <stdio.h>
#include <iostream>

using namespace activemq;
using namespace activemq::network;
using namespace activemq::io;
using namespace activemq::util;
using namespace activemq::exceptions;
using namespace std;

////////////////////////////////////////////////////////////////////////////////
SocketInputStream::SocketInputStream(network::Socket::SocketHandle socket) {

    this->socket = socket;
    this->closed = false;
}

////////////////////////////////////////////////////////////////////////////////
SocketInputStream::~SocketInputStream() {
}

////////////////////////////////////////////////////////////////////////////////
void SocketInputStream::close() throw(cms::CMSException ) {
    this->closed = true;
}

////////////////////////////////////////////////////////////////////////////////
std::size_t SocketInputStream::available() const
        throw (activemq::io::IOException) {

    // The windows version
#if defined(HAVE_WINSOCK2_H)

    unsigned long numBytes = 0;

    if( ::ioctlsocket(socket, FIONREAD, &numBytes ) == SOCKET_ERROR ) {
        throw SocketException(
                __FILE__, __LINE__,
                "SocketInputStream::available - ioctlsocket failed" );
    }

    return (std::size_t)numBytes;

#else // !defined(HAVE_WINSOCK2_H)
    // If FIONREAD is defined - use ioctl to find out how many bytes
    // are available.
#if defined(FIONREAD)

    std::size_t numBytes = 0;
    if( ::ioctl (socket, FIONREAD, &numBytes) != -1 ) {
        return numBytes;
    }

#endif

    // If we didn't get anything we can use select.  This is a little
    // less functional.  We will poll on the socket - if there is data
    // available, we'll return 1, otherwise we'll return zero.
#if defined(HAVE_SELECT)

    fd_set rd;
    FD_ZERO(&rd);
    FD_SET( socket, &rd );
    struct timeval tv;
    tv.tv_sec = 0;
    tv.tv_usec = 0;
    int returnCode = ::select(socket+1, &rd, NULL, NULL, &tv);
    if(returnCode == -1) {
        throw IOException(__FILE__, __LINE__, SocketError::getErrorString().c_str() );
    }
    return (returnCode == 0)? 0 : 1;

#else

    return 0;

#endif /* HAVE_SELECT */

#endif // !defined(HAVE_WINSOCK2_H)
}

////////////////////////////////////////////////////////////////////////////////
unsigned char SocketInputStream::read() throw (IOException) {

    unsigned char c;
    std::size_t len = read( &c, 1);
    if (len != sizeof(c) && !closed) {
        throw IOException( __FILE__, __LINE__,
                "activemq::io::SocketInputStream::read - failed reading a byte");
    }

    return c;
}

////////////////////////////////////////////////////////////////////////////////
std::size_t SocketInputStream::read(unsigned char* buffer,
        std::size_t bufferSize) throw (IOException) {
    int len = 0;

    // 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 || closed ) {
            throw IOException( __FILE__, __LINE__,
                    "activemq::io::SocketInputStream::read - The connection is broken" );
        }

    }while( !closed && len == -1 &&
            SocketError::getErrorCode() == SocketError::INTERRUPTED );

    // Check for error.
    if( len == -1 ) {

        // Otherwise, this was a bad error - throw an exception.
        throw IOException( __FILE__, __LINE__,
                "activemq::io::SocketInputStream::read - %s", SocketError::getErrorString().c_str() );
    }

    return len;
}

////////////////////////////////////////////////////////////////////////////////
std::size_t SocketInputStream::skip( std::size_t num AMQCPP_UNUSED )
throw ( io::IOException, exceptions::UnsupportedOperationException ) {

    throw exceptions::UnsupportedOperationException(
    __FILE__, __LINE__,
    "SocketInputStream::skip - skip() method is not supported");
}