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 2009/11/11 17:01:48 UTC
svn commit: r834931 [1/3] - in
/activemq/activemq-cpp/trunk/activemq-cpp/src: main/
main/activemq/commands/ main/activemq/core/ main/activemq/io/ main/cms/
main/decaf/internal/io/ main/decaf/io/ main/decaf/lang/ main/decaf/net/
test/ test/activemq/comm...
Author: tabish
Date: Wed Nov 11 16:01:47 2009
New Revision: 834931
URL: http://svn.apache.org/viewvc?rev=834931&view=rev
Log:
https://issues.apache.org/activemq/browse/AMQCPP-265
Proper implementation of the StreamMessage. Update the CMS StreamMessage API to reflect the exceptions that can be thrown from the StreamMessage command. Add new unit tests to validate that the message behaves as expected.
Modified:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMapMessage.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQStreamMessage.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQStreamMessage.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQTextMessage.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/Message.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSession.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/io/LoggingInputStream.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/io/LoggingInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/cms/StreamMessage.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/io/StandardInputStream.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/io/StandardInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/BlockingByteArrayInputStream.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/BlockingByteArrayInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/BufferedInputStream.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/BufferedInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/ByteArrayInputStream.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/ByteArrayInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/ByteArrayOutputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/DataInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/FilterInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/io/InputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/lang/Byte.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/net/SocketInputStream.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/net/SocketInputStream.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/commands/ActiveMQStreamMessageTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/commands/ActiveMQStreamMessageTest.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/io/BufferedInputStreamTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/io/FilterInputStreamTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/testRegistry.cpp
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am?rev=834931&r1=834930&r2=834931&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am Wed Nov 11 16:01:47 2009
@@ -130,6 +130,7 @@
activemq/transport/tcp/TcpTransport.cpp \
activemq/transport/tcp/TcpTransportFactory.cpp \
activemq/util/ActiveMQProperties.cpp \
+ activemq/util/CMSExceptionSupport.cpp \
activemq/util/CompositeData.cpp \
activemq/util/LongSequenceGenerator.cpp \
activemq/util/MemoryUsage.cpp \
@@ -704,6 +705,7 @@
activemq/transport/tcp/TcpTransport.h \
activemq/transport/tcp/TcpTransportFactory.h \
activemq/util/ActiveMQProperties.h \
+ activemq/util/CMSExceptionSupport.h \
activemq/util/CompositeData.h \
activemq/util/Config.h \
activemq/util/LongSequenceGenerator.h \
@@ -1104,6 +1106,7 @@
decaf/internal/util/concurrent/ConditionImpl.h \
decaf/internal/util/concurrent/MutexImpl.h \
decaf/internal/util/concurrent/SynchronizableImpl.h \
+ decaf/internal/util/concurrent/Transferer.h \
decaf/internal/util/concurrent/unix/ConditionHandle.h \
decaf/internal/util/concurrent/unix/MutexHandle.h \
decaf/internal/util/concurrent/windows/ConditionHandle.h \
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.cpp?rev=834931&r1=834930&r2=834931&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.cpp Wed Nov 11 16:01:47 2009
@@ -66,16 +66,6 @@
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQBytesMessage::checkWriteOnlyBody() const throw ( cms::CMSException ){
- if( !this->isReadOnlyBody() ){
- throw exceptions::ActiveMQException(
- __FILE__, __LINE__,
- "message is in write-only mode and "
- "cannot be read from" ).convertToCMSException();
- }
-}
-
-////////////////////////////////////////////////////////////////////////////////
void ActiveMQBytesMessage::setBodyBytes( const unsigned char* buffer,
std::size_t numBytes )
throw( cms::CMSException ) {
@@ -126,7 +116,7 @@
bool ActiveMQBytesMessage::readBoolean() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readBoolean();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -136,7 +126,7 @@
void ActiveMQBytesMessage::writeBoolean( bool value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeBoolean( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -146,7 +136,7 @@
unsigned char ActiveMQBytesMessage::readByte() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readByte();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -156,7 +146,7 @@
void ActiveMQBytesMessage::writeByte( unsigned char value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeByte( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -167,7 +157,7 @@
throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.read( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -177,7 +167,7 @@
void ActiveMQBytesMessage::writeBytes( const std::vector<unsigned char>& value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.write( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -188,7 +178,7 @@
throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.read( buffer, 0, length );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -200,7 +190,7 @@
std::size_t length ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.write( value, offset, length );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -210,7 +200,7 @@
char ActiveMQBytesMessage::readChar() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readChar();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -220,7 +210,7 @@
void ActiveMQBytesMessage::writeChar( char value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeChar( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -230,7 +220,7 @@
float ActiveMQBytesMessage::readFloat() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readFloat();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -240,7 +230,7 @@
void ActiveMQBytesMessage::writeFloat( float value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeFloat( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -250,7 +240,7 @@
double ActiveMQBytesMessage::readDouble() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readDouble();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -260,7 +250,7 @@
void ActiveMQBytesMessage::writeDouble( double value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeDouble( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -270,7 +260,7 @@
short ActiveMQBytesMessage::readShort() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readShort();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -280,7 +270,7 @@
void ActiveMQBytesMessage::writeShort( short value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeShort( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -290,7 +280,7 @@
unsigned short ActiveMQBytesMessage::readUnsignedShort() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readUnsignedShort();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -300,7 +290,7 @@
void ActiveMQBytesMessage::writeUnsignedShort( unsigned short value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeUnsignedShort( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -310,7 +300,7 @@
int ActiveMQBytesMessage::readInt() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readInt();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -320,7 +310,7 @@
void ActiveMQBytesMessage::writeInt( int value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeInt( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -330,7 +320,7 @@
long long ActiveMQBytesMessage::readLong() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readLong();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -340,7 +330,7 @@
void ActiveMQBytesMessage::writeLong( long long value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeLong( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -350,7 +340,7 @@
std::string ActiveMQBytesMessage::readString() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readString();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -360,7 +350,7 @@
void ActiveMQBytesMessage::writeString( const std::string& value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeChars( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -370,7 +360,7 @@
std::string ActiveMQBytesMessage::readUTF() const throw ( cms::CMSException ) {
try{
- checkWriteOnlyBody();
+ failIfWriteOnlyBody();
return dataInputStream.readUTF();
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -380,7 +370,7 @@
void ActiveMQBytesMessage::writeUTF( const std::string& value ) throw ( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
dataOutputStream.writeUTF( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.h?rev=834931&r1=834930&r2=834931&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQBytesMessage.h Wed Nov 11 16:01:47 2009
@@ -393,14 +393,6 @@
*/
virtual void writeUTF( const std::string& value ) throw ( cms::CMSException );
- protected:
-
- /**
- * Throws an exception if in write-only mode.
- * @throws CMSException.
- */
- void checkWriteOnlyBody() const throw ( cms::CMSException );
-
};
}}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMapMessage.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMapMessage.cpp?rev=834931&r1=834930&r2=834931&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMapMessage.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMapMessage.cpp Wed Nov 11 16:01:47 2009
@@ -186,7 +186,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setBool( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -208,7 +208,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setByte( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -230,7 +230,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setByteArray( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -251,7 +251,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setChar( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -273,7 +273,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setDouble( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -294,7 +294,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setFloat( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -315,7 +315,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setInt( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -336,7 +336,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setLong( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -357,7 +357,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setShort( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -379,7 +379,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyBody();
+ failIfReadOnlyBody();
getMap().setString( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h?rev=834931&r1=834930&r2=834931&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQMessageTemplate.h Wed Nov 11 16:01:47 2009
@@ -25,6 +25,9 @@
#include <activemq/wireformat/openwire/utils/MessagePropertyInterceptor.h>
#include <activemq/wireformat/openwire/marshal/BaseDataStreamMarshaller.h>
+#include <cms/MessageNotReadableException.h>
+#include <cms/MessageNotWriteableException.h>
+
namespace activemq {
namespace commands {
@@ -224,7 +227,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setBooleanProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -241,7 +244,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setByteProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -258,7 +261,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setDoubleProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -275,7 +278,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setFloatProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -292,7 +295,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setIntProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -309,7 +312,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setLongProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -326,7 +329,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setShortProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -343,7 +346,7 @@
throw( cms::CMSException ) {
try{
- checkReadOnlyProperties();
+ failIfReadOnlyProperties();
this->propertiesInterceptor->setStringProperty( name, value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
@@ -548,17 +551,29 @@
protected:
- void checkReadOnlyBody() {
+ virtual void onSend() {
+ this->setReadOnlyBody(true);
+ this->setReadOnlyProperties(true);
+ }
+
+ void failIfWriteOnlyBody() const {
+ if( !this->isReadOnlyBody() ){
+ throw cms::MessageNotReadableException(
+ "message is in write-only mode and cannot be read from", NULL );
+ }
+ }
+
+ void failIfReadOnlyBody() const {
if( this->isReadOnlyBody() ) {
- throw exceptions::ActiveMQException(
- __FILE__, __LINE__, "Message Body is Read-Only." );
+ throw cms::MessageNotWriteableException(
+ "Message Body is Read-Only.", NULL );
}
}
- void checkReadOnlyProperties() {
+ void failIfReadOnlyProperties() const {
if( this->isReadOnlyProperties() ) {
- throw exceptions::ActiveMQException(
- __FILE__, __LINE__, "Message Properties are Read-Only." );
+ throw cms::MessageNotWriteableException(
+ "Message Properties are Read-Only.", NULL );
}
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQStreamMessage.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQStreamMessage.cpp?rev=834931&r1=834930&r2=834931&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQStreamMessage.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/commands/ActiveMQStreamMessage.cpp Wed Nov 11 16:01:47 2009
@@ -15,20 +15,38 @@
* limitations under the License.
*/
#include <activemq/commands/ActiveMQStreamMessage.h>
-#include <activemq/wireformat/openwire/marshal/PrimitiveTypesMarshaller.h>
+#include <activemq/wireformat/openwire/utils/OpenwireStringSupport.h>
+#include <activemq/util/PrimitiveValueNode.h>
+#include <activemq/util/CMSExceptionSupport.h>
+
+#include <cms/MessageEOFException.h>
+#include <cms/MessageFormatException.h>
+#include <cms/MessageNotReadableException.h>
+#include <cms/MessageNotWriteableException.h>
#include <algorithm>
#include <decaf/lang/Math.h>
+#include <decaf/lang/exceptions/NullPointerException.h>
+#include <decaf/lang/Character.h>
+#include <decaf/lang/Boolean.h>
+#include <decaf/lang/Byte.h>
+#include <decaf/lang/Short.h>
+#include <decaf/lang/Integer.h>
+#include <decaf/lang/Long.h>
+#include <decaf/lang/Double.h>
+#include <decaf/lang/Float.h>
+#include <decaf/io/ByteArrayInputStream.h>
using namespace std;
+using namespace cms;
using namespace activemq;
using namespace activemq::util;
using namespace activemq::commands;
using namespace activemq::exceptions;
using namespace activemq::wireformat;
using namespace activemq::wireformat::openwire;
-using namespace activemq::wireformat::openwire::marshal;
+using namespace activemq::wireformat::openwire::utils;
using namespace decaf;
using namespace decaf::io;
using namespace decaf::lang;
@@ -36,13 +54,14 @@
////////////////////////////////////////////////////////////////////////////////
ActiveMQStreamMessage::ActiveMQStreamMessage() :
- ActiveMQMessageTemplate< cms::StreamMessage >(), currentPos( 0 ) {
+ ActiveMQMessageTemplate< cms::StreamMessage >() {
this->clearBody();
}
////////////////////////////////////////////////////////////////////////////////
ActiveMQStreamMessage::~ActiveMQStreamMessage() {
+ this->reset();
}
////////////////////////////////////////////////////////////////////////////////
@@ -59,14 +78,24 @@
////////////////////////////////////////////////////////////////////////////////
void ActiveMQStreamMessage::copyDataStructure( const DataStructure* src ) {
- ActiveMQMessageTemplate<cms::StreamMessage>::copyDataStructure( src );
- const ActiveMQStreamMessage* srcStream =
- dynamic_cast< const ActiveMQStreamMessage* >( src );
+ // Protect against invalid self assignment.
+ if( this == src ) {
+ return;
+ }
+
+ const ActiveMQStreamMessage* srcPtr = dynamic_cast<const ActiveMQStreamMessage*>( src );
- if( srcStream != NULL && srcStream->list.get() != NULL ) {
- this->list.reset( new util::PrimitiveList( *srcStream->list ) );
+ if( srcPtr == NULL || src == NULL ) {
+ throw decaf::lang::exceptions::NullPointerException(
+ __FILE__, __LINE__,
+ "ActiveMQStreamMessage::copyDataStructure - src is NULL or invalid" );
}
+
+ ActiveMQStreamMessage* nonConstSrc = const_cast<ActiveMQStreamMessage*>( srcPtr );
+ nonConstSrc->storeContent();
+
+ ActiveMQMessageTemplate<cms::StreamMessage>::copyDataStructure( src );
}
////////////////////////////////////////////////////////////////////////////////
@@ -86,364 +115,817 @@
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::checkWriteOnlyBody() const throw ( cms::CMSException ){
- if( !this->isReadOnlyBody() ){
- throw exceptions::ActiveMQException(
- __FILE__, __LINE__,
- "message is in write-only mode and "
- "cannot be read from" ).convertToCMSException();
- }
+void ActiveMQStreamMessage::clearBody() throw( cms::CMSException ) {
+
+ // Invoke base class's version.
+ ActiveMQMessageTemplate<cms::StreamMessage>::clearBody();
+
+ this->dataIn.reset(NULL);
+ this->dataOut.reset(NULL);
+ this->remainingBytes = -1;
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::beforeMarshal( WireFormat* wireFormat )
- throw ( decaf::io::IOException ) {
+void ActiveMQStreamMessage::onSend() {
- try{
+ this->storeContent();
+ ActiveMQMessageTemplate<cms::StreamMessage>::onSend();
+}
- // Let the base class do its thing.
- ActiveMQMessageTemplate<cms::StreamMessage>::beforeMarshal( wireFormat );
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQStreamMessage::reset() throw ( cms::CMSException ) {
- if( this->list.get() != NULL && !this->list->isEmpty() ) {
- // Marshal as Content.
- PrimitiveTypesMarshaller::marshal( this->list.get(), getContent() );
- } else {
- clearBody();
- }
+ try{
+ storeContent();
+ this->bytesOut.reset(NULL);
+ this->dataIn.reset(NULL);
+ this->dataOut.reset(NULL);
+ this->remainingBytes = -1;
+ this->setReadOnlyBody(true);
}
- AMQ_CATCH_RETHROW( IOException )
- AMQ_CATCH_EXCEPTION_CONVERT( Exception, IOException )
- AMQ_CATCHALL_THROW( IOException )
+ AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-PrimitiveList& ActiveMQStreamMessage::getList() throw ( NullPointerException ) {
+bool ActiveMQStreamMessage::readBoolean() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
try{
- this->checkListIsUnmarshalled();
- return *( this->list );
+ initializeReading();
+
+ this->dataIn->mark(10);
+
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "Reached the end of the Stream", NULL );
+ }
+ if( type == PrimitiveValueNode::BOOLEAN_TYPE ) {
+ return this->dataIn->readBoolean();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Boolean::valueOf( this->dataIn->readUTF() ).booleanValue();
+ }
+
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to boolean." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( "not a boolean type", NULL );
+ }
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_RETHROW( NullPointerException )
- AMQ_CATCH_EXCEPTION_CONVERT( Exception, NullPointerException )
- AMQ_CATCHALL_THROW( NullPointerException )
}
////////////////////////////////////////////////////////////////////////////////
-const PrimitiveList& ActiveMQStreamMessage::getList() const
- throw ( decaf::lang::exceptions::NullPointerException ) {
+void ActiveMQStreamMessage::writeBoolean( bool value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
-
- this->checkListIsUnmarshalled();
- return *( this->list );
+ this->dataOut->write( PrimitiveValueNode::BOOLEAN_TYPE );
+ this->dataOut->writeBoolean( value );
}
- AMQ_CATCH_RETHROW( NullPointerException )
- AMQ_CATCH_EXCEPTION_CONVERT( Exception, NullPointerException )
- AMQ_CATCHALL_THROW( NullPointerException )
+ AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::checkListIsUnmarshalled() const
- throw ( decaf::lang::exceptions::NullPointerException ) {
+unsigned char ActiveMQStreamMessage::readByte() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
+ initializeReading();
try {
- if( this->list.get() == NULL ) {
+ this->dataIn->mark(10);
+ int type = this->dataIn->read();
- this->list.reset( new PrimitiveList() );
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::BYTE_TYPE ) {
+ return this->dataIn->readByte();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Byte::valueOf( this->dataIn->readUTF() ).byteValue();
+ }
- if( getContent().size() != 0 ){
- PrimitiveTypesMarshaller::unmarshal( this->list.get(), getContent() );
- }
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to byte." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a byte type", NULL );
+ }
- if( this->list.get() == NULL ) {
- throw NullPointerException(
- __FILE__, __LINE__,
- "ActiveMQStreamMessage::getMap() - All attempts to create a "
- "list have failed." );
- }
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::create(e);
}
+
+ throw CMSExceptionSupport::createMessageFormatException( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_RETHROW( NullPointerException )
- AMQ_CATCH_EXCEPTION_CONVERT( Exception, NullPointerException )
- AMQ_CATCHALL_THROW( NullPointerException )
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::reset() throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeByte( unsigned char value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- this->setReadOnlyBody( true );
- this->currentPos = 0;
+ this->dataOut->write( PrimitiveValueNode::BYTE_TYPE );
+ this->dataOut->writeByte( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-bool ActiveMQStreamMessage::readBoolean() const throw ( cms::CMSException ) {
+std::size_t ActiveMQStreamMessage::readBytes( std::vector<unsigned char>& value ) const
+ throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkWriteOnlyBody();
- return this->getList().getBool( this->currentPos++ );
- }
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
+ return this->readBytes( &value[0], value.size() );
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeBoolean( bool value ) throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeBytes( const std::vector<unsigned char>& value )
+ throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkReadOnlyBody();
- this->getList().add( value );
+ this->dataOut->write( PrimitiveValueNode::BYTE_ARRAY_TYPE );
+ this->dataOut->writeInt( value.size() );
+ this->dataOut->write( &value[0], 0, value.size() );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-unsigned char ActiveMQStreamMessage::readByte() const throw ( cms::CMSException ) {
+std::size_t ActiveMQStreamMessage::readBytes( unsigned char* buffer, std::size_t length ) const
+ throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkWriteOnlyBody();
- return this->getList().getByte( this->currentPos++ );
+ initializeReading();
+ try {
+
+ if( buffer == NULL ) {
+ throw NullPointerException( __FILE__, __LINE__, "Passed buffer was NULL" );
+ }
+
+ if( this->remainingBytes == -1 ) {
+
+ this->dataIn->mark( length + 1 );
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+
+ if( type != PrimitiveValueNode::BYTE_ARRAY_TYPE ) {
+ throw MessageFormatException( "Not a byte array", NULL );
+ }
+
+ this->remainingBytes = this->dataIn->readInt();
+
+ } else if( this->remainingBytes == 0 ) {
+ remainingBytes = -1;
+ return -1;
+ }
+
+ if( length <= (size_t)this->remainingBytes ) {
+ // small buffer
+ this->remainingBytes -= length;
+ this->dataIn->readFully( buffer, 0, length );
+ return length;
+ } else {
+ // big buffer
+ int rc = this->dataIn->read( buffer, 0, this->remainingBytes );
+ this->remainingBytes = 0;
+ return rc;
+ }
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeByte( unsigned char value ) throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeBytes( const unsigned char* value,
+ std::size_t offset,
+ std::size_t length ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkReadOnlyBody();
- this->getList().add( value );
+ this->dataOut->write( PrimitiveValueNode::BYTE_ARRAY_TYPE );
+ this->dataOut->writeInt( length );
+ this->dataOut->write( value, offset, length );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-std::size_t ActiveMQStreamMessage::readBytes( std::vector<unsigned char>& value ) const
- throw ( cms::CMSException ) {
+char ActiveMQStreamMessage::readChar() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkWriteOnlyBody();
- value = this->getList().getByteArray( this->currentPos++ );
- return value.size();
+ initializeReading();
+ try {
+
+ this->dataIn->mark( 17 );
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::CHAR_TYPE ) {
+ return this->dataIn->readChar();
+ }
+
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to char." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a char type", NULL );
+ }
+
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& ioe ) {
+ throw CMSExceptionSupport::create( ioe );
+ }
+
+ throw CMSExceptionSupport::create( ex );;
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeBytes( const std::vector<unsigned char>& value )
- throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeChar( char value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkReadOnlyBody();
- this->getList().add( value );
+ this->dataOut->write( PrimitiveValueNode::CHAR_TYPE );
+ this->dataOut->writeChar( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-std::size_t ActiveMQStreamMessage::readBytes( unsigned char*& buffer, std::size_t length ) const
- throw ( cms::CMSException ) {
+float ActiveMQStreamMessage::readFloat() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
+ initializeReading();
+ try {
- if( buffer == NULL ) {
- throw cms::CMSException( "Passed Buffer was NULL", NULL );
+ this->dataIn->mark( 33 );
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::FLOAT_TYPE ) {
+ return this->dataIn->readFloat();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Float::valueOf( this->dataIn->readUTF() ).floatValue();
}
- checkWriteOnlyBody();
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to float." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a float type", NULL );
+ }
- std::vector<unsigned char> temp = this->getList().getByteArray( this->currentPos++ );
- std::size_t amount = Math::min( (unsigned int)length, (unsigned int)temp.size() );
+ } catch( NumberFormatException& ex ) {
- for( std::size_t i = 0; i < amount; ++i ) {
- buffer[i] = temp.at( i );
+ try {
+ this->dataIn->reset();
+ } catch( IOException& ioe ) {
+ throw CMSExceptionSupport::create( ioe );
}
- return amount;
+ throw CMSExceptionSupport::create( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeBytes( const unsigned char* value,
- std::size_t offset,
- std::size_t length ) throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeFloat( float value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
-
- checkReadOnlyBody();
-
- std::vector<unsigned char> temp;
- for( std::size_t i = offset; i < length; ++i ) {
- temp.push_back( value[i] );
- }
-
- this->getList().add( temp );
+ this->dataOut->write( PrimitiveValueNode::FLOAT_TYPE );
+ this->dataOut->writeFloat( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-char ActiveMQStreamMessage::readChar() const throw ( cms::CMSException ) {
+double ActiveMQStreamMessage::readDouble() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkWriteOnlyBody();
- return this->getList().getChar( this->currentPos++ );
+ initializeReading();
+ try {
+
+ this->dataIn->mark( 33 );
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::DOUBLE_TYPE ) {
+ return this->dataIn->readDouble();
+ }
+ if( type == PrimitiveValueNode::FLOAT_TYPE ) {
+ return this->dataIn->readFloat();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Double::valueOf( this->dataIn->readUTF() ).doubleValue();
+ }
+
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to double." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a double type", NULL );
+ }
+
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& ioe ) {
+ throw CMSExceptionSupport::create( ioe );
+ }
+
+ throw CMSExceptionSupport::create( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeChar( char value ) throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeDouble( double value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkReadOnlyBody();
- this->getList().add( value );
+ this->dataOut->write( PrimitiveValueNode::DOUBLE_TYPE );
+ this->dataOut->writeDouble( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-float ActiveMQStreamMessage::readFloat() const throw ( cms::CMSException ) {
+short ActiveMQStreamMessage::readShort() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkWriteOnlyBody();
- return this->getList().getFloat( this->currentPos++ );
- }
- AMQ_CATCH_RETHROW( ActiveMQException )
- AMQ_CATCH_EXCEPTION_CONVERT( Exception, ActiveMQException )
- AMQ_CATCHALL_THROW( ActiveMQException )
-}
+ initializeReading();
+ try {
-////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeFloat( float value ) throw ( cms::CMSException ) {
+ this->dataIn->mark(17);
+ int type = this->dataIn->read();
- try{
- checkReadOnlyBody();
- this->getList().add( value );
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::SHORT_TYPE ) {
+ return this->dataIn->readShort();
+ }
+ if( type == PrimitiveValueNode::BYTE_TYPE ) {
+ return this->dataIn->readByte();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Short::valueOf( this->dataIn->readUTF() ).shortValue();
+ }
+
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to short." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a short type", NULL );
+ }
+
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::create(e);
+ }
+
+ throw CMSExceptionSupport::create( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-double ActiveMQStreamMessage::readDouble() const throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeShort( short value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkWriteOnlyBody();
- return this->getList().getDouble( this->currentPos++ );
+ this->dataOut->write( PrimitiveValueNode::SHORT_TYPE );
+ this->dataOut->writeShort( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeDouble( double value ) throw ( cms::CMSException ) {
+unsigned short ActiveMQStreamMessage::readUnsignedShort() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkReadOnlyBody();
- this->getList().add( value );
+ initializeReading();
+ try {
+
+ this->dataIn->mark(17);
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::SHORT_TYPE ) {
+ return this->dataIn->readUnsignedShort();
+ }
+ if( type == PrimitiveValueNode::BYTE_TYPE ) {
+ return this->dataIn->readByte();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Short::valueOf( this->dataIn->readUTF() ).shortValue();
+ }
+
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to short." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a short type", NULL );
+ }
+
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::create(e);
+ }
+
+ throw CMSExceptionSupport::create( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-short ActiveMQStreamMessage::readShort() const throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeUnsignedShort( unsigned short value )
+ throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkWriteOnlyBody();
- return this->getList().getShort( this->currentPos++ );
+ this->dataOut->write( PrimitiveValueNode::SHORT_TYPE );
+ this->dataOut->writeUnsignedShort( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeShort( short value ) throw ( cms::CMSException ) {
+int ActiveMQStreamMessage::readInt() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkReadOnlyBody();
- this->getList().add( value );
+ initializeReading();
+ try {
+
+ this->dataIn->mark(33);
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::INTEGER_TYPE ) {
+ return this->dataIn->readInt();
+ }
+ if( type == PrimitiveValueNode::SHORT_TYPE ) {
+ return this->dataIn->readShort();
+ }
+ if( type == PrimitiveValueNode::BYTE_TYPE ) {
+ return this->dataIn->readByte();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Integer::valueOf( this->dataIn->readUTF() ).intValue();
+ }
+
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to int." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a int type", NULL );
+ }
+
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::create(e);
+ }
+
+ throw CMSExceptionSupport::create( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-unsigned short ActiveMQStreamMessage::readUnsignedShort() const throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeInt( int value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkWriteOnlyBody();
- return (unsigned short)this->getList().getShort( this->currentPos++ );
+ this->dataOut->write( PrimitiveValueNode::INTEGER_TYPE );
+ this->dataOut->writeInt( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeUnsignedShort( unsigned short value ) throw ( cms::CMSException ) {
+long long ActiveMQStreamMessage::readLong() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkReadOnlyBody();
- this->getList().add( (short)value );
+ initializeReading();
+ try {
+
+ this->dataIn->mark(65);
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::LONG_TYPE ) {
+ return this->dataIn->readLong();
+ }
+ if( type == PrimitiveValueNode::INTEGER_TYPE ) {
+ return this->dataIn->readInt();
+ }
+ if( type == PrimitiveValueNode::SHORT_TYPE ) {
+ return this->dataIn->readShort();
+ }
+ if( type == PrimitiveValueNode::BYTE_TYPE ) {
+ return this->dataIn->readByte();
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return Long::valueOf( this->dataIn->readUTF() ).longValue();
+ }
+
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ this->dataIn->reset();
+ throw NullPointerException( __FILE__, __LINE__, "Cannot convert NULL value to long." );
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a long type", NULL );
+ }
+
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::create(e);
+ }
+
+ throw CMSExceptionSupport::create( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-int ActiveMQStreamMessage::readInt() const throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeLong( long long value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkWriteOnlyBody();
- return this->getList().getInt( this->currentPos++ );
+ this->dataOut->write( PrimitiveValueNode::LONG_TYPE );
+ this->dataOut->writeLong( value );
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeInt( int value ) throw ( cms::CMSException ) {
+std::string ActiveMQStreamMessage::readString() const throw ( cms::MessageEOFException,
+ cms::MessageFormatException,
+ cms::MessageNotReadableException,
+ cms::CMSException ) {
- try{
- checkReadOnlyBody();
- this->getList().add( value );
+ initializeReading();
+ try {
+
+ this->dataIn->mark( 65 );
+ int type = this->dataIn->read();
+
+ if( type == -1 ) {
+ throw MessageEOFException( "reached end of data", NULL );
+ }
+ if( type == PrimitiveValueNode::NULL_TYPE ) {
+ return "";
+ }
+ if( type == PrimitiveValueNode::BIG_STRING_TYPE ) {
+ return OpenwireStringSupport::readString( *dataIn );
+ }
+ if( type == PrimitiveValueNode::STRING_TYPE ) {
+ return this->dataIn->readUTF();
+ }
+ if( type == PrimitiveValueNode::LONG_TYPE ) {
+ return Long( this->dataIn->readLong() ).toString();
+ }
+ if( type == PrimitiveValueNode::INTEGER_TYPE ) {
+ return Integer( this->dataIn->readInt() ).toString();
+ }
+ if( type == PrimitiveValueNode::SHORT_TYPE ) {
+ return Short( this->dataIn->readShort() ).toString();
+ }
+ if( type == PrimitiveValueNode::BYTE_TYPE ) {
+ return Byte( this->dataIn->readByte() ).toString();
+ }
+ if( type == PrimitiveValueNode::FLOAT_TYPE ) {
+ return Float( this->dataIn->readFloat() ).toString();
+ }
+ if( type == PrimitiveValueNode::DOUBLE_TYPE ) {
+ return Double( this->dataIn->readDouble() ).toString();
+ }
+ if( type == PrimitiveValueNode::BOOLEAN_TYPE ) {
+ return ( this->dataIn->readBoolean() ? Boolean::_TRUE : Boolean::_FALSE ).toString();
+ }
+
+ if( type == PrimitiveValueNode::CHAR_TYPE ) {
+ return Character( this->dataIn->readChar() ).toString();
+ } else {
+ this->dataIn->reset();
+ throw MessageFormatException( " not a String type", NULL );
+ }
+
+ } catch( NumberFormatException& ex ) {
+
+ try {
+ this->dataIn->reset();
+ } catch( IOException& ioe ) {
+ throw CMSExceptionSupport::create( ioe );
+ }
+
+ throw CMSExceptionSupport::create( ex );
+
+ } catch( EOFException& e ) {
+ throw CMSExceptionSupport::createMessageEOFException( e );
+ } catch( IOException& e ) {
+ throw CMSExceptionSupport::createMessageFormatException( e );
+ } catch( Exception& e ) {
+ throw CMSExceptionSupport::create( e );
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-long long ActiveMQStreamMessage::readLong() const throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::writeString( const std::string& value ) throw ( cms::MessageNotWriteableException,
+ cms::CMSException ) {
+ initializeWriting();
try{
- checkWriteOnlyBody();
- return this->getList().getLong( this->currentPos++ );
+ if( value.size() <= Short::MAX_VALUE / 4 ) {
+ this->dataOut->write( PrimitiveValueNode::STRING_TYPE );
+ this->dataOut->writeUTF( value );
+ } else {
+ this->dataOut->write( PrimitiveValueNode::BIG_STRING_TYPE );
+ OpenwireStringSupport::writeString( *(this->dataOut), &value );
+ }
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeLong( long long value ) throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::storeContent() {
- try{
- checkReadOnlyBody();
- this->getList().add( value );
+ if( this->dataOut.get() != NULL ) {
+
+ this->dataOut->close();
+ this->setContent( this->bytesOut->toByteArrayRef() );
+ this->dataOut.reset(NULL);
+ this->bytesOut.reset(NULL);
}
- AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-std::string ActiveMQStreamMessage::readString() const throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::initializeReading() const {
- try{
- checkWriteOnlyBody();
- return this->getList().getString( this->currentPos++ );
+ this->failIfWriteOnlyBody();
+ try {
+ if( this->dataIn.get() == NULL) {
+ ByteArrayInputStream* is = new ByteArrayInputStream( this->getContent() );
+ this->dataIn.reset( new DataInputStream( is, true ) );
+ }
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}
////////////////////////////////////////////////////////////////////////////////
-void ActiveMQStreamMessage::writeString( const std::string& value ) throw ( cms::CMSException ) {
+void ActiveMQStreamMessage::initializeWriting() {
+ this->failIfReadOnlyBody();
try{
- checkReadOnlyBody();
- this->getList().add( value );
+ if( this->dataOut.get() == NULL ) {
+ this->bytesOut.reset( new ByteArrayOutputStream() );
+ this->dataOut.reset( new DataOutputStream( this->bytesOut.get() ) );
+ }
}
AMQ_CATCH_ALL_THROW_CMSEXCEPTION()
}