text: /**
   * 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/wireformat/openwire/marshal/BaseDataStreamMarshaller.h>
  #include <activemq/wireformat/openwire/utils/HexTable.h>
  #include <activemq/commands/MessageId.h>
  #include <activemq/commands/ProducerId.h>
  #include <activemq/commands/TransactionId.h>
  #include <activemq/commands/LocalTransactionId.h>
  #include <activemq/commands/XATransactionId.h>
  #include <activemq/commands/BrokerError.h>
  #include <activemq/exceptions/ActiveMQException.h>
  #include <decaf/lang/Long.h>
  #include <decaf/lang/Integer.h>
  #include <decaf/lang/Pointer.h>
  #include <activemq/util/Config.h>
  
  using namespace std;
  using namespace activemq;
  using namespace activemq::util;
  using namespace activemq::exceptions;
  using namespace activemq::commands;
  using namespace activemq::wireformat;
  using namespace activemq::wireformat::openwire;
  using namespace activemq::wireformat::openwire::marshal;
  using namespace activemq::wireformat::openwire::utils;
  using namespace decaf::io;
  using namespace decaf::util;
  using namespace decaf::lang;
  
  ////////////////////////////////////////////////////////////////////////////////
  utils::HexTable BaseDataStreamMarshaller::hexTable;
  
  ////////////////////////////////////////////////////////////////////////////////
  commands::DataStructure* BaseDataStreamMarshaller::tightUnmarshalCachedObject(OpenWireFormat* wireFormat, decaf::io::DataInputStream* dataIn,utils::BooleanStream* bs) {
      try {
          return wireFormat->tightUnmarshalNestedObject(dataIn, bs);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  int BaseDataStreamMarshaller::tightMarshalCachedObject1(OpenWireFormat* wireFormat, commands::DataStructure* data, utils::BooleanStream* bs) {
      try {
          return wireFormat->tightMarshalNestedObject1(data, bs);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::tightMarshalCachedObject2(OpenWireFormat* wireFormat, commands::DataStructure* data, decaf::io::DataOutputStream* dataOut,utils::BooleanStream* bs) {
      try {
          wireFormat->tightMarshalNestedObject2(data, dataOut, bs);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::looseMarshalCachedObject(OpenWireFormat* wireFormat, commands::DataStructure* data, decaf::io::DataOutputStream* dataOut) {
      try {
          wireFormat->looseMarshalNestedObject(data, dataOut);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  commands::DataStructure* BaseDataStreamMarshaller::looseUnmarshalCachedObject(OpenWireFormat* wireFormat, decaf::io::DataInputStream* dataIn) {
      try {
          return wireFormat->looseUnmarshalNestedObject(dataIn);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  int BaseDataStreamMarshaller::tightMarshalNestedObject1(OpenWireFormat* wireFormat, commands::DataStructure* object, utils::BooleanStream* bs) {
      try {
          return wireFormat->tightMarshalNestedObject1(object, bs);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::tightMarshalNestedObject2(OpenWireFormat* wireFormat, commands::DataStructure* object, decaf::io::DataOutputStream* dataOut, utils::BooleanStream* bs) {
      try {
          wireFormat->tightMarshalNestedObject2(object, dataOut, bs);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  commands::DataStructure* BaseDataStreamMarshaller::tightUnmarshalNestedObject(OpenWireFormat* wireFormat, decaf::io::DataInputStream* dataIn,utils::BooleanStream* bs) {
      try {
          return wireFormat->tightUnmarshalNestedObject(dataIn, bs);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  commands::DataStructure* BaseDataStreamMarshaller::looseUnmarshalNestedObject(OpenWireFormat* wireFormat, decaf::io::DataInputStream* dataIn) {
  
      try {
          return wireFormat->looseUnmarshalNestedObject(dataIn);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::looseMarshalNestedObject(OpenWireFormat* wireFormat, commands::DataStructure* object, decaf::io::DataOutputStream* dataOut) {
  
      try {
          wireFormat->looseMarshalNestedObject(object, dataOut);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::string BaseDataStreamMarshaller::tightUnmarshalString(decaf::io::DataInputStream* dataIn, utils::BooleanStream* bs) {
  
      try {
  
          if (bs->readBoolean()) {
              if (bs->readBoolean()) {
                  return this->readAsciiString(dataIn);
              } else {
                  return dataIn->readUTF();
              }
          } else {
              return "";
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  int BaseDataStreamMarshaller::tightMarshalString1(const std::string& value, utils::BooleanStream* bs) {
      try {
  
          bs->writeBoolean(value != "");
          if (value != "") {
              size_t strlen = value.length();
  
              int utflen = 0;
              int c = 0;
              bool isOnlyAscii = true;
  
              for (size_t i = 0; i < strlen; ++i) {
  
                  c = value[i];
                  if ((c >= 0x0001) && (c <= 0x007F)) { // ASCII char
                      utflen++;
                  } else if (c > 0x07FF) {
                      utflen += 3;
                      isOnlyAscii = false;
                  } else {
                      isOnlyAscii = false;
                      utflen += 2;
                  }
              }
  
              if (utflen >= 0x10000) {
                  throw IOException(__FILE__, __LINE__, "BaseDataStreamMarshaller::tightMarshalString1 - "
                          "Encountered a String value that is too long to encode.");
              }
  
              bs->writeBoolean(isOnlyAscii);
  
              return utflen + 2;
          } else {
              return 0;
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(ActiveMQException, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::tightMarshalString2(const std::string& value, decaf::io::DataOutputStream* dataOut, utils::BooleanStream* bs) {
  
      try {
  
          if (bs->readBoolean()) {
              // If we verified it only holds ascii values
              if (bs->readBoolean()) {
                  dataOut->writeShort((short) value.length());
                  dataOut->writeBytes(value);
              } else {
                  dataOut->writeUTF(value);
              }
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::looseMarshalString(const std::string value, decaf::io::DataOutputStream* dataOut) {
  
      try {
          dataOut->writeBoolean(value != "");
          if (value != "") {
              dataOut->writeUTF(value);
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::string BaseDataStreamMarshaller::looseUnmarshalString(decaf::io::DataInputStream* dataIn) {
  
      try {
          if (dataIn->readBoolean()) {
              return dataIn->readUTF();
          } else {
              return "";
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  int BaseDataStreamMarshaller::tightMarshalLong1(OpenWireFormat* wireFormat AMQCPP_UNUSED, long long value, utils::BooleanStream* bs) {
  
      try {
  
          if (value == 0L) {
  
              bs->writeBoolean(false);
              bs->writeBoolean(false);
              return 0;
  
          } else {
  
              unsigned long long ul = (unsigned long long) value;
              if ((ul & 0xFFFFFFFFFFFF0000ULL) == 0ULL) {
                  bs->writeBoolean(false);
                  bs->writeBoolean(true);
                  return 2;
              } else if ((ul & 0xFFFFFFFF00000000ULL) == 0ULL) {
                  bs->writeBoolean(true);
                  bs->writeBoolean(false);
                  return 4;
              } else {
                  bs->writeBoolean(true);
                  bs->writeBoolean(true);
                  return 8;
              }
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::tightMarshalLong2(OpenWireFormat* wireFormat AMQCPP_UNUSED, long long value, decaf::io::DataOutputStream* dataOut,
          utils::BooleanStream* bs) {
      try {
  
          if (bs->readBoolean()) {
  
              if (bs->readBoolean()) {
                  dataOut->writeLong(value);
              } else {
                  dataOut->writeInt((int) value);
              }
  
          } else {
  
              if (bs->readBoolean()) {
                  dataOut->writeShort((short) value);
              }
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  long long BaseDataStreamMarshaller::tightUnmarshalLong(OpenWireFormat* wireFormat AMQCPP_UNUSED, decaf::io::DataInputStream* dataIn, utils::BooleanStream* bs) {
  
      try {
          if (bs->readBoolean()) {
  
              if (bs->readBoolean()) {
                  return dataIn->readLong();
              } else {
                  return (unsigned int) dataIn->readInt();
              }
  
          } else {
  
              if (bs->readBoolean()) {
                  return dataIn->readUnsignedShort();
              } else {
                  return 0;
              }
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::looseMarshalLong(OpenWireFormat* wireFormat AMQCPP_UNUSED, long long value, decaf::io::DataOutputStream* dataOut) {
  
      try {
          dataOut->writeLong(value);
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  long long BaseDataStreamMarshaller::looseUnmarshalLong(OpenWireFormat* wireFormat AMQCPP_UNUSED, decaf::io::DataInputStream* dataIn) {
  
      try {
          return dataIn->readLong();
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  commands::DataStructure* BaseDataStreamMarshaller::tightUnmarshalBrokerError(OpenWireFormat* wireFormat, decaf::io::DataInputStream* dataIn,utils::BooleanStream* bs) {
  
      try {
  
          if (bs->readBoolean()) {
  
              std::auto_ptr<BrokerError> answer(new BrokerError());
  
              answer->setExceptionClass(tightUnmarshalString(dataIn, bs));
              answer->setMessage(tightUnmarshalString(dataIn, bs));
  
              if (wireFormat->isStackTraceEnabled()) {
                  short length = dataIn->readShort();
                  std::vector<Pointer<BrokerError::StackTraceElement> > stackTrace;
  
                  for (int i = 0; i < length; ++i) {
  
                      Pointer<BrokerError::StackTraceElement> element(new BrokerError::StackTraceElement);
  
                      element->ClassName = tightUnmarshalString(dataIn, bs);
                      element->MethodName = tightUnmarshalString(dataIn, bs);
                      element->FileName = tightUnmarshalString(dataIn, bs);
                      element->LineNumber = dataIn->readInt();
                      stackTrace.push_back(element);
                  }
  
                  answer->setStackTraceElements(stackTrace);
                  answer->setCause(Pointer<BrokerError>(dynamic_cast<BrokerError*>(tightUnmarshalBrokerError(wireFormat, dataIn, bs))));
              }
  
              return answer.release();
  
          } else {
              return NULL;
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  int BaseDataStreamMarshaller::tightMarshalBrokerError1(OpenWireFormat* wireFormat, commands::DataStructure* data, utils::BooleanStream* bs) {
  
      try {
  
          BrokerError* error = dynamic_cast<BrokerError*>(data);
  
          if (error == NULL) {
  
              bs->writeBoolean(false);
              return 0;
  
          } else {
  
              int rc = 0;
              bs->writeBoolean(true);
              rc += tightMarshalString1(error->getExceptionClass(), bs);
              rc += tightMarshalString1(error->getMessage(), bs);
  
              if (wireFormat->isStackTraceEnabled()) {
  
                  rc += 2;
  
                  for (unsigned int i = 0; i < error->getStackTraceElements().size(); ++i) {
  
                      const Pointer<BrokerError::StackTraceElement> element = error->getStackTraceElements()[i];
                      rc += tightMarshalString1(element->ClassName, bs);
                      rc += tightMarshalString1(element->MethodName, bs);
                      rc += tightMarshalString1(element->FileName, bs);
                      rc += 4;
                  }
                  rc += tightMarshalBrokerError1(wireFormat, error->getCause().get(), bs);
              }
  
              return rc;
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::tightMarshalBrokerError2(OpenWireFormat* wireFormat, commands::DataStructure* data, decaf::io::DataOutputStream* dataOut,
          utils::BooleanStream* bs) {
  
      try {
  
          if (bs->readBoolean()) {
  
              BrokerError* error = dynamic_cast<BrokerError*>(data);
  
              tightMarshalString2(error->getExceptionClass(), dataOut, bs);
              tightMarshalString2(error->getMessage(), dataOut, bs);
  
              if (wireFormat->isStackTraceEnabled()) {
  
                  int length = (short) error->getStackTraceElements().size();
                  dataOut->writeShort((short) length);
  
                  for (int i = 0; i < length; ++i) {
  
                      Pointer<BrokerError::StackTraceElement> element = error->getStackTraceElements()[i];
  
                      tightMarshalString2(element->ClassName, dataOut, bs);
                      tightMarshalString2(element->MethodName, dataOut, bs);
                      tightMarshalString2(element->FileName, dataOut, bs);
                      dataOut->writeInt(element->LineNumber);
                  }
  
                  tightMarshalBrokerError2(wireFormat, error->getCause().get(), dataOut, bs);
              }
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  commands::DataStructure* BaseDataStreamMarshaller::looseUnmarshalBrokerError(OpenWireFormat* wireFormat, decaf::io::DataInputStream* dataIn) {
  
      try {
  
          if (dataIn->readBoolean()) {
  
              std::auto_ptr<BrokerError> answer(new BrokerError());
  
              answer->setExceptionClass(looseUnmarshalString(dataIn));
              answer->setMessage(looseUnmarshalString(dataIn));
  
              if (wireFormat->isStackTraceEnabled()) {
  
                  short length = dataIn->readShort();
                  std::vector<Pointer<BrokerError::StackTraceElement> > stackTrace;
  
                  for (int i = 0; i < length; ++i) {
  
                      Pointer<BrokerError::StackTraceElement> element(new BrokerError::StackTraceElement);
  
                      element->ClassName = looseUnmarshalString(dataIn);
                      element->MethodName = looseUnmarshalString(dataIn);
                      element->FileName = looseUnmarshalString(dataIn);
                      element->LineNumber = dataIn->readInt();
  
                      stackTrace.push_back(element);
                  }
                  answer->setStackTraceElements(stackTrace);
                  answer->setCause(Pointer<BrokerError>(dynamic_cast<BrokerError*>(looseUnmarshalBrokerError(wireFormat, dataIn))));
              }
  
              return answer.release();
          } else {
              return NULL;
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  void BaseDataStreamMarshaller::looseMarshalBrokerError(OpenWireFormat* wireFormat, commands::DataStructure* data, decaf::io::DataOutputStream* dataOut) {
  
      try {
  
          BrokerError* error = dynamic_cast<BrokerError*>(data);
  
          dataOut->write(error != NULL);
  
          if (error != NULL) {
  
              looseMarshalString(error->getExceptionClass(), dataOut);
              looseMarshalString(error->getMessage(), dataOut);
  
              if (wireFormat->isStackTraceEnabled()) {
  
                  size_t length = error->getStackTraceElements().size();
  
                  dataOut->writeShort((short) length);
  
                  for (size_t i = 0; i < length; ++i) {
  
                      Pointer<BrokerError::StackTraceElement> element(error->getStackTraceElements()[i]);
  
                      looseMarshalString(element->ClassName, dataOut);
                      looseMarshalString(element->MethodName, dataOut);
                      looseMarshalString(element->FileName, dataOut);
  
                      dataOut->writeInt(element->LineNumber);
                  }
  
                  looseMarshalBrokerError(wireFormat, error->getCause().get(), dataOut);
              }
          }
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::vector<unsigned char> BaseDataStreamMarshaller::tightUnmarshalByteArray(decaf::io::DataInputStream* dataIn, utils::BooleanStream* bs) {
  
      try {
  
          std::vector<unsigned char> data;
          if (bs->readBoolean()) {
              int size = dataIn->readInt();
              if (size > 0) {
                  data.resize(size);
                  dataIn->readFully(&data[0], (int) data.size());
              }
          }
  
          return data;
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::vector<unsigned char> BaseDataStreamMarshaller::looseUnmarshalByteArray(decaf::io::DataInputStream* dataIn) {
  
      try {
  
          if (dataIn->readBoolean()) {
              int size = dataIn->readInt();
              std::vector<unsigned char> data;
              if (size > 0) {
                  data.resize(size);
                  dataIn->readFully(&data[0], (int) data.size());
              }
              return data;
          }
  
          return std::vector<unsigned char>();
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::vector<unsigned char> BaseDataStreamMarshaller::tightUnmarshalConstByteArray(decaf::io::DataInputStream* dataIn, utils::BooleanStream* bs AMQCPP_UNUSED,int size) {
  
      try {
          std::vector<unsigned char> data;
          if (size > 0) {
              data.resize(size);
              dataIn->readFully(&data[0], (int) data.size());
          }
          return data;
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::vector<unsigned char> BaseDataStreamMarshaller::looseUnmarshalConstByteArray(decaf::io::DataInputStream* dataIn, int size) {
  
      try {
          std::vector<unsigned char> data;
          if (size > 0) {
              data.resize(size);
              dataIn->readFully(&data[0], (int) data.size());
          }
          return data;
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::string BaseDataStreamMarshaller::toString(const commands::MessageId* id) {
      if (id == NULL) {
          return "";
      }
  
      return toString(id->getProducerId().get()) + ":" + Long::toString(id->getProducerSequenceId());
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::string BaseDataStreamMarshaller::toString(const commands::ProducerId* id) {
      return id->getConnectionId() + ":" + Long::toString(id->getSessionId()) + ":" + Long::toString(id->getValue());
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::string BaseDataStreamMarshaller::toString(const commands::TransactionId* txnId) {
  
      const LocalTransactionId* ltxnId = dynamic_cast<const LocalTransactionId*>(txnId);
      const XATransactionId* xaTxnId = dynamic_cast<const XATransactionId*>(txnId);
  
      if (ltxnId != NULL) {
          return Long::toString(ltxnId->getValue());
      } else if (xaTxnId != NULL) {
          return string("XID:") + Integer::toString(xaTxnId->getFormatId()) + ":" +
                 toHexFromBytes(xaTxnId->getGlobalTransactionId()) + ":" +
                 toHexFromBytes(xaTxnId->getBranchQualifier());
      }
  
      return "";
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::string BaseDataStreamMarshaller::toHexFromBytes(const std::vector<unsigned char>& data) {
  
      std::string buffer = "";
  
      for (unsigned int i = 0; i < data.size(); i++) {
          buffer.append(hexTable[data[i]]);
      }
  
      return buffer;
  }
  
  ////////////////////////////////////////////////////////////////////////////////
  std::string BaseDataStreamMarshaller::readAsciiString(decaf::io::DataInputStream* dataIn) {
  
      try {
  
          std::string text;
          int size = dataIn->readShort();
  
          if (size > 0) {
  
              std::vector<char> data(size);
              dataIn->readFully((unsigned char*) &data[0], size);
  
              // Now build a string and copy data into it.
              text.insert(text.begin(), data.begin(), data.end());
          }
  
          return text;
      }
      AMQ_CATCH_RETHROW(IOException)
      AMQ_CATCH_EXCEPTION_CONVERT(Exception, IOException)
      AMQ_CATCHALL_THROW(IOException)
  }
  