This is an automated email from the ASF dual-hosted git repository.
tabish121 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-protonj2.git
The following commit(s) were added to refs/heads/main by this push:
new bb58aac4 PROTON-3002 Add thread safety measures to the internal
registries
bb58aac4 is described below
commit bb58aac424bf096b61a44f494cac118e71009e3d
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.
---
.../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 fbafdd8d..2a7200d9 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
@@ -895,6 +895,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 512f6a48..7a9804ef 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
@@ -898,6 +898,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 3f1e7df7..5080c441 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
@@ -1034,8 +1037,8 @@ public class ProtonDecoder implements Decoder {
final boolean canCache;
- if (descriptor instanceof Symbol symbol && symbol.getLength() >
UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT) {
- canCache = false;
+ if (descriptor instanceof Symbol symbol) {
+ canCache = symbol.getLength() <=
UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT;
} else {
canCache = descriptor instanceof UnsignedLong;
}
@@ -1049,7 +1052,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 e3e8df82..a99965e7 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];
@@ -1019,8 +1020,8 @@ public final class ProtonStreamDecoder implements
StreamDecoder {
final boolean canCache;
- if (descriptor instanceof Symbol symbol && symbol.getLength() >
UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT) {
- canCache = false;
+ if (descriptor instanceof Symbol symbol) {
+ canCache = symbol.getLength() <=
UNKNOWN_DESCRIBED_TYPE_DESCRIPTOR_SIZE_LIMIT;
} else {
canCache = descriptor instanceof UnsignedLong;
}
@@ -1034,7 +1035,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 8328216f..f8c34136 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]
