You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by lq...@apache.org on 2017/09/28 13:36:07 UTC
[10/12] qpid-broker-j git commit: QPID-7932: [Java Broker,
AMQP 1.0] Improve error handling when deserializing composite types
QPID-7932: [Java Broker, AMQP 1.0] Improve error handling when deserializing composite types
The constructors are now auto-generated with generation being driven from annotations.
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/6a267175
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/6a267175
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/6a267175
Branch: refs/heads/master
Commit: 6a2671755d49f849e69d1392eeaef2c1a9783713
Parents: 85459e5
Author: Lorenz Quack <lq...@apache.org>
Authored: Mon Sep 25 10:36:14 2017 +0100
Committer: Lorenz Quack <lq...@apache.org>
Committed: Thu Sep 28 14:30:17 2017 +0100
----------------------------------------------------------------------
.../server/protocol/v1_0/CompositeType.java | 34 ++
.../v1_0/CompositeTypeConstructorGenerator.java | 443 ++++++++++++++++++
.../protocol/v1_0/CompositeTypeField.java | 35 ++
.../javax.annotation.processing.Processor | 1 +
.../protocol/v1_0/DeserializationFactories.java | 345 ++++++++++++++
.../qpid/server/protocol/v1_0/Session_1_0.java | 86 +---
.../codec/AbstractCompositeTypeConstructor.java | 297 ++++++++++++
.../codec/AbstractDescribedTypeConstructor.java | 68 +--
.../v1_0/codec/CompoundTypeConstructor.java | 193 --------
.../protocol/v1_0/codec/ListConstructor.java | 95 ++++
.../protocol/v1_0/codec/MapConstructor.java | 144 ++++++
.../protocol/v1_0/codec/ValueHandler.java | 6 +-
.../protocol/v1_0/framing/FrameHandler.java | 5 +
.../protocol/v1_0/type/CompositeTypeField.java | 33 --
.../type/codec/AMQPDescribedTypeRegistry.java | 2 +-
.../v1_0/type/messaging/AbstractSection.java | 6 +-
.../protocol/v1_0/type/messaging/Accepted.java | 2 +
.../type/messaging/AmqpSequenceSection.java | 4 +-
.../v1_0/type/messaging/AmqpValueSection.java | 4 +-
.../messaging/ApplicationPropertiesSection.java | 4 +-
.../v1_0/type/messaging/DataSection.java | 4 +-
.../v1_0/type/messaging/DeleteOnClose.java | 2 +
.../v1_0/type/messaging/DeleteOnNoLinks.java | 2 +
.../messaging/DeleteOnNoLinksOrMessages.java | 2 +
.../v1_0/type/messaging/DeleteOnNoMessages.java | 2 +
.../messaging/DeliveryAnnotationsSection.java | 4 +-
.../v1_0/type/messaging/FooterSection.java | 4 +-
.../protocol/v1_0/type/messaging/Header.java | 14 +-
.../v1_0/type/messaging/HeaderSection.java | 4 +-
.../messaging/MessageAnnotationsSection.java | 4 +-
.../protocol/v1_0/type/messaging/Modified.java | 16 +-
.../v1_0/type/messaging/Properties.java | 30 +-
.../v1_0/type/messaging/PropertiesSection.java | 4 +-
.../protocol/v1_0/type/messaging/Received.java | 8 +-
.../protocol/v1_0/type/messaging/Rejected.java | 6 +-
.../protocol/v1_0/type/messaging/Released.java | 2 +
.../protocol/v1_0/type/messaging/Source.java | 45 +-
.../v1_0/type/messaging/StdDistMode.java | 42 +-
.../protocol/v1_0/type/messaging/Target.java | 66 ++-
.../v1_0/type/messaging/TerminusDurability.java | 48 +-
.../type/messaging/TerminusExpiryPolicy.java | 64 ++-
.../messaging/codec/AcceptedConstructor.java | 72 ---
.../codec/DeleteOnCloseConstructor.java | 72 ---
.../codec/DeleteOnNoLinksConstructor.java | 72 ---
.../DeleteOnNoLinksOrMessagesConstructor.java | 72 ---
.../codec/DeleteOnNoMessagesConstructor.java | 72 ---
.../type/messaging/codec/HeaderConstructor.java | 209 ---------
.../messaging/codec/ModifiedConstructor.java | 153 ------
.../messaging/codec/PropertiesConstructor.java | 426 -----------------
.../messaging/codec/ReceivedConstructor.java | 126 -----
.../messaging/codec/RejectedConstructor.java | 98 ----
.../messaging/codec/ReleasedConstructor.java | 71 ---
.../type/messaging/codec/SourceConstructor.java | 385 ---------------
.../type/messaging/codec/TargetConstructor.java | 270 -----------
.../v1_0/type/security/SaslChallenge.java | 27 +-
.../protocol/v1_0/type/security/SaslCode.java | 70 ++-
.../protocol/v1_0/type/security/SaslInit.java | 37 +-
.../v1_0/type/security/SaslMechanisms.java | 22 +-
.../v1_0/type/security/SaslOutcome.java | 33 +-
.../v1_0/type/security/SaslResponse.java | 27 +-
.../codec/SaslChallengeConstructor.java | 99 ----
.../security/codec/SaslInitConstructor.java | 153 ------
.../codec/SaslMechanismsConstructor.java | 106 -----
.../security/codec/SaslOutcomeConstructor.java | 126 -----
.../security/codec/SaslResponseConstructor.java | 99 ----
.../v1_0/type/transaction/Coordinator.java | 24 +-
.../protocol/v1_0/type/transaction/Declare.java | 19 +-
.../v1_0/type/transaction/Declared.java | 6 +-
.../v1_0/type/transaction/Discharge.java | 26 +-
.../type/transaction/TransactionErrors.java | 62 +--
.../type/transaction/TransactionalState.java | 8 +-
.../v1_0/type/transaction/TxnCapabilities.java | 146 ------
.../v1_0/type/transaction/TxnCapability.java | 75 ++-
.../codec/CoordinatorConstructor.java | 117 -----
.../transaction/codec/DeclareConstructor.java | 100 ----
.../transaction/codec/DeclaredConstructor.java | 99 ----
.../transaction/codec/DischargeConstructor.java | 126 -----
.../codec/TransactionalStateConstructor.java | 126 -----
.../protocol/v1_0/type/transport/AmqpError.java | 1 -
.../protocol/v1_0/type/transport/Attach.java | 32 +-
.../protocol/v1_0/type/transport/Begin.java | 20 +-
.../protocol/v1_0/type/transport/Close.java | 6 +-
.../v1_0/type/transport/ConnectionError.java | 1 -
.../protocol/v1_0/type/transport/Detach.java | 10 +-
.../v1_0/type/transport/Disposition.java | 16 +-
.../protocol/v1_0/type/transport/End.java | 6 +-
.../protocol/v1_0/type/transport/Error.java | 18 +-
.../protocol/v1_0/type/transport/Flow.java | 26 +-
.../protocol/v1_0/type/transport/LinkError.java | 1 -
.../protocol/v1_0/type/transport/Open.java | 24 +-
.../v1_0/type/transport/SessionError.java | 1 -
.../protocol/v1_0/type/transport/Transfer.java | 26 +-
.../type/transport/codec/AttachConstructor.java | 465 -------------------
.../type/transport/codec/BeginConstructor.java | 303 ------------
.../type/transport/codec/CloseConstructor.java | 99 ----
.../type/transport/codec/DetachConstructor.java | 153 ------
.../transport/codec/DispositionConstructor.java | 234 ----------
.../type/transport/codec/EndConstructor.java | 99 ----
.../type/transport/codec/ErrorConstructor.java | 224 ---------
.../type/transport/codec/FlowConstructor.java | 370 ---------------
.../type/transport/codec/OpenConstructor.java | 373 ---------------
.../transport/codec/TransferConstructor.java | 369 ---------------
.../protocol/v1_0/store/LinkStoreTestCase.java | 2 +-
.../tests/protocol/v1_0/DecodeErrorTest.java | 196 ++++++++
.../protocol/v1_0/messaging/MessageFormat.java | 49 --
105 files changed, 2115 insertions(+), 7020 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeType.java
----------------------------------------------------------------------
diff --git a/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeType.java b/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeType.java
new file mode 100644
index 0000000..41281c7
--- /dev/null
+++ b/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeType.java
@@ -0,0 +1,34 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v1_0;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface CompositeType
+{
+ String symbolicDescriptor();
+ long numericDescriptor();
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeConstructorGenerator.java
----------------------------------------------------------------------
diff --git a/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeConstructorGenerator.java b/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeConstructorGenerator.java
new file mode 100644
index 0000000..47308c8
--- /dev/null
+++ b/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeConstructorGenerator.java
@@ -0,0 +1,443 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v1_0;
+
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import javax.annotation.processing.AbstractProcessor;
+import javax.annotation.processing.Filer;
+import javax.annotation.processing.ProcessingEnvironment;
+import javax.annotation.processing.RoundEnvironment;
+import javax.lang.model.SourceVersion;
+import javax.lang.model.element.AnnotationMirror;
+import javax.lang.model.element.AnnotationValue;
+import javax.lang.model.element.Element;
+import javax.lang.model.element.ElementKind;
+import javax.lang.model.element.ExecutableElement;
+import javax.lang.model.element.PackageElement;
+import javax.lang.model.element.TypeElement;
+import javax.lang.model.element.VariableElement;
+import javax.lang.model.type.ArrayType;
+import javax.lang.model.type.DeclaredType;
+import javax.lang.model.type.TypeKind;
+import javax.lang.model.type.TypeMirror;
+import javax.lang.model.util.Elements;
+import javax.lang.model.util.Types;
+import javax.tools.Diagnostic;
+import javax.tools.JavaFileObject;
+
+import org.apache.qpid.server.License;
+
+
+public class CompositeTypeConstructorGenerator extends AbstractProcessor
+{
+ private static final List<String> RESTRICTED_TYPES = Arrays.asList(
+ "org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError",
+ "org.apache.qpid.server.protocol.v1_0.type.transport.ConnectionError",
+ "org.apache.qpid.server.protocol.v1_0.type.transport.SessionError",
+ "org.apache.qpid.server.protocol.v1_0.type.transport.LinkError",
+ "org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionErrors",
+ "org.apache.qpid.server.protocol.v1_0.type.transport.ReceiverSettleMode",
+ "org.apache.qpid.server.protocol.v1_0.type.transport.SenderSettleMode",
+ "org.apache.qpid.server.protocol.v1_0.type.transport.Role",
+ "org.apache.qpid.server.protocol.v1_0.type.security.SaslCode",
+ "org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.SoleConnectionDetectionPolicy",
+ "org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.SoleConnectionEnforcementPolicy",
+ "org.apache.qpid.server.protocol.v1_0.type.messaging.StdDistMode",
+ "org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability",
+ "org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusExpiryPolicy",
+ "org.apache.qpid.server.protocol.v1_0.type.transaction.TxnCapability");
+
+
+ @Override
+ public SourceVersion getSupportedSourceVersion()
+ {
+ return SourceVersion.latest();
+ }
+
+ @Override
+ public Set<String> getSupportedAnnotationTypes()
+ {
+ return Collections.singleton(CompositeType.class.getName());
+ }
+
+ @Override
+ public boolean process(final Set<? extends TypeElement> annotations, final RoundEnvironment roundEnvironment)
+ {
+ if(roundEnvironment.processingOver())
+ {
+ return true;
+ }
+
+ Filer filer = processingEnv.getFiler();
+ try
+ {
+ for (Element e : roundEnvironment.getElementsAnnotatedWith(CompositeType.class))
+ {
+ if(e.getKind() == ElementKind.CLASS)
+ {
+ generateCompositeTypeConstructor(filer, (TypeElement) e);
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ try(StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw))
+ {
+ e.printStackTrace(pw);
+ processingEnv.getMessager().printMessage(Diagnostic.Kind.ERROR, "Unexpected Error: " + sw.toString());
+ }
+ catch (IOException ioe)
+ {
+ processingEnv.getMessager().printMessage(Diagnostic.Kind.ERROR, "Error: " + ioe.getLocalizedMessage());
+ }
+ }
+ return true;
+ }
+
+
+ private void generateCompositeTypeConstructor(final Filer filer, final TypeElement typeElement)
+ {
+ String objectQualifiedClassName = typeElement.getQualifiedName().toString();
+ String objectSimpleName = typeElement.getSimpleName().toString();
+ String compositeTypeConstructorNameSimpleName = objectSimpleName + "Constructor";
+ PackageElement packageElement = (PackageElement) typeElement.getEnclosingElement();
+ final String compositeTypeConstructorPackage = packageElement.getQualifiedName() + ".codec";
+ String compositeTypeConstructorName = compositeTypeConstructorPackage + "." + compositeTypeConstructorNameSimpleName;
+ final CompositeType annotation = typeElement.getAnnotation(CompositeType.class);
+
+ processingEnv.getMessager().printMessage(Diagnostic.Kind.NOTE, "Generating composite constructor file for " + objectQualifiedClassName);
+
+ try
+ {
+ JavaFileObject factoryFile = filer.createSourceFile(compositeTypeConstructorName);
+ PrintWriter pw = new PrintWriter(new OutputStreamWriter(factoryFile.openOutputStream(), "UTF-8"));
+ pw.println("/*");
+ for(String headerLine : License.LICENSE)
+ {
+ pw.println(" *" + headerLine);
+ }
+ pw.println(" */");
+ pw.println();
+ pw.print("package ");
+ pw.print(compositeTypeConstructorPackage);
+ pw.println(";");
+ pw.println();
+
+ pw.println("import java.util.List;");
+ pw.println();
+ pw.println("import org.apache.qpid.server.protocol.v1_0.codec.AbstractCompositeTypeConstructor;");
+ pw.println("import org.apache.qpid.server.protocol.v1_0.codec.DescribedTypeConstructorRegistry;");
+ pw.println("import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;");
+ pw.println("import org.apache.qpid.server.protocol.v1_0.type.Symbol;");
+ pw.println("import org.apache.qpid.server.protocol.v1_0.type.UnsignedLong;");
+ pw.println("import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;");
+ pw.println("import org.apache.qpid.server.protocol.v1_0.type.transport.Error;");
+ pw.println("import " + objectQualifiedClassName + ";");
+ pw.println();
+
+ pw.println("public final class " + compositeTypeConstructorNameSimpleName + " extends AbstractCompositeTypeConstructor<"+ objectSimpleName +">");
+ pw.println("{");
+ pw.println(" private static final " + compositeTypeConstructorNameSimpleName + " INSTANCE = new " + compositeTypeConstructorNameSimpleName + "();");
+ pw.println();
+
+ pw.println(" public static void register(DescribedTypeConstructorRegistry registry)");
+ pw.println(" {");
+ pw.println(" registry.register(Symbol.valueOf(\"" + annotation.symbolicDescriptor() + "\"), INSTANCE);");
+ pw.println(String.format(" registry.register(UnsignedLong.valueOf(%#016x), INSTANCE);", annotation.numericDescriptor()));
+ pw.println(" }");
+ pw.println();
+
+ pw.println(" @Override");
+ pw.println(" protected String getTypeName()");
+ pw.println(" {");
+ pw.println(" return " + objectSimpleName + ".class.getSimpleName();");
+ pw.println(" }");
+ pw.println();
+
+ pw.println(" @Override");
+ pw.println(" protected " + objectSimpleName + " construct(final FieldValueReader fieldValueReader) throws AmqpErrorException");
+ pw.println(" {");
+ pw.println(" " + objectSimpleName + " obj = new " + objectSimpleName + "();");
+ pw.println();
+ generateAssigners(pw, typeElement);
+ pw.println(" return obj;");
+ pw.println(" }");
+ pw.println("}");
+ pw.close();
+ }
+ catch (IOException e)
+ {
+ processingEnv.getMessager().printMessage(Diagnostic.Kind.ERROR,
+ "Failed to write composite constructor file: "
+ + compositeTypeConstructorName
+ + " - "
+ + e.getLocalizedMessage());
+ }
+ }
+
+ private void generateAssigners(final PrintWriter pw, final TypeElement typeElement)
+ {
+ Types typeUtils = processingEnv.getTypeUtils();
+
+ final List<AnnotatedField> annotatedFields = new ArrayList<>();
+ for (Element element : typeElement.getEnclosedElements())
+ {
+ if (element instanceof VariableElement && element.getKind() == ElementKind.FIELD)
+ {
+ boolean annotationFound = false;
+ for(AnnotationMirror annotationMirror : element.getAnnotationMirrors())
+ {
+ if(annotationMirror.getAnnotationType().toString().equals("org.apache.qpid.server.protocol.v1_0.CompositeTypeField"))
+ {
+ if (annotationFound)
+ {
+ processingEnv.getMessager()
+ .printMessage(Diagnostic.Kind.ERROR,
+ String.format(
+ "More than one CompositeTypeField annotations on field '%s.%s'",
+ typeElement.getSimpleName(),
+ element.getSimpleName()));
+ }
+ annotationFound = true;
+ annotatedFields.add(new AnnotatedField((VariableElement) element, annotationMirror));
+ }
+ }
+ }
+ }
+
+ annotatedFields.sort(Comparator.comparingInt(AnnotatedField::getIndex));
+
+ for (int index = 0; index < annotatedFields.size(); ++index)
+ {
+ AnnotatedField annotatedField = annotatedFields.get(index);
+ final VariableElement variableElement = annotatedField.getVariableElement();
+ final String fieldName = stripUnderscore(variableElement.getSimpleName().toString());
+ if (annotatedField.getIndex() != index)
+ {
+ processingEnv.getMessager().printMessage(Diagnostic.Kind.ERROR,String.format(
+ "Unexpected CompositeTypeField index '%d' is specified on field '%s' of '%s'. Expected %d.",
+ annotatedField.getIndex(),
+ fieldName,
+ typeElement.getSimpleName(),
+ index));
+ }
+
+ final String baseIndent = " ";
+ if (variableElement.asType().getKind() == TypeKind.ARRAY)
+ {
+ final TypeMirror componentType = ((ArrayType) variableElement.asType()).getComponentType();
+ final String functionString;
+ if (annotatedField.getFactory() != null)
+ {
+ functionString = "x -> " + annotatedField.getFactory() + "(x)";
+ }
+ else if (RESTRICTED_TYPES.contains(componentType))
+ {
+ functionString = variableElement.asType().toString() + "::valueOf";
+ }
+ else
+ {
+ functionString = "x -> (" + componentType + ") x";
+ }
+ pw.println(String.format(" %s %s = fieldValueReader.readArrayValue(%d, \"%s\", %s, %s.class, %s);",
+ annotatedField.getVariableElement().asType(),
+ fieldName,
+ index,
+ fieldName,
+ annotatedField.isMandatory(),
+ componentType,
+ functionString));
+ optionallyWrapInNullCheck(!annotatedField.isMandatory(), pw, baseIndent, fieldName, indent -> {
+ pw.println(indent + "obj." + getSetterName(variableElement) + "(" + fieldName + ");");
+ });
+ }
+ else if (annotatedField.getFactory() != null || RESTRICTED_TYPES.contains(variableElement.asType().toString()))
+ {
+ String functionName = annotatedField.getFactory() != null ? annotatedField.getFactory() : variableElement.asType().toString() + ".valueOf";
+ pw.println(String.format(" Object %s = fieldValueReader.readValue(%d, \"%s\", %s, Object.class);",
+ fieldName,
+ index,
+ fieldName,
+ annotatedField.isMandatory()));
+ optionallyWrapInNullCheck(!annotatedField.isMandatory(), pw, baseIndent, fieldName, indent -> {
+ pw.println(indent + "try");
+ pw.println(indent + "{");
+ pw.println(indent + " obj." + getSetterName(variableElement) + "(" + functionName + "(" + fieldName + "));");
+ pw.println(indent + "}");
+ pw.println(indent + "catch (RuntimeException e)");
+ pw.println(indent + "{");
+ pw.println(indent + " Error error = new Error(AmqpError.DECODE_ERROR, \"Could not decode value field '" + fieldName + "' of '" + typeElement.getSimpleName() + "'\");");
+ pw.println(indent + " throw new AmqpErrorException(error, e);");
+ pw.println(indent + "}");
+ });
+ }
+ else if (typeUtils.isSameType(typeUtils.erasure(variableElement.asType()),
+ getErasure(processingEnv, "java.util.Map")))
+ {
+ List<? extends TypeMirror> args = ((DeclaredType) variableElement.asType()).getTypeArguments();
+ if (args.size() != 2)
+ {
+ processingEnv.getMessager().printMessage(Diagnostic.Kind.ERROR,
+ "Map types must have exactly two type arguments");
+ }
+ pw.println(String.format(" %s %s = fieldValueReader.readMapValue(%d, \"%s\", %s, %s.class, %s.class);",
+ annotatedField.getVariableElement().asType(),
+ fieldName,
+ index,
+ fieldName,
+ annotatedField.isMandatory(),
+ args.get(0),
+ args.get(1)));
+ optionallyWrapInNullCheck(!annotatedField.isMandatory(), pw, baseIndent, fieldName, indent -> {
+ pw.println(indent + "obj." + getSetterName(variableElement) + "(" + fieldName + ");");
+ });
+ }
+ else
+ {
+ pw.println(String.format(" %s %s = fieldValueReader.readValue(%d, \"%s\", %s, %s.class);",
+ annotatedField.getVariableElement().asType(),
+ fieldName,
+ index,
+ fieldName,
+ annotatedField.isMandatory(),
+ annotatedField.getVariableElement().asType()));
+ optionallyWrapInNullCheck(!annotatedField.isMandatory(), pw, baseIndent, fieldName, indent -> {
+ pw.println(indent + "obj." + getSetterName(variableElement) + "(" + fieldName + ");");
+ });
+ }
+
+ pw.println();
+ }
+ }
+
+ private void optionallyWrapInNullCheck(boolean wrap, PrintWriter pw, String indent, String fieldName, Consumer<String> f)
+ {
+ if (wrap)
+ {
+ pw.println(indent + "if (" + fieldName + " != null)");
+ pw.println(indent + "{");
+ indent += " ";
+ }
+ f.accept(indent);
+ if (wrap)
+ {
+ indent = indent.substring(4);
+ pw.println(indent + "}");
+ }
+ }
+
+ private String getSetterName(final VariableElement variableElement)
+ {
+ final String fieldName = stripUnderscore(variableElement.getSimpleName().toString());
+ return "set" + fieldName.substring(0, 1).toUpperCase() + fieldName.substring(1);
+ }
+
+ private String stripUnderscore(final String fieldName)
+ {
+ if (fieldName.startsWith("_"))
+ {
+ return fieldName.substring(1);
+ }
+ return fieldName;
+ }
+
+ private static class AnnotatedField
+ {
+ private final VariableElement _variableElement;
+ private final AnnotationMirror _annotationMirror;
+ private final int _index;
+ private final String _factory;
+ private final boolean _mandatory;
+
+ public AnnotatedField(final VariableElement variableElement, final AnnotationMirror annotationMirror)
+ {
+ _variableElement = variableElement;
+ _annotationMirror = annotationMirror;
+ String factory = null;
+ boolean mandatory = false;
+ int index = -1;
+ for (Map.Entry<? extends ExecutableElement,? extends AnnotationValue> entry : annotationMirror.getElementValues().entrySet())
+ {
+ if ("index".contentEquals(entry.getKey().getSimpleName()))
+ {
+ index = (int) entry.getValue().getValue();
+ }
+ else if ("deserializationConverter".contentEquals(entry.getKey().getSimpleName()))
+ {
+ factory = (String) entry.getValue().getValue();
+ }
+ else if ("mandatory".contentEquals(entry.getKey().getSimpleName()))
+ {
+ mandatory = (boolean) entry.getValue().getValue();
+ }
+ }
+ _index = index;
+ _mandatory = mandatory;
+ _factory = factory;
+ }
+
+ public VariableElement getVariableElement()
+ {
+ return _variableElement;
+ }
+
+ public AnnotationMirror getAnnotationMirror()
+ {
+ return _annotationMirror;
+ }
+
+ public int getIndex()
+ {
+ return _index;
+ }
+
+ public String getFactory()
+ {
+ return _factory;
+ }
+
+ public boolean isMandatory()
+ {
+ return _mandatory;
+ }
+ }
+
+ private static TypeMirror getErasure(ProcessingEnvironment processingEnv, final String className)
+ {
+ final Types typeUtils = processingEnv.getTypeUtils();
+ final Elements elementUtils = processingEnv.getElementUtils();
+ return typeUtils.erasure(elementUtils.getTypeElement(className).asType());
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeField.java
----------------------------------------------------------------------
diff --git a/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeField.java b/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeField.java
new file mode 100644
index 0000000..be2893e
--- /dev/null
+++ b/broker-codegen/src/main/java/org/apache/qpid/server/protocol/v1_0/CompositeTypeField.java
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v1_0;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.FIELD)
+public @interface CompositeTypeField
+{
+ int index();
+ boolean mandatory() default false;
+ String deserializationConverter() default "";
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-codegen/src/main/resources/META-INF/services/javax.annotation.processing.Processor
----------------------------------------------------------------------
diff --git a/broker-codegen/src/main/resources/META-INF/services/javax.annotation.processing.Processor b/broker-codegen/src/main/resources/META-INF/services/javax.annotation.processing.Processor
index c8ef012..8e295cf 100644
--- a/broker-codegen/src/main/resources/META-INF/services/javax.annotation.processing.Processor
+++ b/broker-codegen/src/main/resources/META-INF/services/javax.annotation.processing.Processor
@@ -25,3 +25,4 @@ org.apache.qpid.server.model.validation.AttributeFieldValidation
org.apache.qpid.server.model.validation.ManagedAnnotationValidator
org.apache.qpid.server.model.validation.OperationAnnotationValidator
org.apache.qpid.server.model.validation.ContentHeaderAnnotationValidator
+org.apache.qpid.server.protocol.v1_0.CompositeTypeConstructorGenerator
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/DeserializationFactories.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/DeserializationFactories.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/DeserializationFactories.java
new file mode 100644
index 0000000..f76edf4
--- /dev/null
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/DeserializationFactories.java
@@ -0,0 +1,345 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v1_0;
+
+import java.lang.reflect.Array;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
+import org.apache.qpid.server.protocol.v1_0.type.DistributionMode;
+import org.apache.qpid.server.protocol.v1_0.type.ErrorCondition;
+import org.apache.qpid.server.protocol.v1_0.type.LifetimePolicy;
+import org.apache.qpid.server.protocol.v1_0.type.Symbol;
+import org.apache.qpid.server.protocol.v1_0.type.TxnCapability;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.StdDistMode;
+import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionErrors;
+import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
+import org.apache.qpid.server.protocol.v1_0.type.transport.ConnectionError;
+import org.apache.qpid.server.protocol.v1_0.type.transport.LinkError;
+import org.apache.qpid.server.protocol.v1_0.type.transport.SessionError;
+
+public class DeserializationFactories
+{
+ @SuppressWarnings("unused")
+ public static Map<Symbol, Object> convertToNodeProperties(final Object value) throws AmqpErrorException
+ {
+ if (value != null)
+ {
+ if (!(value instanceof Map))
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Cannot construct 'node-properties' from type '%s'",
+ value.getClass().getSimpleName()));
+ }
+ Map<Symbol, Object> nodeProperties = new LinkedHashMap<>();
+ Map<?, ?> map = (Map<?, ?>) value;
+ for (Map.Entry<?,?> entry : map.entrySet())
+ {
+ Object key = entry.getKey();
+ if (!(key instanceof Symbol))
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("'node-properties' must have only keys of type 'symbol' but got '%s'",
+ key.getClass().getSimpleName()));
+ }
+ if (Session_1_0.LIFETIME_POLICY.equals(key))
+ {
+ final Object lifetimePolicy = entry.getValue();
+ if (!(lifetimePolicy instanceof LifetimePolicy))
+ {
+ String typeName = lifetimePolicy == null ? null : lifetimePolicy.getClass().getSimpleName();
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Cannot construct 'lifetime-policy' from type '%s'",
+ typeName));
+ }
+ nodeProperties.put((Symbol) key, lifetimePolicy);
+ }
+ else if (Symbol.valueOf("supported-dist-modes").equals(key))
+ {
+ final Object distributionMode = entry.getValue();
+ final DistributionMode[] converted;
+ if (distributionMode == null)
+ {
+ converted = null;
+
+ }
+ else if (distributionMode.getClass().isArray())
+ {
+ converted = new DistributionMode[Array.getLength(distributionMode)];
+ for (int i = 0; i < converted.length; ++i)
+ {
+ final Object item = Array.get(distributionMode, i);
+ if (item == null)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ "'null' not allowed in 'supported-distribution-modes'");
+ }
+ converted[i] = convertToDistributionMode(item);
+ }
+ }
+ else
+ {
+ converted = new DistributionMode[] {convertToDistributionMode(distributionMode)};
+ }
+ nodeProperties.put((Symbol) key, converted);
+ }
+ else
+ {
+ nodeProperties.put((Symbol) key, entry.getValue());
+ }
+ }
+ return nodeProperties;
+ }
+ else
+ {
+ return null;
+ }
+ }
+
+ @SuppressWarnings("unused")
+ public static DistributionMode convertToDistributionMode(final Object value) throws AmqpErrorException
+ {
+ DistributionMode distributionMode = null;
+ if (value != null)
+ {
+ if (value instanceof DistributionMode)
+ {
+ distributionMode = (DistributionMode) value;
+ }
+ else if (value instanceof Symbol)
+ {
+ distributionMode = StdDistMode.valueOf(value);
+ if (distributionMode == null)
+ {
+ distributionMode = new UnknownDistributionMode((Symbol) value);
+ }
+ }
+ else
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Cannot construct 'distribution-mode' from type '%s'",
+ value.getClass().getSimpleName()));
+ }
+ }
+ return distributionMode;
+ }
+
+ @SuppressWarnings("unused")
+ public static TxnCapability convertToTxnCapability(final Object value) throws AmqpErrorException
+ {
+ TxnCapability capability = null;
+ if (value != null)
+ {
+ if (value instanceof TxnCapability)
+ {
+ capability = (TxnCapability) value;
+ }
+ else if (value instanceof Symbol)
+ {
+ capability = org.apache.qpid.server.protocol.v1_0.type.transaction.TxnCapability.valueOf(value);
+ if (capability == null)
+ {
+ capability = new UnknownTxnCapability((Symbol) value);
+ }
+ }
+ else
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Cannot construct 'txn-capability' from type '%s'",
+ value.getClass().getSimpleName()));
+ }
+ }
+ return capability;
+ }
+
+ @SuppressWarnings("unsued")
+ public static ErrorCondition convertToErrorCondition(final Object value) throws AmqpErrorException
+ {
+ ErrorCondition condition = null;
+ if (value != null)
+ {
+ if (value instanceof ErrorCondition)
+ {
+ condition = (ErrorCondition) value;
+ }
+ else if (value instanceof Symbol)
+ {
+ condition = AmqpError.valueOf(value);
+ if (condition == null)
+ {
+ condition = ConnectionError.valueOf(value);
+ if (condition == null)
+ {
+ condition = SessionError.valueOf(value);
+ if (condition == null)
+ {
+ condition = LinkError.valueOf(value);
+ if (condition == null)
+ {
+ condition = TransactionErrors.valueOf(value);
+ if (condition == null)
+ {
+ condition = new UnknownErrorCondition((Symbol) value);
+ }
+ }
+ }
+ }
+ }
+ }
+ else
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Cannot construct 'error-condition' from type '%s'",
+ value.getClass().getSimpleName()));
+ }
+ }
+ return condition;
+ }
+
+ private static final class UnknownErrorCondition implements ErrorCondition
+ {
+ private final Symbol _value;
+
+ public UnknownErrorCondition(final Symbol value)
+ {
+ _value = value;
+ }
+
+ @Override
+ public Symbol getValue()
+ {
+ return _value;
+ }
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ final UnknownErrorCondition that = (UnknownErrorCondition) o;
+
+ if (!_value.equals(that._value))
+ {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return _value.hashCode();
+ }
+
+ @Override
+ public String toString()
+ {
+ return _value.toString();
+ }
+ }
+
+ private static class UnknownTxnCapability implements TxnCapability
+ {
+ private final Symbol _value;
+
+ public UnknownTxnCapability(final Symbol value)
+ {
+ _value = value;
+ }
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ final UnknownTxnCapability that = (UnknownTxnCapability) o;
+
+ return _value.equals(that._value);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return _value.hashCode();
+ }
+
+ @Override
+ public String toString()
+ {
+ return _value.toString();
+ }
+ }
+
+ private static class UnknownDistributionMode implements DistributionMode
+ {
+ private final Symbol _value;
+
+ public UnknownDistributionMode(final Symbol value)
+ {
+ _value = value;
+ }
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+
+ final UnknownDistributionMode that = (UnknownDistributionMode) o;
+
+ return _value.equals(that._value);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return _value.hashCode();
+ }
+
+ @Override
+ public String toString()
+ {
+ return _value.toString();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index 801810e..a78df1a 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -988,15 +988,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
@Override
public void block(final Queue<?> queue)
{
- getAMQPConnection().doOnIOThreadAsync(
- new Runnable()
- {
- @Override
- public void run()
- {
- doBlock(queue);
- }
- });
+ getAMQPConnection().doOnIOThreadAsync(() -> doBlock(queue));
}
private void doBlock(final Queue<?> queue)
@@ -1025,15 +1017,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
@Override
public void unblock(final Queue<?> queue)
{
- getAMQPConnection().doOnIOThreadAsync(
- new Runnable()
- {
- @Override
- public void run()
- {
- doUnblock(queue);
- }
- });
+ getAMQPConnection().doOnIOThreadAsync(() -> doUnblock(queue));
}
private void doUnblock(final Queue<?> queue)
@@ -1058,15 +1042,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
@Override
public void block()
{
- getAMQPConnection().doOnIOThreadAsync(
- new Runnable()
- {
- @Override
- public void run()
- {
- doBlock();
- }
- });
+ getAMQPConnection().doOnIOThreadAsync(this::doBlock);
}
private void doBlock()
@@ -1088,15 +1064,7 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
@Override
public void unblock()
{
- getAMQPConnection().doOnIOThreadAsync(
- new Runnable()
- {
- @Override
- public void run()
- {
- doUnblock();
- }
- });
+ getAMQPConnection().doOnIOThreadAsync(this::doUnblock);
}
private void doUnblock()
@@ -1388,38 +1356,32 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
@Override
public void onSuccess(final T endpoint)
{
- doOnIOThreadAsync(new Runnable()
- {
- @Override
- public void run()
+ doOnIOThreadAsync(() -> {
+ _associatedLinkEndpoints.add(endpoint);
+ endpoint.setLocalHandle(findNextAvailableOutputHandle());
+ if (endpoint instanceof ErrantLinkEndpoint)
{
- _associatedLinkEndpoints.add(endpoint);
- endpoint.setLocalHandle(findNextAvailableOutputHandle());
- if (endpoint instanceof ErrantLinkEndpoint)
+ endpoint.sendAttach();
+ ((ErrantLinkEndpoint) endpoint).closeWithError();
+ }
+ else
+ {
+ if (endpoint instanceof StandardReceivingLinkEndpoint
+ && (_blockingEntities.contains(Session_1_0.this)
+ || _blockingEntities.contains(((StandardReceivingLinkEndpoint) endpoint).getReceivingDestination())))
+ {
+ endpoint.setStopped(true);
+ }
+ _inputHandleToEndpoint.put(_attach.getHandle(), endpoint);
+ if (!_endpointToOutputHandle.containsKey(endpoint))
{
+ _endpointToOutputHandle.put(endpoint, endpoint.getLocalHandle());
endpoint.sendAttach();
- ((ErrantLinkEndpoint) endpoint).closeWithError();
+ endpoint.start();
}
else
{
- if (endpoint instanceof StandardReceivingLinkEndpoint
- && (_blockingEntities.contains(Session_1_0.this)
- || _blockingEntities.contains(((StandardReceivingLinkEndpoint) endpoint).getReceivingDestination())))
- {
- endpoint.setStopped(true);
- }
- _inputHandleToEndpoint.put(_attach.getHandle(), endpoint);
- if (!_endpointToOutputHandle.containsKey(endpoint))
- {
- _endpointToOutputHandle.put(endpoint, endpoint.getLocalHandle());
- endpoint.sendAttach();
- endpoint.start();
- }
- else
- {
- // TODO - link stealing???
- }
-
+ // TODO - link stealing???
}
}
});
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractCompositeTypeConstructor.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractCompositeTypeConstructor.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractCompositeTypeConstructor.java
new file mode 100644
index 0000000..c417b69
--- /dev/null
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractCompositeTypeConstructor.java
@@ -0,0 +1,297 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v1_0.codec;
+
+
+import java.lang.reflect.Array;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.bytebuffer.QpidByteBufferUtils;
+import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
+import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
+
+public abstract class AbstractCompositeTypeConstructor<T> implements DescribedTypeConstructor<T>
+{
+ @Override
+ public TypeConstructor<T> construct(final Object descriptor,
+ final List<QpidByteBuffer> in,
+ final int[] originalPositions,
+ final ValueHandler valueHandler) throws AmqpErrorException
+ {
+ return new FieldValueReader();
+ }
+
+ protected abstract String getTypeName();
+
+ protected abstract T construct(FieldValueReader x) throws AmqpErrorException;
+
+ protected class FieldValueReader implements TypeConstructor<T>
+ {
+ private List<QpidByteBuffer> _in;
+ private ValueHandler _valueHandler;
+ private int _count;
+
+ @Override
+ public T construct(final List<QpidByteBuffer> in, final ValueHandler handler) throws AmqpErrorException
+ {
+ _in = in;
+ _valueHandler = handler;
+ return constructType();
+ }
+
+ private T constructType() throws AmqpErrorException
+ {
+ final TypeConstructor typeConstructor = _valueHandler.readConstructor(_in);
+ if (typeConstructor instanceof ListConstructor)
+ {
+ ListConstructor listConstructor = (ListConstructor) typeConstructor;
+ int size;
+ long remaining = QpidByteBufferUtils.remaining(_in);
+ if (remaining < listConstructor.getSize())
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Not sufficient data for deserialization of '%s'."
+ + " Expected at least %d bytes. Got %d bytes.",
+ getTypeName(),
+ listConstructor.getSize(),
+ remaining));
+ }
+
+ if (listConstructor.getSize() == 1)
+ {
+ size = QpidByteBufferUtils.get(_in) & 0xFF;
+ _count = QpidByteBufferUtils.get(_in) & 0xFF;
+ }
+ else
+ {
+ size = QpidByteBufferUtils.getInt(_in);
+ _count = QpidByteBufferUtils.getInt(_in);
+ }
+
+ remaining -= listConstructor.getSize();
+ if (remaining < size)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Not sufficient data for deserialization of '%s'."
+ + " Expected at least %d bytes. Got %d bytes.",
+ getTypeName(),
+ size,
+ remaining));
+ }
+ }
+ else if (typeConstructor instanceof ZeroListConstructor)
+ {
+ _count = 0;
+ }
+ else
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Unexpected format when deserializing of '%s'",
+ getTypeName()));
+ }
+ return AbstractCompositeTypeConstructor.this.construct(this);
+ }
+
+
+ public <F> F readValue(final int fieldIndex,
+ final String fieldName,
+ final boolean mandatory,
+ final Class<F> expectedType) throws AmqpErrorException
+ {
+ if (fieldIndex >= _count)
+ {
+ if (mandatory)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Mandatory field '%s' of '%s' was not provided",
+ fieldName,
+ getTypeName()));
+ }
+ return null;
+ }
+
+ Object value = _valueHandler.parse(_in);
+
+
+ if (value == null && mandatory)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Mandatory field '%s' of '%s' was not provided",
+ fieldName,
+ getTypeName()));
+ }
+
+ if (value != null && !expectedType.isAssignableFrom(value.getClass()))
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format(
+ "Wrong type for field '%s' of '%s'. Expected '%s' but got '%s'.",
+ fieldName,
+ getTypeName(),
+ expectedType.getSimpleName(),
+ value.getClass().getSimpleName()));
+ }
+
+ return (F) value;
+ }
+
+ public <K, V> Map<K, V> readMapValue(final int fieldIndex,
+ final String fieldName,
+ final boolean mandatory,
+ final Class<K> expectedKeyType,
+ final Class<V> expectedValueType)
+ throws AmqpErrorException
+ {
+ if (fieldIndex >= _count)
+ {
+ if (mandatory)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Mandatory field '%s' of '%s' was not provided",
+ fieldName,
+ getTypeName()));
+ }
+ return null;
+ }
+
+ TypeConstructor typeConstructor = _valueHandler.readConstructor(_in);
+ if (typeConstructor instanceof MapConstructor)
+ {
+ MapConstructor mapConstructor = ((MapConstructor) typeConstructor);
+
+ return mapConstructor.construct(_in,
+ _valueHandler,
+ expectedKeyType,
+ expectedValueType);
+ }
+ else if (typeConstructor instanceof NullTypeConstructor)
+ {
+ if (mandatory)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ "Mandatory field '%s' of '%s' was not provided",
+ fieldName,
+ getTypeName());
+ }
+ }
+ else
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Could not decode value field '%s' of '%s'",
+ fieldName,
+ getTypeName()));
+ }
+
+ return null;
+ }
+
+ public <F> F[] readArrayValue(final int fieldIndex,
+ final String fieldName,
+ final boolean mandatory,
+ final Class<F> expectedType,
+ final Converter<F> converter) throws AmqpErrorException
+ {
+ if (fieldIndex >= _count)
+ {
+ if (mandatory)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Mandatory field '%s' of '%s' was not provided",
+ fieldName,
+ getTypeName()));
+ }
+ return null;
+ }
+
+ Object value = _valueHandler.parse(_in);
+
+ if (mandatory && value == null)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Mandatory field '%s' of '%s' was not provided",
+ fieldName,
+ getTypeName()));
+ }
+
+ if (value != null)
+ {
+ if (value.getClass().isArray())
+ {
+ if (expectedType.isAssignableFrom(value.getClass().getComponentType()))
+ {
+ return (F[]) value;
+ }
+ else
+ {
+ final Object[] objects = (Object[]) value;
+ F[] array = (F[]) Array.newInstance(expectedType, objects.length);
+ try
+ {
+ for (int i = 0; i < objects.length; ++i)
+ {
+ array[i] = converter.convert(objects[i]);
+ }
+ }
+ catch (RuntimeException e)
+ {
+ Error error = new Error(AmqpError.DECODE_ERROR,
+ String.format("Could not decode value field '%s' of '%s'", fieldName, getTypeName()));
+ throw new AmqpErrorException(error, e);
+ }
+ return array;
+ }
+ }
+ else if (expectedType.isAssignableFrom(value.getClass()))
+ {
+ F[] array = (F[]) Array.newInstance(expectedType, 1);
+ array[0] = (F) value;
+ return array;
+ }
+ else
+ {
+ try
+ {
+ final F convertedValue = converter.convert(value);
+ F[] array = (F[]) Array.newInstance(expectedType, 1);
+ array[0] = convertedValue;
+ return array;
+ }
+ catch (RuntimeException e)
+ {
+ Error error = new Error(AmqpError.DECODE_ERROR,
+ String.format("Could not decode value field '%s' of '%s'", fieldName, getTypeName()));
+ throw new AmqpErrorException(error, e);
+ }
+ }
+ }
+
+ return null;
+ }
+ }
+
+ public interface Converter<T>
+ {
+ T convert(Object o) throws AmqpErrorException;
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractDescribedTypeConstructor.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractDescribedTypeConstructor.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractDescribedTypeConstructor.java
index e8a14ff..c8da269 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractDescribedTypeConstructor.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/AbstractDescribedTypeConstructor.java
@@ -20,43 +20,29 @@
*/
package org.apache.qpid.server.protocol.v1_0.codec;
-import java.lang.annotation.Annotation;
-import java.lang.reflect.Field;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
-import org.apache.qpid.server.protocol.v1_0.type.CompositeTypeField;
-import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
public abstract class AbstractDescribedTypeConstructor<T extends Object> implements DescribedTypeConstructor<T>
{
- private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDescribedTypeConstructor.class);
-
@Override
public TypeConstructor<T> construct(final Object descriptor,
final List<QpidByteBuffer> in,
final int[] originalPositions, final ValueHandler valueHandler)
throws AmqpErrorException
{
-
return new TypeConstructorFromUnderlying<>(this, valueHandler.readConstructor(in));
}
- protected abstract T construct(Object underlying);
+ protected abstract T construct(Object underlying) throws AmqpErrorException;
private static class TypeConstructorFromUnderlying<S extends Object> implements TypeConstructor<S>
{
private final TypeConstructor _describedConstructor;
private AbstractDescribedTypeConstructor<S> _describedTypeConstructor;
- private static final Map<Class<?>, CompositeTypeValidator> _validators = new ConcurrentHashMap<>();
public TypeConstructorFromUnderlying(final AbstractDescribedTypeConstructor<S> describedTypeConstructor,
final TypeConstructor describedConstructor)
@@ -68,57 +54,7 @@ public abstract class AbstractDescribedTypeConstructor<T extends Object> impleme
@Override
public S construct(final List<QpidByteBuffer> in, final ValueHandler handler) throws AmqpErrorException
{
- final S constructedObject =
- _describedTypeConstructor.construct(_describedConstructor.construct(in, handler));
- CompositeTypeValidator<S> validator =
- _validators.computeIfAbsent(constructedObject.getClass(), k -> createValidator(constructedObject));
- validator.validate(constructedObject);
- return constructedObject;
- }
-
- private CompositeTypeValidator<S> createValidator(final S constructedObject)
- {
- final List<Field> mandatoryFields = new ArrayList<>();
- for (Field field : constructedObject.getClass().getDeclaredFields())
- {
- Annotation[] annotations = field.getDeclaredAnnotationsByType(CompositeTypeField.class);
- for (Annotation annotation : annotations)
- {
- if (annotation instanceof CompositeTypeField && ((CompositeTypeField) annotation).mandatory())
- {
- field.setAccessible(true);
- mandatoryFields.add(field);
- }
- }
- }
- return objectToValidate ->
- {
- try
- {
- if (!mandatoryFields.isEmpty())
- {
- for (Field field : mandatoryFields)
- {
- if (field.get(objectToValidate) == null)
- {
- throw new AmqpErrorException(AmqpError.DECODE_ERROR,
- String.format("Missing mandatory field '%s'.",
- field.getName()));
- }
- }
- }
- }
- catch (IllegalAccessException e)
- {
- LOGGER.error(String.format("Error validating AMQP 1.0 object '%s'", constructedObject.toString()), e);
- throw new AmqpErrorException(AmqpError.INTERNAL_ERROR, "Failure during object validation");
- }
- };
- }
-
- private interface CompositeTypeValidator<S>
- {
- void validate(final S constructedObject) throws AmqpErrorException;
+ return _describedTypeConstructor.construct(_describedConstructor.construct(in, handler));
}
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/CompoundTypeConstructor.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/CompoundTypeConstructor.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/CompoundTypeConstructor.java
deleted file mode 100644
index 8d54691..0000000
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/CompoundTypeConstructor.java
+++ /dev/null
@@ -1,193 +0,0 @@
-/*
- *
- * 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.
- *
- */
-package org.apache.qpid.server.protocol.v1_0.codec;
-
-import java.util.ArrayList;
-import java.util.Formatter;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.server.bytebuffer.QpidByteBufferUtils;
-import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
-import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
-import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
-
-public class CompoundTypeConstructor<T> extends VariableWidthTypeConstructor<T>
-{
- private final CompoundTypeAssembler.Factory<T> _assemblerFactory;
-
- public static final CompoundTypeAssembler.Factory LIST_ASSEMBLER_FACTORY =
- new CompoundTypeAssembler.Factory()
- {
-
- @Override
- public CompoundTypeAssembler newInstance()
- {
- return new ListAssembler();
- }
- };
-
-
-
- private static class ListAssembler implements CompoundTypeAssembler<List>
- {
- private List _list;
-
- @Override
- public void init(final int count) throws AmqpErrorException
- {
- _list = new ArrayList(count);
- }
-
- @Override
- public void addItem(final Object obj) throws AmqpErrorException
- {
- _list.add(obj);
- }
-
- @Override
- public List complete() throws AmqpErrorException
- {
- return _list;
- }
-
- @Override
- public String toString()
- {
- return "ListAssembler{" +
- "_list=" + _list +
- '}';
- }
- }
-
-
- public static final CompoundTypeAssembler.Factory MAP_ASSEMBLER_FACTORY =
- new CompoundTypeAssembler.Factory<Map>()
- {
-
- @Override
- public CompoundTypeAssembler<Map> newInstance()
- {
- return new MapAssembler();
- }
- };
-
- private static class MapAssembler implements CompoundTypeAssembler<Map>
- {
- private Map _map;
- private Object _lastKey;
- private static final Object NOT_A_KEY = new Object();
-
-
- @Override
- public void init(final int count) throws AmqpErrorException
- {
- // Can't have an odd number of elements in a map
- if((count & 0x1) == 1)
- {
- Error error = new Error();
- error.setCondition(AmqpError.DECODE_ERROR);
- Formatter formatter = new Formatter();
- formatter.format("map cannot have odd number of elements: %d", count);
- error.setDescription(formatter.toString());
- throw new AmqpErrorException(error);
- }
- _map = new HashMap(count);
- _lastKey = NOT_A_KEY;
- }
-
- @Override
- public void addItem(final Object obj) throws AmqpErrorException
- {
- if(_lastKey != NOT_A_KEY)
- {
- if(_map.put(_lastKey, obj) != null)
- {
- Error error = new Error();
- error.setCondition(AmqpError.DECODE_ERROR);
- Formatter formatter = new Formatter();
- formatter.format("map cannot have duplicate keys: %s has values (%s, %s)", _lastKey, _map.get(_lastKey), obj);
- error.setDescription(formatter.toString());
-
- throw new AmqpErrorException(error);
- }
- _lastKey = NOT_A_KEY;
- }
- else
- {
- _lastKey = obj;
- }
-
- }
-
- @Override
- public Map complete() throws AmqpErrorException
- {
- return _map;
- }
- }
-
-
- public static <X> CompoundTypeConstructor<X> getInstance(int i,
- CompoundTypeAssembler.Factory<X> assemblerFactory)
- {
- return new CompoundTypeConstructor<>(i, assemblerFactory);
- }
-
-
- private CompoundTypeConstructor(int size,
- final CompoundTypeAssembler.Factory<T> assemblerFactory)
- {
- super(size);
- _assemblerFactory = assemblerFactory;
- }
-
- @Override
- public T construct(final List<QpidByteBuffer> in, final ValueHandler handler) throws AmqpErrorException
- {
- int size;
- int count;
-
- if(getSize() == 1)
- {
- size = QpidByteBufferUtils.get(in) & 0xFF;
- count = QpidByteBufferUtils.get(in) & 0xFF;
- }
- else
- {
- size = QpidByteBufferUtils.getInt(in);
- count = QpidByteBufferUtils.getInt(in);
- }
-
- CompoundTypeAssembler<T> assembler = _assemblerFactory.newInstance();
-
- assembler.init(count);
-
- for(int i = 0; i < count; i++)
- {
- assembler.addItem(handler.parse(in));
- }
-
- return assembler.complete();
- }
-}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/ListConstructor.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/ListConstructor.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/ListConstructor.java
new file mode 100644
index 0000000..87011d2
--- /dev/null
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/ListConstructor.java
@@ -0,0 +1,95 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v1_0.codec;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.bytebuffer.QpidByteBufferUtils;
+import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
+import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
+
+public class ListConstructor extends VariableWidthTypeConstructor<List>
+{
+ private ListConstructor(final int size)
+ {
+ super(size);
+ }
+
+ @Override
+ public List construct(final List<QpidByteBuffer> in, final ValueHandler handler) throws AmqpErrorException
+ {
+ int size;
+ int count;
+ long remaining = QpidByteBufferUtils.remaining(in);
+ if (remaining < getSize())
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Not sufficient data for deserialization of 'list'."
+ + " Expected at least %d bytes. Got %d bytes.",
+ getSize(),
+ remaining));
+ }
+
+ if(getSize() == 1)
+ {
+ size = QpidByteBufferUtils.get(in) & 0xFF;
+ count = QpidByteBufferUtils.get(in) & 0xFF;
+ }
+ else
+ {
+ size = QpidByteBufferUtils.getInt(in);
+ count = QpidByteBufferUtils.getInt(in);
+ }
+ remaining -= getSize();
+ if (remaining < size)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Not sufficient data for deserialization of 'list'."
+ + " Expected at least %d bytes. Got %d bytes.",
+ size,
+ remaining));
+ }
+ return construct(in, handler, size, count);
+ }
+
+ protected List construct(final List<QpidByteBuffer> in,
+ final ValueHandler handler,
+ final int size,
+ final int count)
+ throws AmqpErrorException
+ {
+ List list = new ArrayList(count);
+
+ for(int i = 0; i < count; i++)
+ {
+ list.add(handler.parse(in));
+ }
+
+ return list;
+ }
+
+ public static TypeConstructor getInstance(final int size)
+ {
+ return new ListConstructor(size);
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/6a267175/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/MapConstructor.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/MapConstructor.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/MapConstructor.java
new file mode 100644
index 0000000..4c791e8
--- /dev/null
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/codec/MapConstructor.java
@@ -0,0 +1,144 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.qpid.server.protocol.v1_0.codec;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.bytebuffer.QpidByteBufferUtils;
+import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException;
+import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
+
+public class MapConstructor extends VariableWidthTypeConstructor<Map>
+{
+ private MapConstructor(final int size)
+ {
+ super(size);
+ }
+
+ @Override
+ public Map construct(final List<QpidByteBuffer> in, final ValueHandler handler) throws AmqpErrorException
+ {
+ return construct(in, handler, Object.class, Object.class);
+ }
+
+ public <T,S> Map<T, S> construct(final List<QpidByteBuffer> in, final ValueHandler handler,
+ Class<T> keyType,
+ Class<S> valueType) throws AmqpErrorException
+ {
+ int size;
+ int count;
+
+ long remaining = QpidByteBufferUtils.remaining(in);
+ if (remaining < getSize())
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Not sufficient data for deserialization of 'map'."
+ + " Expected at least %d bytes. Got %d bytes.",
+ getSize(),
+ remaining));
+ }
+
+ if(getSize() == 1)
+ {
+ size = QpidByteBufferUtils.get(in) & 0xFF;
+ count = QpidByteBufferUtils.get(in) & 0xFF;
+ }
+ else
+ {
+ size = QpidByteBufferUtils.getInt(in);
+ count = QpidByteBufferUtils.getInt(in);
+ }
+ remaining -= getSize();
+ if (remaining < size)
+ {
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR,
+ String.format("Not sufficient data for deserialization of 'map'."
+ + " Expected at least %d bytes. Got %d bytes.",
+ size,
+ remaining));
+ }
+
+ return construct(in, handler, size, count, keyType, valueType);
+ }
+
+
+ private <T, S> Map<T,S> construct(final List<QpidByteBuffer> in,
+ final ValueHandler handler,
+ final int size,
+ final int count,
+ Class<T> keyType,
+ Class<S> valueType)
+ throws AmqpErrorException
+ {
+
+ // Can't have an odd number of elements in a map
+ if ((count & 0x1) == 1)
+ {
+ String message = String.format("Map cannot have odd number of elements: %d", count);
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR, message);
+ }
+
+ Map<T, S> map = new LinkedHashMap<>(count);
+
+ final int mapSize = count / 2;
+ for(int i = 0; i < mapSize; i++)
+ {
+ Object key = handler.parse(in);
+ if (key != null && !keyType.isAssignableFrom(key.getClass()))
+ {
+ String message = String.format("Expected key type is '%s' but got '%s'",
+ keyType.getSimpleName(),
+ key.getClass().getSimpleName());
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR, message);
+ }
+
+ Object value = handler.parse(in);
+ if (value != null && !valueType.isAssignableFrom(value.getClass()))
+ {
+ String message = String.format("Expected key type is '%s' but got '%s'",
+ valueType.getSimpleName(),
+ value.getClass().getSimpleName());
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR, message);
+ }
+
+ Object oldValue;
+ if ((oldValue = map.put((T)key, (S)value)) != null)
+ {
+ String message = String.format("Map cannot have duplicate keys: %s has values (%s, %s)",
+ key,
+ oldValue,
+ value);
+ throw new AmqpErrorException(AmqpError.DECODE_ERROR, message);
+ }
+
+ }
+ return map;
+ }
+
+
+ public static MapConstructor getInstance(int size)
+ {
+ return new MapConstructor(size);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org