This is an automated email from the ASF dual-hosted git repository.
tabish121 pushed a commit to branch 1.x
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
The following commit(s) were added to refs/heads/1.x by this push:
new d31f40ac PROTON-3002 Add thread safety measures to the internal
registries
d31f40ac is described below
commit d31f40acd48a19970c7b12f6f3db34c6d5d770b5
Author: Timothy Bish <[email protected]>
AuthorDate: Wed Nov 4 14:21:22 2026 -0500
PROTON-3002 Add thread safety measures to the internal registries
Ensure that updates to the encoder and decoder described types
registries are thread safe per the contract.
(cherry picked from commit bb58aac424bf096b61a44f494cac118e71009e3d)
---
.../java/org/apache/qpid/protonj2/codec/CodecFactory.java | 8 ++++----
.../main/java/org/apache/qpid/protonj2/codec/Decoder.java | 4 ++++
.../main/java/org/apache/qpid/protonj2/codec/Encoder.java | 4 ++++
.../org/apache/qpid/protonj2/codec/StreamDecoder.java | 4 ++++
.../qpid/protonj2/codec/decoders/ProtonDecoder.java | 15 +++++++++------
.../qpid/protonj2/codec/decoders/ProtonStreamDecoder.java | 9 +++++----
.../qpid/protonj2/codec/encoders/ProtonEncoder.java | 8 +++++---
7 files changed, 35 insertions(+), 17 deletions(-)
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/CodecFactory.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/CodecFactory.java
index c0076dce..a19ba876 100644
--- a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/CodecFactory.java
+++ b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/CodecFactory.java
@@ -25,10 +25,10 @@ import
org.apache.qpid.protonj2.codec.encoders.ProtonEncoderFactory;
*/
public final class CodecFactory {
- private static Encoder amqpTypeEncoder;
- private static Encoder saslTypeEncoder;
- private static Decoder amqpTypeDecoder;
- private static Decoder saslTypeDecoder;
+ private static volatile Encoder amqpTypeEncoder;
+ private static volatile Encoder saslTypeEncoder;
+ private static volatile Decoder amqpTypeDecoder;
+ private static volatile Decoder saslTypeDecoder;
private CodecFactory() {
}
diff --git a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Decoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Decoder.java
index 137a082e..5dc86c7b 100644
--- a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Decoder.java
+++ b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Decoder.java
@@ -876,6 +876,10 @@ public interface Decoder {
/**
* Allows custom {@link DescribedTypeDecoder} instances to be registered
with this {@link Decoder}
* which will be used if the described type encoding is encountered during
decode operations.
+ * <p>
+ * Implementations are not required to treat this method as thread safe
and users should register
+ * type prior to putting a decoder into use in an environment where
multiple threads can access the
+ * same decoder instance.
*
* @param <V> The type that the decoder reads.
*
diff --git a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Encoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Encoder.java
index 82582150..2c3e5fff 100644
--- a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Encoder.java
+++ b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/Encoder.java
@@ -986,6 +986,10 @@ public interface Encoder {
* Register a {@link DescribedTypeEncoder} which can be used when writing
custom types using this
* encoder. When an Object write is performed the type encoder registry
will be consulted in order
* to find the best match for the given {@link Object} instance.
+ * <p>
+ * Implementations are not required to treat this method as thread safe
and users should register
+ * type prior to putting a encoder into use in an environment where
multiple threads can access the
+ * same encoder instance.
*
* @param <V> The type that the encoder handles.
*
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/StreamDecoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/StreamDecoder.java
index 15b70a83..899b192b 100644
--- a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/StreamDecoder.java
+++ b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/StreamDecoder.java
@@ -879,6 +879,10 @@ public interface StreamDecoder {
/**
* Allows custom {@link StreamDescribedTypeDecoder} instances to be
registered with this {@link StreamDecoder}
* which will be used if the described type encoding is encountered during
decode operations.
+ * <p>
+ * Implementations are not required to treat this method as thread safe
and users should register
+ * type prior to putting a decoder into use in an environment where
multiple threads can access the
+ * same decoder instance.
*
* @param <V> The type that the decoder reads.
*
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
index 8f9a7155..f5b0a475 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonDecoder.java
@@ -19,10 +19,10 @@ package org.apache.qpid.protonj2.codec.decoders;
import java.lang.reflect.Array;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;
import org.apache.qpid.protonj2.buffer.ProtonBuffer;
@@ -169,15 +169,18 @@ public class ProtonDecoder implements Decoder {
// Registry of decoders for described types which can be updated with user
defined
// decoders as well as the default decoders.
- private final Map<Object, DescribedTypeDecoder<?>> describedTypeDecoders =
new HashMap<>();
+ private final Map<Object, DescribedTypeDecoder<?>> describedTypeDecoders =
new ConcurrentHashMap<>();
// Registry of decoders for described types which are not registered which
can be updated with a
// limited number of cached decoders to speed up processing
- private final Map<Object, UnknownDescribedTypeDecoder>
unknownDescribedTypeDecoders = new HashMap<>();
+ private final Map<Object, UnknownDescribedTypeDecoder>
unknownDescribedTypeDecoders = new ConcurrentHashMap<>();
// Quick access to decoders that handle AMQP types like Transfer,
Properties etc.
private final DescribedTypeDecoder<?>[] amqpTypeDecoders = new
DescribedTypeDecoder[256];
+ // Provides a mechanism for users to register a decoder for reserved
descriptors, the library
+ // does not allow this normally but some users may wish to decode reserved
types for messages
+ // or other implementation specific use cases.
private final BiFunction<ProtonBuffer, DecoderState, Object>
reservedDescriptorDecoder;
// Internal Decoders used to prevent user to access Proton specific
decoding methods
@@ -1016,8 +1019,8 @@ public class ProtonDecoder implements Decoder {
final boolean canCache;
- if (descriptor instanceof Symbol && ((Symbol) descriptor).getLength()
> UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT) {
- canCache = false;
+ if (descriptor instanceof Symbol) {
+ canCache = ((Symbol) descriptor).getLength() <=
UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT;
} else {
canCache = descriptor instanceof UnsignedLong;
}
@@ -1031,7 +1034,7 @@ public class ProtonDecoder implements Decoder {
};
if (canCache && unknownDescribedTypeDecoders.size() <
UNKNOWN_DESCRIBED_TYPES_CACHE_LIMIT) {
- unknownDescribedTypeDecoders.put(descriptor, typeDecoder);
+ unknownDescribedTypeDecoders.putIfAbsent(descriptor, typeDecoder);
}
return typeDecoder;
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
index 426ccd5c..7dcc05fe 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/decoders/ProtonStreamDecoder.java
@@ -25,6 +25,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.qpid.protonj2.buffer.ProtonBuffer;
import org.apache.qpid.protonj2.codec.DecodeException;
@@ -175,7 +176,7 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
// Registry of decoders for described types which are not registered which
can be updated with a
// limited number of cached decoders to speed up processing
- private final Map<Object, UnknownDescribedTypeDecoder>
unknownDescribedTypeDecoders = new HashMap<>();
+ private final Map<Object, UnknownDescribedTypeDecoder>
unknownDescribedTypeDecoders = new ConcurrentHashMap<>();
// Quick access to decoders that handle AMQP types like Transfer,
Properties etc.
private final StreamDescribedTypeDecoder<?>[] amqpTypeDecoders = new
StreamDescribedTypeDecoder[256];
@@ -1001,8 +1002,8 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
final boolean canCache;
- if (descriptor instanceof Symbol && ((Symbol) descriptor).getLength()
> UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT) {
- canCache = false;
+ if (descriptor instanceof Symbol) {
+ canCache = ((Symbol) descriptor).getLength() <=
UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT;
} else {
canCache = descriptor instanceof UnsignedLong;
}
@@ -1016,7 +1017,7 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
};
if (canCache && unknownDescribedTypeDecoders.size() <
UNKNOWN_DESCRIBED_TYPES_CACHE_LIMIT) {
- unknownDescribedTypeDecoders.put(descriptor, streamTypeDecoder);
+ unknownDescribedTypeDecoders.putIfAbsent(descriptor,
streamTypeDecoder);
}
return streamTypeDecoder;
diff --git
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
index 02f9fa5c..8fecf306 100644
---
a/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
+++
b/protonj2/src/main/java/org/apache/qpid/protonj2/codec/encoders/ProtonEncoder.java
@@ -17,10 +17,10 @@
package org.apache.qpid.protonj2.codec.encoders;
import java.util.Date;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import org.apache.qpid.protonj2.buffer.ProtonBuffer;
@@ -108,7 +108,7 @@ public final class ProtonEncoder implements Encoder {
private ProtonEncoderState singleThreadedState;
- private final Map<Class<?>, TypeEncoder<?>> typeEncoders = new HashMap<>();
+ private final Map<Class<?>, TypeEncoder<?>> typeEncoders = new
ConcurrentHashMap<>();
{
typeEncoders.put(arrayEncoder.getTypeClass(), arrayEncoder);
typeEncoders.put(binaryEncoder.getTypeClass(), binaryEncoder);
@@ -791,7 +791,9 @@ public final class ProtonEncoder implements Encoder {
// Ensure that next time we find the encoder immediately and don't
need to
// go through this process again.
- typeEncoders.put(typeClass, encoder);
+ if (encoder != null) {
+ typeEncoders.put(typeClass, encoder);
+ }
return encoder;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
