QPID-4630: HA Fix starting from persistent store.

Reengineered code for converting a Variant to a FieldValue by reinstating
code previously removed.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1454435 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 88 insertions, 45 deletions

diff --git a/qpid/cpp/include/qpid/amqp_0_10/Codecs.h b/qpid/cpp/include/qpid/amqp_0_10/Codecs.h
index 99a06e0a32..bcf3110923 100644
--- a/qpid/cpp/include/qpid/amqp_0_10/Codecs.h
+++ b/qpid/cpp/include/qpid/amqp_0_10/Codecs.h
@@ -76,7 +76,8 @@ QPID_COMMON_EXTERN void translate(const qpid::types::Variant::Map& from, const s
 QPID_COMMON_EXTERN void translate(const qpid::framing::FieldTable& from,
                                   qpid::types::Variant::Map& to);
 
-QPID_COMMON_EXTERN boost::shared_ptr<framing::FieldValue> fieldValue(const types::Variant&);
+QPID_COMMON_EXTERN void translate(const types::Variant& from,
+                                  boost::shared_ptr<qpid::framing::FieldValue> to);
 
 }} // namespace qpid::amqp_0_10
 
diff --git a/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp b/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp
index e7637b16ad..a4db88f128 100644
--- a/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp
+++ b/qpid/cpp/src/qpid/amqp_0_10/Codecs.cpp
@@ -52,7 +52,6 @@ template <class T, class U, class F> void convert(const T& from, U& to, F f)
 }
 
 Variant::Map::value_type toVariantMapEntry(const FieldTable::value_type& in);
-Variant toVariant(boost::shared_ptr<FieldValue> in);
 
 template <class T, class U, class F> void translate(boost::shared_ptr<FieldValue> in, U& u, F f) 
 {
@@ -61,13 +60,6 @@ template <class T, class U, class F> void translate(boost::shared_ptr<FieldValue
     convert(t, u, f);
 }
 
-template <class T, class U, class F> T* toFieldValueCollection(const U& u, F f) 
-{
-    typename T::ValueType t;
-    convert(u, t, f);
-    return new T(t);
-}
-
 void setEncodingFor(Variant& out, uint8_t code)
 {
     switch(code){
@@ -173,34 +165,6 @@ Variant toVariant(boost::shared_ptr<FieldValue> in)
     return out;
 }
 
-boost::shared_ptr<FieldValue> convertString(const std::string& value, const std::string& encoding)
-{
-    bool large = value.size() > std::numeric_limits<uint16_t>::max();
-    if (encoding.empty() || encoding == amqp0_10_binary || encoding == binary) {
-        if (large) {
-            return boost::shared_ptr<FieldValue>(new Var32Value(value, 0xa0));
-        } else {
-            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x90));
-        }
-    } else if (encoding == utf8) {
-        if (!large)
-            return boost::shared_ptr<FieldValue>(new Str16Value(value));
-        throw Exception(QPID_MSG("Could not encode utf8 character string - too long (" << value.size() << " bytes)"));
-    } else if (encoding == utf16) {
-        if (!large)
-            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x96));
-        throw Exception(QPID_MSG("Could not encode utf16 character string - too long (" << value.size() << " bytes)"));
-    } else if (encoding == iso885915) {
-        if (!large)
-            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x94));
-        throw Exception(QPID_MSG("Could not encode iso-8859-15 character string - too long (" << value.size() << " bytes)"));
-    } else {
-        // the encoding was not recognised
-        QPID_LOG(warning, "Unknown byte encoding: [" << encoding << "], encoding as vbin32.");
-        return boost::shared_ptr<FieldValue>(new Var32Value(value, 0xa0));
-    }
-}
-
 Variant::Map::value_type toVariantMapEntry(const FieldTable::value_type& in)
 {
     return Variant::Map::value_type(in.first, toVariant(in.second));
@@ -527,13 +491,89 @@ void translate(const FieldTable& from, Variant::Map& to)
     convert(from, to, &toVariantMapEntry);
 }
 
-boost::shared_ptr<framing::FieldValue> fieldValue(const types::Variant& value) {
-    // TODO aconway 2013-03-05: Nasty implementation
-    Variant::Map m;
-    m[std::string()] = value;
-    framing::FieldTable f;
-    translate(m, f);
-    return f.get(std::string());
+namespace {
+boost::shared_ptr<FieldValue> convertString(const std::string& value, const std::string& encoding);
+FieldTableValue* toFieldTableValue(const Variant::Map& map);
+ListValue* toListValue(const Variant::List& list);
+
+boost::shared_ptr<FieldValue> toFieldValue(const Variant& in)
+{
+    boost::shared_ptr<FieldValue> out;
+    switch (in.getType()) {
+        case VAR_VOID: out = boost::shared_ptr<FieldValue>(new VoidValue()); break;
+        case VAR_BOOL: out = boost::shared_ptr<FieldValue>(new BoolValue(in.asBool())); break;
+        case VAR_UINT8: out = boost::shared_ptr<FieldValue>(new Unsigned8Value(in.asUint8())); break;
+        case VAR_UINT16: out = boost::shared_ptr<FieldValue>(new Unsigned16Value(in.asUint16())); break;
+        case VAR_UINT32: out = boost::shared_ptr<FieldValue>(new Unsigned32Value(in.asUint32())); break;
+        case VAR_UINT64: out = boost::shared_ptr<FieldValue>(new Unsigned64Value(in.asUint64())); break;
+        case VAR_INT8: out = boost::shared_ptr<FieldValue>(new Integer8Value(in.asInt8())); break;
+        case VAR_INT16: out = boost::shared_ptr<FieldValue>(new Integer16Value(in.asInt16())); break;
+        case VAR_INT32: out = boost::shared_ptr<FieldValue>(new Integer32Value(in.asInt32())); break;
+        case VAR_INT64: out = boost::shared_ptr<FieldValue>(new Integer64Value(in.asInt64())); break;
+        case VAR_FLOAT: out = boost::shared_ptr<FieldValue>(new FloatValue(in.asFloat())); break;
+        case VAR_DOUBLE: out = boost::shared_ptr<FieldValue>(new DoubleValue(in.asDouble())); break;
+        case VAR_STRING: out = convertString(in.asString(), in.getEncoding()); break;
+        case VAR_UUID: out = boost::shared_ptr<FieldValue>(new UuidValue(in.asUuid().data())); break;
+        case VAR_MAP:
+            out = boost::shared_ptr<FieldValue>(toFieldTableValue(in.asMap()));
+            break;
+        case VAR_LIST:
+            out = boost::shared_ptr<FieldValue>(toListValue(in.asList()));
+    }
+    return out;
+}
+
+boost::shared_ptr<FieldValue> convertString(const std::string& value, const std::string& encoding)
+{
+    bool large = value.size() > std::numeric_limits<uint16_t>::max();
+    if (encoding.empty() || encoding == amqp0_10_binary || encoding == binary) {
+        if (large) {
+            return boost::shared_ptr<FieldValue>(new Var32Value(value, 0xa0));
+        } else {
+            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x90));
+        }
+    } else if (encoding == utf8) {
+        if (!large)
+            return boost::shared_ptr<FieldValue>(new Str16Value(value));
+        throw Exception(QPID_MSG("Could not encode utf8 character string - too long (" << value.size() << " bytes)"));
+    } else if (encoding == utf16) {
+        if (!large)
+            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x96));
+        throw Exception(QPID_MSG("Could not encode utf16 character string - too long (" << value.size() << " bytes)"));
+    } else if (encoding == iso885915) {
+        if (!large)
+            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x94));
+        throw Exception(QPID_MSG("Could not encode iso-8859-15 character string - too long (" << value.size() << " bytes)"));
+    } else {
+        // the encoding was not recognised
+        QPID_LOG(warning, "Unknown byte encoding: [" << encoding << "], encoding as vbin32.");
+        return boost::shared_ptr<FieldValue>(new Var32Value(value, 0xa0));
+    }
+}
+
+FieldTable::value_type toFieldTableEntry(const Variant::Map::value_type& in)
+{
+    return FieldTable::value_type(in.first, toFieldValue(in.second));
+}
+
+FieldTableValue* toFieldTableValue(const Variant::Map& map)
+{
+    FieldTable ft;
+    convert(map, ft, &toFieldTableEntry);
+    return new FieldTableValue(ft);
+}
+
+ListValue* toListValue(const Variant::List& list)
+{
+    List l;
+    convert(list, l, &toFieldValue);
+    return new ListValue(l);
+}
+}
+
+void translate(const types::Variant& from, boost::shared_ptr<framing::FieldValue> to)
+{
+    to = toFieldValue(from);
 }
 
 const std::string ListCodec::contentType("amqp/list");
diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index 91b497c57b..84fab543ec 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -1589,7 +1589,9 @@ void Queue::UsageBarrier::destroy()
 void Queue::addArgument(const string& key, const types::Variant& value) {
     settings.original[key] = value;
     qpid::amqp_0_10::translate(settings.asMap(), encodableSettings);
-    settings.storeSettings.set(key, qpid::amqp_0_10::fieldValue(value));
+    boost::shared_ptr<qpid::framing::FieldValue> v;
+    qpid::amqp_0_10::translate(value, v);
+    settings.storeSettings.set(key, v);
     if (mgmtObject != 0) mgmtObject->set_arguments(settings.asMap());
 }