You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sd...@apache.org on 2021/12/24 13:47:06 UTC
[ignite-3] branch main updated: IGNITE-15947 Implement class descriptor exchange (#523)
This is an automated email from the ASF dual-hosted git repository.
sdanilov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 30798fa IGNITE-15947 Implement class descriptor exchange (#523)
30798fa is described below
commit 30798fa801e5430631fc55b1fc2ebb9c64622cca
Author: Semyon Danilov <sa...@yandex.ru>
AuthorDate: Fri Dec 24 16:47:01 2021 +0300
IGNITE-15947 Implement class descriptor exchange (#523)
---
.../client/ItMetaStorageRaftGroupTest.java | 5 -
.../client/ItMetaStorageServiceTest.java | 5 -
.../internal/network/processor/TypeUtils.java | 13 ++
.../MessageCollectionItemTypeConverter.java | 2 +-
.../serialization/MessageReaderMethodResolver.java | 7 +
.../serialization/MessageWriterMethodResolver.java | 7 +
.../ignite/network/annotations/Marshallable.java} | 32 ++-
.../network/serialization/MessageReader.java | 10 +
.../network/serialization/MessageWriter.java | 11 +
.../network/netty/ItConnectionManagerTest.java | 3 +-
.../network/recovery/ItRecoveryHandshakeTest.java | 7 +-
.../network/scalecube/ItNodeRestartsTest.java | 6 -
.../scalecube/ItScaleCubeNetworkMessagingTest.java | 6 -
.../ignite/utils/ClusterServiceTestUtils.java | 7 +-
.../internal/network/NetworkMessageTypes.java | 12 ++
.../network/direct/DirectMessageReader.java | 29 ++-
.../network/direct/DirectMessageWriter.java | 26 ++-
.../direct/stream/DirectByteBufferStream.java | 19 ++
.../stream/DirectByteBufferStreamImplV1.java | 110 +++++++++-
.../network/message/ClassDescriptorMessage.java | 86 ++++++++
.../FieldDescriptorMessage.java} | 40 ++--
.../internal/network/netty/ConnectionManager.java | 17 +-
.../internal/network/netty/InboundDecoder.java | 23 ++-
.../ignite/internal/network/netty/NettyClient.java | 19 +-
.../ignite/internal/network/netty/NettyServer.java | 22 +-
.../internal/network/netty/OutboundEncoder.java | 26 +--
.../ClassDescriptorFactoryContext.java | 11 +
.../network/serialization/FieldDescriptor.java | 15 +-
.../PerSessionSerializationService.java | 207 +++++++++++++++++++
...alizationType.java => SerializationResult.java} | 42 ++--
.../serialization/SerializationService.java | 106 ++++++++++
.../network/serialization/SerializationType.java | 18 ++
...lizationType.java => UserObjectSerializer.java} | 44 ++--
.../scalecube/ScaleCubeClusterServiceFactory.java | 6 +-
.../internal/network/netty/InboundDecoderTest.java | 18 +-
.../internal/network/netty/NettyServerTest.java | 7 +-
.../network/serialization/MarshallableTest.java | 229 +++++++++++++++++++++
.../serialization/MessageWithMarshallable.java} | 31 +--
.../serialization/SimpleSerializableObject.java} | 41 ++--
.../apache/ignite/network/TestMessageTypes.java | 2 +
.../apache/ignite/internal/raft/ItLozaTest.java | 5 -
.../service/ItAbstractListenerSnapshotTest.java | 5 -
.../ignite/raft/jraft/core/ItCliServiceTest.java | 117 ++++++-----
.../apache/ignite/raft/jraft/core/ItNodeTest.java | 19 +-
.../ignite/raft/server/RaftServerAbstractTest.java | 5 -
.../apache/ignite/raft/jraft/core/TestCluster.java | 20 +-
.../ignite/raft/jraft/rpc/IgniteRpcTest.java | 35 ++--
.../ItDistributedConfigurationPropertiesTest.java | 2 -
.../ItDistributedConfigurationStorageTest.java | 2 -
.../distributed/ItInternalTableScanTest.java | 5 -
.../distributed/ItTxDistributedTestSingleNode.java | 5 -
51 files changed, 1197 insertions(+), 350 deletions(-)
diff --git a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageRaftGroupTest.java b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageRaftGroupTest.java
index 654f0d7..3f3a386 100644
--- a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageRaftGroupTest.java
+++ b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageRaftGroupTest.java
@@ -50,11 +50,9 @@ import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
@@ -98,8 +96,6 @@ public class ItMetaStorageRaftGroupTest {
/** Network factory. */
private static final TestScaleCubeClusterServiceFactory NETWORK_FACTORY = new TestScaleCubeClusterServiceFactory();
- private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistryImpl();
-
/** Expected server result entry. */
private static final org.apache.ignite.internal.metastorage.server.Entry EXPECTED_SRV_RESULT_ENTRY1 =
new org.apache.ignite.internal.metastorage.server.Entry(
@@ -182,7 +178,6 @@ public class ItMetaStorageRaftGroupTest {
testInfo,
addr.port(),
nodeFinder,
- SERIALIZATION_REGISTRY,
NETWORK_FACTORY
)
)
diff --git a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageServiceTest.java b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageServiceTest.java
index 61a236f..7d9244e 100644
--- a/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageServiceTest.java
+++ b/modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageServiceTest.java
@@ -65,11 +65,9 @@ import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
@@ -109,8 +107,6 @@ public class ItMetaStorageServiceTest {
/** Network factory. */
private static final TestScaleCubeClusterServiceFactory NETWORK_FACTORY = new TestScaleCubeClusterServiceFactory();
- private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistryImpl();
-
/** Expected server result entry. */
private static final org.apache.ignite.internal.metastorage.server.Entry EXPECTED_SRV_RESULT_ENTRY =
new org.apache.ignite.internal.metastorage.server.Entry(
@@ -213,7 +209,6 @@ public class ItMetaStorageServiceTest {
testInfo,
addr.port(),
nodeFinder,
- SERIALIZATION_REGISTRY,
NETWORK_FACTORY
)
)
diff --git a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/TypeUtils.java b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/TypeUtils.java
index ff5fbf6..a64bcfe 100644
--- a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/TypeUtils.java
+++ b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/TypeUtils.java
@@ -60,6 +60,19 @@ public class TypeUtils {
}
/**
+ * Returns {@code true} if the <i>erasure</i> of the first type is a subtype of the second type.
+ *
+ * @param type1 first type (represented by a mirror)
+ * @param type2 second type (represented by a {@code Class})
+ * @return {@code true} if the erasure of the first type is a subtype of the second type, {@code false} otherwise.
+ */
+ public boolean isSubType(TypeMirror type1, Class<?> type2) {
+ TypeMirror type2Mirror = typeMirrorFromClass(type2);
+
+ return types.isSubtype(erasure(type1), erasure(type2Mirror));
+ }
+
+ /**
* Returns the primitive type represented by its boxed value or {@code null} if the given type is not a boxed primitive type.
*
* @param type boxed wrapper of a primitive type
diff --git a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageCollectionItemTypeConverter.java b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageCollectionItemTypeConverter.java
index a0f6d80..d200135 100644
--- a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageCollectionItemTypeConverter.java
+++ b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageCollectionItemTypeConverter.java
@@ -121,7 +121,7 @@ class MessageCollectionItemTypeConverter {
return MessageCollectionItemType.UUID;
} else if (typeUtils.isSameType(parameterType, IgniteUuid.class)) {
return MessageCollectionItemType.IGNITE_UUID;
- } else if (typeUtils.isSameType(parameterType, NetworkMessage.class)) {
+ } else if (typeUtils.isSubType(parameterType, NetworkMessage.class)) {
return MessageCollectionItemType.MSG;
} else if (typeUtils.isSameType(parameterType, BitSet.class)) {
return MessageCollectionItemType.BIT_SET;
diff --git a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageReaderMethodResolver.java b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageReaderMethodResolver.java
index 8d746ae..99e0dc2 100644
--- a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageReaderMethodResolver.java
+++ b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageReaderMethodResolver.java
@@ -24,6 +24,7 @@ import javax.lang.model.element.ExecutableElement;
import javax.lang.model.type.ArrayType;
import javax.lang.model.type.DeclaredType;
import javax.lang.model.type.TypeMirror;
+import org.apache.ignite.network.annotations.Marshallable;
import org.apache.ignite.network.serialization.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -60,6 +61,12 @@ class MessageReaderMethodResolver {
String methodName = methodNameResolver.resolveBaseMethodName(parameterType);
+ if (getter.getAnnotation(Marshallable.class) != null) {
+ return CodeBlock.builder()
+ .add("readMarshallable($S)", parameterName)
+ .build();
+ }
+
switch (methodName) {
case "ObjectArray":
return resolveReadObjectArray((ArrayType) parameterType, parameterName);
diff --git a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageWriterMethodResolver.java b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageWriterMethodResolver.java
index 46eaebe..38aeb5e 100644
--- a/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageWriterMethodResolver.java
+++ b/modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageWriterMethodResolver.java
@@ -28,6 +28,7 @@ import javax.lang.model.type.DeclaredType;
import javax.lang.model.type.TypeKind;
import javax.lang.model.type.TypeMirror;
import org.apache.ignite.internal.network.processor.ProcessingException;
+import org.apache.ignite.network.annotations.Marshallable;
import org.apache.ignite.network.serialization.MessageWriter;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -70,6 +71,12 @@ class MessageWriterMethodResolver {
String parameterName = getter.getSimpleName().toString();
+ if (getter.getAnnotation(Marshallable.class) != null) {
+ return CodeBlock.builder()
+ .add("writeMarshallable($S, message.$L())", parameterName, parameterName)
+ .build();
+ }
+
String methodName = methodNameResolver.resolveBaseMethodName(getterReturnType);
switch (methodName) {
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java b/modules/network-api/src/main/java/org/apache/ignite/network/annotations/Marshallable.java
similarity index 57%
copy from modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
copy to modules/network-api/src/main/java/org/apache/ignite/network/annotations/Marshallable.java
index c5cc87d..184186f 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
+++ b/modules/network-api/src/main/java/org/apache/ignite/network/annotations/Marshallable.java
@@ -15,28 +15,20 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.network.serialization;
+package org.apache.ignite.network.annotations;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
/**
- * Serialization type.
+ * Annotation that should be placed on methods of the {@link Transferable} classes which denote objects not supported by the
+ * direct marshaller. This is useful for the user object serialization because we can't generate serializer and deserializer for
+ * user objects at the compile time.
*/
-public enum SerializationType {
- /** Used for predefined descriptors like primitive (or boxed int). See {@link BuiltinType}. */
- BUILTIN(0),
- /** Type for classes that are neither serializable nor externalizable. */
- ARBITRARY(1),
- /** Externalizable. */
- EXTERNALIZABLE(2),
- /** Serializable (but not Externalizable). */
- SERIALIZABLE(3);
-
- private final int value;
-
- SerializationType(int value) {
- this.value = value;
- }
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Marshallable {
- public int value() {
- return value;
- }
}
diff --git a/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageReader.java b/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageReader.java
index 712f536..2289152 100644
--- a/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageReader.java
+++ b/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageReader.java
@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.UUID;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Marshallable;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
/**
@@ -273,6 +274,15 @@ public interface MessageReader {
MessageCollectionItemType valType, boolean linked);
/**
+ * Reads a field annotated with {@link Marshallable}.
+ *
+ * @param name Field name.
+ * @param <T> Field's type.
+ * @return Marshallable object.
+ */
+ <T> T readMarshallable(String name);
+
+ /**
* Tells whether the last invocation of any of the {@code readXXX(...)} methods has fully written the value. {@code False} is returned
* if there were not enough remaining bytes in a byte buffer.
*
diff --git a/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageWriter.java b/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageWriter.java
index a746eba..7252ac3 100644
--- a/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageWriter.java
+++ b/modules/network-api/src/main/java/org/apache/ignite/network/serialization/MessageWriter.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.UUID;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Marshallable;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
/**
@@ -301,6 +302,16 @@ public interface MessageWriter {
MessageCollectionItemType valType);
/**
+ * Writes a field annotated with {@link Marshallable}.
+ *
+ * @param name Field name.
+ * @param object Marshallable object.
+ * @param <T> Object's type.
+ * @return Whether a value was fully written.
+ */
+ <T> boolean writeMarshallable(String name, T object);
+
+ /**
* Returns {@code true} if the header of the current message has been written, {@code false} otherwise.
*
* @return Whether the message header has already been written.
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java
index ecd9062..8229dbd 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.configuration.testframework.InjectConfiguratio
import org.apache.ignite.internal.network.NetworkMessagesFactory;
import org.apache.ignite.internal.network.recovery.RecoveryClientHandshakeManager;
import org.apache.ignite.internal.network.recovery.RecoveryServerHandshakeManager;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.NettyBootstrapFactory;
@@ -348,7 +349,7 @@ public class ItConnectionManagerTest {
var manager = new ConnectionManager(
cfg,
- registry,
+ new SerializationService(registry, null),
consistentId,
() -> new RecoveryServerHandshakeManager(launchId, consistentId, messageFactory),
() -> new RecoveryClientHandshakeManager(launchId, consistentId, messageFactory),
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/recovery/ItRecoveryHandshakeTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/recovery/ItRecoveryHandshakeTest.java
index f5c4ca8..16d846c 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/recovery/ItRecoveryHandshakeTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/recovery/ItRecoveryHandshakeTest.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.network.NetworkMessagesFactory;
import org.apache.ignite.internal.network.handshake.HandshakeAction;
import org.apache.ignite.internal.network.netty.ConnectionManager;
import org.apache.ignite.internal.network.netty.NettySender;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.TestMessageSerializationRegistryImpl;
@@ -399,6 +400,7 @@ public class ItRecoveryHandshakeTest {
ClientStageFail clientHandshakeFailAt
) {
var registry = new TestMessageSerializationRegistryImpl();
+ var serializationService = new SerializationService(registry, null);
var messageFactory = new NetworkMessagesFactory();
@@ -415,7 +417,7 @@ public class ItRecoveryHandshakeTest {
var manager = new ConnectionManager(
cfg,
- registry,
+ serializationService,
consistentId,
() -> new FailingRecoveryServerHandshakeManager(launchId, consistentId, serverHandshakeFailAt, messageFactory),
() -> new FailingRecoveryClientHandshakeManager(launchId, consistentId, clientHandshakeFailAt, messageFactory),
@@ -437,6 +439,7 @@ public class ItRecoveryHandshakeTest {
*/
private ConnectionManager startManager(int port) {
var registry = new TestMessageSerializationRegistryImpl();
+ var serializationService = new SerializationService(registry, null);
var messageFactory = new NetworkMessagesFactory();
@@ -453,7 +456,7 @@ public class ItRecoveryHandshakeTest {
var manager = new ConnectionManager(
cfg,
- registry,
+ serializationService,
consistentId,
() -> new RecoveryServerHandshakeManager(launchId, consistentId, messageFactory),
() -> new RecoveryClientHandshakeManager(launchId, consistentId, messageFactory),
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItNodeRestartsTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItNodeRestartsTest.java
index c149a09..7e3a668 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItNodeRestartsTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItNodeRestartsTest.java
@@ -28,8 +28,6 @@ import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NodeFinder;
import org.apache.ignite.network.StaticNodeFinder;
-import org.apache.ignite.network.TestMessageSerializationRegistryImpl;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.utils.ClusterServiceTestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
@@ -42,9 +40,6 @@ class ItNodeRestartsTest {
/** Logger. */
private static final IgniteLogger LOG = IgniteLogger.forClass(ItNodeRestartsTest.class);
- /** Serialization registry. */
- private final MessageSerializationRegistry serializationRegistry = new TestMessageSerializationRegistryImpl();
-
/** Network factory. */
private final TestScaleCubeClusterServiceFactory networkFactory = new TestScaleCubeClusterServiceFactory();
@@ -117,7 +112,6 @@ class ItNodeRestartsTest {
testInfo,
addr.port(),
nodeFinder,
- serializationRegistry,
networkFactory
);
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItScaleCubeNetworkMessagingTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItScaleCubeNetworkMessagingTest.java
index 500c43d..bd2c4f8 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItScaleCubeNetworkMessagingTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/network/scalecube/ItScaleCubeNetworkMessagingTest.java
@@ -51,12 +51,10 @@ import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.NodeFinder;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.TestMessage;
-import org.apache.ignite.network.TestMessageSerializationRegistryImpl;
import org.apache.ignite.network.TestMessageTypes;
import org.apache.ignite.network.TestMessagesFactory;
import org.apache.ignite.network.TopologyEventHandler;
import org.apache.ignite.network.annotations.MessageGroup;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.utils.ClusterServiceTestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
@@ -422,9 +420,6 @@ class ItScaleCubeNetworkMessagingTest {
/** Network factory. */
private final TestScaleCubeClusterServiceFactory networkFactory = new TestScaleCubeClusterServiceFactory();
- /** Serialization registry. */
- private final MessageSerializationRegistry serializationRegistry = new TestMessageSerializationRegistryImpl();
-
/** Members of the cluster. */
final List<ClusterService> members;
@@ -471,7 +466,6 @@ class ItScaleCubeNetworkMessagingTest {
testInfo,
addr.port(),
nodeFinder,
- serializationRegistry,
networkFactory
);
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/utils/ClusterServiceTestUtils.java b/modules/network/src/integrationTest/java/org/apache/ignite/utils/ClusterServiceTestUtils.java
index 2992b9b..2bf246c 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/utils/ClusterServiceTestUtils.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/utils/ClusterServiceTestUtils.java
@@ -31,6 +31,7 @@ import org.apache.ignite.internal.configuration.ConfigurationManager;
import org.apache.ignite.internal.configuration.storage.TestConfigurationStorage;
import org.apache.ignite.network.ClusterLocalConfiguration;
import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.MessagingService;
import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.network.NetworkAddress;
@@ -38,7 +39,6 @@ import org.apache.ignite.network.NodeFinder;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.TopologyService;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.junit.jupiter.api.TestInfo;
/**
@@ -53,17 +53,16 @@ public class ClusterServiceTestUtils {
* @param testInfo Test info.
* @param port Local port.
* @param nodeFinder Node finder.
- * @param msgSerializationRegistry Message serialization registry.
* @param clusterSvcFactory Cluster service factory.
*/
public static ClusterService clusterService(
TestInfo testInfo,
int port,
NodeFinder nodeFinder,
- MessageSerializationRegistry msgSerializationRegistry,
TestScaleCubeClusterServiceFactory clusterSvcFactory
) {
- var ctx = new ClusterLocalConfiguration(testNodeName(testInfo, port), msgSerializationRegistry);
+ var messageSerializationRegistry = new MessageSerializationRegistryImpl();
+ var ctx = new ClusterLocalConfiguration(testNodeName(testInfo, port), messageSerializationRegistry);
ConfigurationManager nodeConfigurationMgr = new ConfigurationManager(
Collections.singleton(NetworkConfiguration.KEY),
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/NetworkMessageTypes.java b/modules/network/src/main/java/org/apache/ignite/internal/network/NetworkMessageTypes.java
index 1ae3f04..bf1fd07 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/NetworkMessageTypes.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/NetworkMessageTypes.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.network;
+import org.apache.ignite.internal.network.message.ClassDescriptorMessage;
+import org.apache.ignite.internal.network.message.FieldDescriptorMessage;
import org.apache.ignite.internal.network.message.ScaleCubeMessage;
import org.apache.ignite.internal.network.recovery.message.HandshakeStartMessage;
import org.apache.ignite.internal.network.recovery.message.HandshakeStartResponseMessage;
@@ -41,4 +43,14 @@ public class NetworkMessageTypes {
* Type for {@link HandshakeStartResponseMessage}.
*/
public static final short HANDSHAKE_START_RESPONSE = 3;
+
+ /**
+ * Type for {@link ClassDescriptorMessage}.
+ */
+ public static final short CLASS_DESCRIPTOR_MESSAGE = 4;
+
+ /**
+ * Type for {@link FieldDescriptorMessage}.
+ */
+ public static final short FIELD_DESCRIPTOR_MESSAGE = 5;
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageReader.java b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageReader.java
index f4271a5..6f299a9 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageReader.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageReader.java
@@ -26,10 +26,10 @@ import org.apache.ignite.internal.network.direct.state.DirectMessageState;
import org.apache.ignite.internal.network.direct.state.DirectMessageStateItem;
import org.apache.ignite.internal.network.direct.stream.DirectByteBufferStream;
import org.apache.ignite.internal.network.direct.stream.DirectByteBufferStreamImplV1;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.serialization.MessageReader;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.jetbrains.annotations.Nullable;
@@ -46,11 +46,14 @@ public class DirectMessageReader implements MessageReader {
/**
* Constructor.
*
- * @param serializationRegistry Message serializers.
+ * @param serializationService Serialization service.
* @param protoVer Protocol version.
*/
- public DirectMessageReader(MessageSerializationRegistry serializationRegistry, byte protoVer) {
- state = new DirectMessageState<>(StateItem.class, () -> new StateItem(serializationRegistry, protoVer));
+ public DirectMessageReader(
+ PerSessionSerializationService serializationService,
+ byte protoVer
+ ) {
+ state = new DirectMessageState<>(StateItem.class, () -> new StateItem(serializationService, protoVer));
}
/** {@inheritDoc} */
@@ -383,6 +386,18 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override
+ public <T> T readMarshallable(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
+ T object = stream.readMarshallable(this);
+
+ lastRead = stream.lastFinished();
+
+ return object;
+ }
+
+ /** {@inheritDoc} */
+ @Override
public boolean isLastRead() {
return lastRead;
}
@@ -430,13 +445,13 @@ public class DirectMessageReader implements MessageReader {
/**
* Constructor.
*
- * @param serializationRegistry Serialization registry.
+ * @param serializationService Serialization service.
* @param protoVer Protocol version.
*/
- StateItem(MessageSerializationRegistry serializationRegistry, byte protoVer) {
+ StateItem(PerSessionSerializationService serializationService, byte protoVer) {
switch (protoVer) {
case 1:
- stream = new DirectByteBufferStreamImplV1(serializationRegistry);
+ stream = new DirectByteBufferStreamImplV1(serializationService);
break;
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageWriter.java b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageWriter.java
index 567cbef..4d77358 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageWriter.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/DirectMessageWriter.java
@@ -26,9 +26,9 @@ import org.apache.ignite.internal.network.direct.state.DirectMessageState;
import org.apache.ignite.internal.network.direct.state.DirectMessageStateItem;
import org.apache.ignite.internal.network.direct.stream.DirectByteBufferStream;
import org.apache.ignite.internal.network.direct.stream.DirectByteBufferStreamImplV1;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.NetworkMessage;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.network.serialization.MessageWriter;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.jetbrains.annotations.Nullable;
@@ -43,11 +43,11 @@ public class DirectMessageWriter implements MessageWriter {
/**
* Constructor.
*
- * @param serializationRegistry Serialization registry.
+ * @param serializationService Serialization service.
* @param protoVer Protocol version.
*/
- public DirectMessageWriter(MessageSerializationRegistry serializationRegistry, byte protoVer) {
- state = new DirectMessageState<>(StateItem.class, () -> new StateItem(serializationRegistry, protoVer));
+ public DirectMessageWriter(PerSessionSerializationService serializationService, byte protoVer) {
+ state = new DirectMessageState<>(StateItem.class, () -> new StateItem(serializationService, protoVer));
}
/** {@inheritDoc} */
@@ -387,6 +387,16 @@ public class DirectMessageWriter implements MessageWriter {
state.reset();
}
+ /** {@inheritDoc} */
+ @Override
+ public <T> boolean writeMarshallable(String name, T object) {
+ DirectByteBufferStream stream = state.item().stream;
+
+ stream.writeMarshallable(object, this);
+
+ return stream.lastFinished();
+ }
+
/**
* State item.
*/
@@ -408,13 +418,13 @@ public class DirectMessageWriter implements MessageWriter {
/**
* Constructor.
*
- * @param registry Serialization registry.
- * @param protoVer Protocol version.
+ * @param serializationService Serialization service.
+ * @param protoVer Protocol version.
*/
- StateItem(MessageSerializationRegistry registry, byte protoVer) {
+ StateItem(PerSessionSerializationService serializationService, byte protoVer) {
switch (protoVer) {
case 1:
- stream = new DirectByteBufferStreamImplV1(registry);
+ stream = new DirectByteBufferStreamImplV1(serializationService);
break;
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStream.java b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStream.java
index 401e18b..321f047 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStream.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStream.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.UUID;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Marshallable;
import org.apache.ignite.network.serialization.MessageReader;
import org.apache.ignite.network.serialization.MessageWriter;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -433,4 +434,22 @@ public interface DirectByteBufferStream {
*/
<M extends Map<?, ?>> M readMap(MessageCollectionItemType keyType, MessageCollectionItemType valType,
boolean linked, MessageReader reader);
+
+ /**
+ * Writes a field annotated with {@link Marshallable}.
+ *
+ * @param object Marshallable object.
+ * @param writer Writer.
+ * @param <T> Object's type.
+ */
+ <T> void writeMarshallable(T object, MessageWriter writer);
+
+ /**
+ * Reads a field annotated with {@link Marshallable}.
+ *
+ * @param reader Reader.
+ * @param <T> Field's type.
+ * @return Marshallable object.
+ */
+ <T> T readMarshallable(MessageReader reader);
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java
index 0093d6d..0122280 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java
@@ -44,6 +44,9 @@ import java.util.List;
import java.util.Map;
import java.util.RandomAccess;
import java.util.UUID;
+import org.apache.ignite.internal.network.message.ClassDescriptorMessage;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
+import org.apache.ignite.internal.network.serialization.SerializationResult;
import org.apache.ignite.internal.util.ArrayFactory;
import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.IgniteUtils;
@@ -51,7 +54,6 @@ import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.serialization.MessageDeserializer;
import org.apache.ignite.network.serialization.MessageReader;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.network.serialization.MessageSerializer;
import org.apache.ignite.network.serialization.MessageWriter;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -64,7 +66,8 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
/** Poison object. */
private static final Object NULL = new Object();
- private final MessageSerializationRegistry serializationRegistry;
+ /** Message serialization registry. */
+ private final PerSessionSerializationService serializationService;
private ByteBuffer buf;
@@ -100,6 +103,9 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
@Nullable
private MessageDeserializer<NetworkMessage> msgDeserializer;
+ @Nullable
+ private MessageSerializer<NetworkMessage> msgSerializer;
+
private Iterator<?> mapIt;
private Iterator<?> it;
@@ -136,6 +142,12 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
private long uuidLocId;
+ private int marshallableState;
+
+ private byte[] marshallable;
+
+ private List<ClassDescriptorMessage> descriptors;
+
protected boolean lastFinished;
/** byte-array representation of string. */
@@ -144,10 +156,10 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
/**
* Constructor.
*
- * @param serializationRegistry Message mappers.
+ * @param serializationService Serialization service. .
*/
- public DirectByteBufferStreamImplV1(MessageSerializationRegistry serializationRegistry) {
- this.serializationRegistry = serializationRegistry;
+ public DirectByteBufferStreamImplV1(PerSessionSerializationService serializationService) {
+ this.serializationService = serializationService;
}
/** {@inheritDoc} */
@@ -589,12 +601,17 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
writer.setCurrentWriteClass(msg.getClass());
- MessageSerializer<NetworkMessage> serializer =
- serializationRegistry.createSerializer(msg.groupType(), msg.messageType());
+ if (msgSerializer == null) {
+ msgSerializer = serializationService.createMessageSerializer(msg.groupType(), msg.messageType());
+ }
writer.setBuffer(buf);
- lastFinished = serializer.writeMessage(msg, writer);
+ lastFinished = msgSerializer.writeMessage(msg, writer);
+
+ if (lastFinished) {
+ msgSerializer = null;
+ }
} finally {
writer.afterInnerMessageWrite(lastFinished);
}
@@ -1172,7 +1189,7 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
return null;
}
- msgDeserializer = serializationRegistry.createDeserializer(msgGroupType, msgType);
+ msgDeserializer = serializationService.createMessageDeserializer(msgGroupType, msgType);
}
// if the deserializer is not null then we have definitely finished parsing the header and can read the message
@@ -1343,6 +1360,81 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
return map0;
}
+ /** {@inheritDoc} */
+ @Override
+ public <T> void writeMarshallable(T object, MessageWriter writer) {
+ switch (marshallableState) {
+ case 0:
+ if (marshallable == null) {
+ // If object was not serialized to a byte array, serialize it
+ SerializationResult res = serializationService.writeMarshallable(object);
+ List<Integer> descriptorIds = res.ids();
+ marshallable = res.array();
+ // Get descriptors that were not previously sent to the remote node
+ descriptors = serializationService.createClassDescriptorsMessages(descriptorIds);
+ }
+
+ writeCollection(descriptors, MessageCollectionItemType.MSG, writer);
+
+ if (!lastFinished) {
+ return;
+ }
+
+ marshallableState++;
+
+ //noinspection fallthrough
+ case 1:
+ writeByteArray(marshallable);
+
+ if (!lastFinished) {
+ return;
+ }
+
+ marshallable = null;
+ descriptors = null;
+ marshallableState = 0;
+ break;
+
+ default:
+ throw new IllegalArgumentException("Unknown marshallableState: " + marshallableState);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public <T> T readMarshallable(MessageReader reader) {
+ switch (marshallableState) {
+ case 0:
+ descriptors = readCollection(MessageCollectionItemType.MSG, reader);
+
+ if (!lastFinished) {
+ return null;
+ }
+
+ marshallableState++;
+
+ //noinspection fallthrough
+ case 1:
+ marshallable = readByteArray();
+
+ if (!lastFinished) {
+ return null;
+ }
+
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown marshallableState: " + marshallableState);
+ }
+
+ T read = serializationService.readMarshallable(descriptors, marshallable);
+
+ marshallableState = 0;
+ marshallable = null;
+ descriptors = null;
+
+ return read;
+ }
+
/**
* Writes array.
*
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/message/ClassDescriptorMessage.java b/modules/network/src/main/java/org/apache/ignite/internal/network/message/ClassDescriptorMessage.java
new file mode 100644
index 0000000..8cf7745
--- /dev/null
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/message/ClassDescriptorMessage.java
@@ -0,0 +1,86 @@
+/*
+ * 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.ignite.internal.network.message;
+
+import java.util.Collection;
+import org.apache.ignite.internal.network.NetworkMessageTypes;
+import org.apache.ignite.internal.network.serialization.ClassDescriptor;
+import org.apache.ignite.internal.network.serialization.Serialization;
+import org.apache.ignite.internal.network.serialization.SerializationType;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Transferable;
+
+/** Message for the {@link ClassDescriptor}. */
+@Transferable(NetworkMessageTypes.CLASS_DESCRIPTOR_MESSAGE)
+public interface ClassDescriptorMessage extends NetworkMessage {
+ /**
+ * Class name.
+ *
+ * @see ClassDescriptor#className()
+ */
+ String className();
+
+ /**
+ * Descriptor id.
+ *
+ * @see ClassDescriptor#descriptorId()
+ */
+ int descriptorId();
+
+ /**
+ * List of the class fields' descriptors.
+ *
+ * @see ClassDescriptor#fields()
+ */
+ Collection<FieldDescriptorMessage> fields();
+
+ /**
+ * The type of the serialization mechanism for the class.
+ *
+ * @see SerializationType#value()
+ */
+ int serializationType();
+
+ /**
+ * Has serialization override.
+ *
+ * @see Serialization#hasSerializationOverride()
+ */
+ boolean hasSerializationOverride();
+
+ /**
+ * Has writeReplace.
+ *
+ * @see Serialization#hasWriteReplace()
+ */
+ boolean hasWriteReplace();
+
+ /**
+ * Has readResolve.
+ *
+ * @see Serialization#hasReadResolve()
+ */
+ boolean hasReadResolve();
+
+ /**
+ * Whether the class is final.
+ *
+ * @see ClassDescriptor#isFinal()
+ */
+ boolean isFinal();
+}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java b/modules/network/src/main/java/org/apache/ignite/internal/network/message/FieldDescriptorMessage.java
similarity index 54%
copy from modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
copy to modules/network/src/main/java/org/apache/ignite/internal/network/message/FieldDescriptorMessage.java
index c5cc87d..aa7a3ec 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/message/FieldDescriptorMessage.java
@@ -15,28 +15,28 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.network.serialization;
+package org.apache.ignite.internal.network.message;
-/**
- * Serialization type.
- */
-public enum SerializationType {
- /** Used for predefined descriptors like primitive (or boxed int). See {@link BuiltinType}. */
- BUILTIN(0),
- /** Type for classes that are neither serializable nor externalizable. */
- ARBITRARY(1),
- /** Externalizable. */
- EXTERNALIZABLE(2),
- /** Serializable (but not Externalizable). */
- SERIALIZABLE(3);
+import org.apache.ignite.internal.network.NetworkMessageTypes;
+import org.apache.ignite.internal.network.serialization.FieldDescriptor;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Transferable;
- private final int value;
+/** Message for the {@link FieldDescriptor}. */
+@Transferable(NetworkMessageTypes.FIELD_DESCRIPTOR_MESSAGE)
+public interface FieldDescriptorMessage extends NetworkMessage {
+ /**
+ * Name of the field.
+ */
+ String name();
- SerializationType(int value) {
- this.value = value;
- }
+ /**
+ * Field type's descriptor id.
+ */
+ int typeDescriptorId();
- public int value() {
- return value;
- }
+ /**
+ * Field's class name.
+ */
+ String className();
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
index 87a0c4f..9719e23 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
@@ -39,11 +39,11 @@ import java.util.stream.Stream;
import org.apache.ignite.configuration.schemas.network.NetworkView;
import org.apache.ignite.configuration.schemas.network.OutboundView;
import org.apache.ignite.internal.network.handshake.HandshakeManager;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.network.NetworkMessage;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -69,8 +69,8 @@ public class ConnectionManager {
/** Clients. */
private final Map<SocketAddress, NettyClient> clients = new ConcurrentHashMap<>();
- /** Serialization registry. */
- private final MessageSerializationRegistry serializationRegistry;
+ /** Serialization service. */
+ private final SerializationService serializationService;
/** Message listeners. */
private final List<BiConsumer<SocketAddress, NetworkMessage>> listeners = new CopyOnWriteArrayList<>();
@@ -91,7 +91,7 @@ public class ConnectionManager {
* Constructor.
*
* @param networkConfiguration Network configuration.
- * @param registry Serialization registry.
+ * @param serializationService Serialization service.
* @param consistentId Consistent id of this node.
* @param serverHandshakeManagerFactory Server handshake manager factory.
* @param clientHandshakeManagerFactory Client handshake manager factory.
@@ -99,23 +99,22 @@ public class ConnectionManager {
*/
public ConnectionManager(
NetworkView networkConfiguration,
- MessageSerializationRegistry registry,
+ SerializationService serializationService,
String consistentId,
Supplier<HandshakeManager> serverHandshakeManagerFactory,
Supplier<HandshakeManager> clientHandshakeManagerFactory,
NettyBootstrapFactory bootstrapFactory
) {
- this.serializationRegistry = registry;
+ this.serializationService = serializationService;
this.consistentId = consistentId;
this.clientHandshakeManagerFactory = clientHandshakeManagerFactory;
this.server = new NettyServer(
- consistentId,
networkConfiguration,
serverHandshakeManagerFactory,
this::onNewIncomingChannel,
this::onMessage,
- serializationRegistry,
+ serializationService,
bootstrapFactory
);
@@ -227,7 +226,7 @@ public class ConnectionManager {
private NettyClient connect(SocketAddress address) {
var client = new NettyClient(
address,
- serializationRegistry,
+ serializationService,
clientHandshakeManagerFactory.get(),
this::onMessage
);
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InboundDecoder.java b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InboundDecoder.java
index 1f1c3ba..49c2ed0 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InboundDecoder.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/InboundDecoder.java
@@ -26,11 +26,11 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.ignite.internal.network.direct.DirectMarshallingUtils;
import org.apache.ignite.internal.network.direct.DirectMessageReader;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.serialization.MessageDeserializer;
import org.apache.ignite.network.serialization.MessageReader;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
/**
* Decodes {@link ByteBuf}s into {@link NetworkMessage}s.
@@ -45,16 +45,16 @@ public class InboundDecoder extends ByteToMessageDecoder {
/** Message deserializer channel attribute key. */
private static final AttributeKey<MessageDeserializer<NetworkMessage>> DESERIALIZER_KEY = AttributeKey.valueOf("DESERIALIZER");
- /** Serialization registry. */
- private final MessageSerializationRegistry serializationRegistry;
+ /** Serialization service. */
+ private final PerSessionSerializationService serializationService;
/**
* Constructor.
*
- * @param serializationRegistry Serialization registry.
+ * @param serializationService Serialization service.
*/
- public InboundDecoder(MessageSerializationRegistry serializationRegistry) {
- this.serializationRegistry = serializationRegistry;
+ public InboundDecoder(PerSessionSerializationService serializationService) {
+ this.serializationService = serializationService;
}
/** {@inheritDoc} */
@@ -66,7 +66,7 @@ public class InboundDecoder extends ByteToMessageDecoder {
MessageReader reader = readerAttr.get();
if (reader == null) {
- reader = new DirectMessageReader(serializationRegistry, ConnectionManager.DIRECT_PROTOCOL_VERSION);
+ reader = new DirectMessageReader(serializationService, ConnectionManager.DIRECT_PROTOCOL_VERSION);
readerAttr.set(reader);
}
@@ -81,7 +81,7 @@ public class InboundDecoder extends ByteToMessageDecoder {
// Read message type.
if (msg == null) {
if (buffer.remaining() >= NetworkMessage.MSG_TYPE_SIZE_BYTES) {
- msg = serializationRegistry.createDeserializer(DirectMarshallingUtils.getShort(buffer),
+ msg = serializationService.createMessageDeserializer(DirectMarshallingUtils.getShort(buffer),
DirectMarshallingUtils.getShort(buffer));
} else {
break;
@@ -111,6 +111,13 @@ public class InboundDecoder extends ByteToMessageDecoder {
} else {
messageAttr.set(msg);
}
+
+ if (readBytes == 0) {
+ // Zero bytes were read from the buffer.
+ // This could've happened because for some types (like floats) we have to read N bytes at a time and there could be
+ // less than N bytes available.
+ break;
+ }
} catch (Throwable e) {
if (LOG.isDebugEnabled()) {
LOG.debug(
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyClient.java b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyClient.java
index 2c4d28a..6a00747 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyClient.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyClient.java
@@ -28,9 +28,10 @@ import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.apache.ignite.internal.network.handshake.HandshakeManager;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.NetworkMessage;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.jetbrains.annotations.Nullable;
/**
@@ -40,8 +41,8 @@ public class NettyClient {
/** A lock for start and stop operations. */
private final Object startStopLock = new Object();
- /** Serialization registry. */
- private final MessageSerializationRegistry serializationRegistry;
+ /** Serialization service. */
+ private final SerializationService serializationService;
/** Destination address. */
private final SocketAddress address;
@@ -70,18 +71,18 @@ public class NettyClient {
* Constructor.
*
* @param address Destination address.
- * @param serializationRegistry Serialization registry.
+ * @param serializationService Serialization service.
* @param manager Client handshake manager.
* @param messageListener Message listener.
*/
public NettyClient(
SocketAddress address,
- MessageSerializationRegistry serializationRegistry,
+ SerializationService serializationService,
HandshakeManager manager,
BiConsumer<SocketAddress, NetworkMessage> messageListener
) {
this.address = address;
- this.serializationRegistry = serializationRegistry;
+ this.serializationService = serializationService;
this.handshakeManager = manager;
this.messageListener = messageListener;
}
@@ -108,12 +109,14 @@ public class NettyClient {
/** {@inheritDoc} */
@Override
public void initChannel(SocketChannel ch) {
+ var sessionSerializationService = new PerSessionSerializationService(serializationService);
+
ch.pipeline().addLast(
- new InboundDecoder(serializationRegistry),
+ new InboundDecoder(sessionSerializationService),
new HandshakeHandler(handshakeManager),
new MessageHandler(messageListener),
new ChunkedWriteHandler(),
- new OutboundEncoder(serializationRegistry),
+ new OutboundEncoder(sessionSerializationService),
new IoExceptionSuppressingHandler()
);
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyServer.java b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyServer.java
index d4e48d1..785ed37 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyServer.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/NettyServer.java
@@ -33,10 +33,11 @@ import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.ignite.configuration.schemas.network.NetworkView;
import org.apache.ignite.internal.network.handshake.HandshakeManager;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.network.NetworkMessage;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -53,8 +54,8 @@ public class NettyServer {
/** Server socket configuration. */
private final NetworkView configuration;
- /** Serialization registry. */
- private final MessageSerializationRegistry serializationRegistry;
+ /** Serialization service. */
+ private final SerializationService serializationService;
/** Incoming message listener. */
private final BiConsumer<SocketAddress, NetworkMessage> messageListener;
@@ -82,27 +83,26 @@ public class NettyServer {
/**
* Constructor.
*
- * @param consistentId Consistent id.
* @param configuration Server configuration.
* @param handshakeManager Handshake manager supplier.
* @param newConnectionListener New connections listener.
* @param messageListener Message listener.
- * @param serializationRegistry Serialization registry.
+ * @param serializationService Serialization service.
+ * @param bootstrapFactory Netty bootstrap factory.
*/
public NettyServer(
- String consistentId,
NetworkView configuration,
Supplier<HandshakeManager> handshakeManager,
Consumer<NettySender> newConnectionListener,
BiConsumer<SocketAddress, NetworkMessage> messageListener,
- MessageSerializationRegistry serializationRegistry,
+ SerializationService serializationService,
NettyBootstrapFactory bootstrapFactory
) {
this.configuration = configuration;
this.handshakeManager = handshakeManager;
this.newConnectionListener = newConnectionListener;
this.messageListener = messageListener;
- this.serializationRegistry = serializationRegistry;
+ this.serializationService = serializationService;
this.bootstrapFactory = bootstrapFactory;
}
@@ -127,6 +127,8 @@ public class NettyServer {
/** {@inheritDoc} */
@Override
public void initChannel(SocketChannel ch) {
+ var sessionSerializationService = new PerSessionSerializationService(serializationService);
+
// Get handshake manager for the new channel.
HandshakeManager manager = handshakeManager.get();
@@ -135,7 +137,7 @@ public class NettyServer {
* Decoder that uses the MessageReader
* to read chunked data.
*/
- new InboundDecoder(serializationRegistry),
+ new InboundDecoder(sessionSerializationService),
// Handshake handler.
new HandshakeHandler(manager),
// Handles decoded NetworkMessages.
@@ -146,7 +148,7 @@ public class NettyServer {
*/
new ChunkedWriteHandler(),
// Converts NetworkMessage to a ChunkedNetworkMessageInput
- new OutboundEncoder(serializationRegistry),
+ new OutboundEncoder(sessionSerializationService),
new IoExceptionSuppressingHandler()
);
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/OutboundEncoder.java b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/OutboundEncoder.java
index ddcf5dd..c629084 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/OutboundEncoder.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/OutboundEncoder.java
@@ -25,8 +25,8 @@ import io.netty.handler.stream.ChunkedInput;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.ignite.internal.network.direct.DirectMessageWriter;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
import org.apache.ignite.network.NetworkMessage;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.network.serialization.MessageSerializer;
/**
@@ -34,24 +34,21 @@ import org.apache.ignite.network.serialization.MessageSerializer;
*/
public class OutboundEncoder extends MessageToMessageEncoder<NetworkMessage> {
/** Serialization registry. */
- private final MessageSerializationRegistry serializationRegistry;
+ private final PerSessionSerializationService serializationService;
/**
* Constructor.
*
- * @param registry Serialization registry.
+ * @param serializationService Serialization service.
*/
- public OutboundEncoder(MessageSerializationRegistry registry) {
- serializationRegistry = registry;
+ public OutboundEncoder(PerSessionSerializationService serializationService) {
+ this.serializationService = serializationService;
}
/** {@inheritDoc} */
@Override
protected void encode(ChannelHandlerContext ctx, NetworkMessage msg, List<Object> out) throws Exception {
- MessageSerializer<NetworkMessage> serializer =
- serializationRegistry.createSerializer(msg.groupType(), msg.messageType());
-
- out.add(new NetworkMessageChunkedInput(msg, serializer, serializationRegistry));
+ out.add(new NetworkMessageChunkedInput(msg, serializationService));
}
/**
@@ -73,17 +70,16 @@ public class OutboundEncoder extends MessageToMessageEncoder<NetworkMessage> {
/**
* Constructor.
*
- * @param msg Network message.
- * @param serializer Serializer.
+ * @param msg Network message.
+ * @param serializationService Serialization service.
*/
private NetworkMessageChunkedInput(
NetworkMessage msg,
- MessageSerializer<NetworkMessage> serializer,
- MessageSerializationRegistry registry
+ PerSessionSerializationService serializationService
) {
this.msg = msg;
- this.serializer = serializer;
- this.writer = new DirectMessageWriter(registry, ConnectionManager.DIRECT_PROTOCOL_VERSION);
+ this.serializer = serializationService.createMessageSerializer(msg.groupType(), msg.messageType());
+ this.writer = new DirectMessageWriter(serializationService, ConnectionManager.DIRECT_PROTOCOL_VERSION);
}
/** {@inheritDoc} */
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/ClassDescriptorFactoryContext.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/ClassDescriptorFactoryContext.java
index 43b148c..a327407 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/ClassDescriptorFactoryContext.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/ClassDescriptorFactoryContext.java
@@ -144,4 +144,15 @@ public class ClassDescriptorFactoryContext {
descriptorMap.put(realDescriptorId, descriptor);
}
+
+ /**
+ * Returns {@code true} if descriptor with the specified descriptor id is built-in, {@code false} otherwise.
+ *
+ *
+ * @param descriptorId Descriptor id.
+ * @return Whether descriptor is built-in.
+ */
+ public static boolean isBuiltIn(int descriptorId) {
+ return descriptorId < DEFAULT_DESCRIPTORS_OFFSET_COUNT;
+ }
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/FieldDescriptor.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/FieldDescriptor.java
index 6e4e722..ff72f20 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/FieldDescriptor.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/FieldDescriptor.java
@@ -45,8 +45,19 @@ public class FieldDescriptor {
* Constructor.
*/
public FieldDescriptor(@NotNull Field field, int typeDescriptorId) {
- this.name = field.getName();
- this.clazz = field.getType();
+ this(field.getName(), field.getType(), typeDescriptorId);
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param fieldName .
+ * @param fieldClazz .
+ * @param typeDescriptorId .
+ */
+ public FieldDescriptor(@NotNull String fieldName, @NotNull Class<?> fieldClazz, int typeDescriptorId) {
+ this.name = fieldName;
+ this.clazz = fieldClazz;
this.typeDescriptorId = typeDescriptorId;
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/PerSessionSerializationService.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/PerSessionSerializationService.java
new file mode 100644
index 0000000..751ed5d
--- /dev/null
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/PerSessionSerializationService.java
@@ -0,0 +1,207 @@
+/*
+ * 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.ignite.internal.network.serialization;
+
+import static java.util.stream.Collectors.toList;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.ignite.internal.network.NetworkMessagesFactory;
+import org.apache.ignite.internal.network.message.ClassDescriptorMessage;
+import org.apache.ignite.internal.network.message.FieldDescriptorMessage;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.serialization.MessageDeserializer;
+import org.apache.ignite.network.serialization.MessageSerializer;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.TestOnly;
+
+/**
+ * Per-session serialization service.
+ * Handles (de-)serialization of messages, object (de-)serialization and class descriptor merging.
+ */
+public class PerSessionSerializationService {
+ /** Network messages factory. */
+ private static final NetworkMessagesFactory MSG_FACTORY = new NetworkMessagesFactory();
+
+ /** Global serialization service. */
+ @NotNull
+ private final SerializationService serializationService;
+
+ /**
+ * Map with merged class descriptors. They are the result of the merging of a local and a remote descriptor.
+ * The key in this map is a <b>remote</b> descriptor id.
+ */
+ private final ConcurrentMap<Integer, ClassDescriptor> mergedDescriptorMap = new ConcurrentHashMap<>();
+
+ /**
+ * Immutable view over {@link #mergedDescriptorMap}. Used by {@link #serializationService}.
+ */
+ private final Map<Integer, ClassDescriptor> descriptorMapView = Collections.unmodifiableMap(mergedDescriptorMap);
+
+ public PerSessionSerializationService(@NotNull SerializationService serializationService) {
+ this.serializationService = serializationService;
+ }
+
+ /**
+ * Creates a message serializer.
+ *
+ * @see SerializationService#createSerializer(short, short)
+ */
+ public <T extends NetworkMessage> MessageSerializer<T> createMessageSerializer(short groupType, short messageType) {
+ return serializationService.createSerializer(groupType, messageType);
+ }
+
+ /**
+ * Creates a message deserializer.
+ *
+ * @see SerializationService#createDeserializer(short, short)
+ */
+ public <T extends NetworkMessage> MessageDeserializer<T> createMessageDeserializer(short groupType, short messageType) {
+ return serializationService.createDeserializer(groupType, messageType);
+ }
+
+ /**
+ * Serializes a marshallable object to a byte array.
+ *
+ * @see SerializationService#writeMarshallable(Object)
+ */
+ public <T> SerializationResult writeMarshallable(T marshallable) {
+ return serializationService.writeMarshallable(marshallable);
+ }
+
+ /**
+ * Deserializes a marshallable object from a byte array.
+ *
+ * @see SerializationService#readMarshallable(Map, byte[])
+ */
+ public <T> T readMarshallable(List<ClassDescriptorMessage> missingDescriptors, byte[] marshallableData) {
+ mergeDescriptors(missingDescriptors);
+
+ return serializationService.readMarshallable(descriptorMapView, marshallableData);
+ }
+
+ /**
+ * Creates a list of messages holding class descriptors.
+ *
+ * @param descriptorIds Ids of class descriptors.
+ * @return List of class descriptor network messages.
+ */
+ public List<ClassDescriptorMessage> createClassDescriptorsMessages(List<Integer> descriptorIds) {
+ return descriptorIds.stream().map(id -> {
+ ClassDescriptor descriptor = serializationService.getClassDescriptor(id);
+
+ List<FieldDescriptorMessage> fields = descriptor.fields().stream()
+ .map(d -> {
+ return MSG_FACTORY.fieldDescriptorMessage()
+ .name(d.name())
+ .typeDescriptorId(d.typeDescriptorId())
+ .className(d.clazz().getName())
+ .build();
+ })
+ .collect(toList());
+
+ Serialization serialization = descriptor.serialization();
+
+ return MSG_FACTORY.classDescriptorMessage()
+ .fields(fields)
+ .isFinal(descriptor.isFinal())
+ .serializationType(serialization.type().value())
+ .hasSerializationOverride(serialization.hasSerializationOverride())
+ .hasWriteReplace(serialization.hasWriteReplace())
+ .hasReadResolve(serialization.hasReadResolve())
+ .descriptorId(descriptor.descriptorId())
+ .className(descriptor.className())
+ .build();
+ }).collect(toList());
+ }
+
+ private void mergeDescriptors(List<ClassDescriptorMessage> remoteDescriptors) {
+ for (ClassDescriptorMessage clsMsg : remoteDescriptors) {
+ int clsDescriptorId = clsMsg.descriptorId();
+
+ boolean isClsBuiltin = serializationService.isBuiltIn(clsDescriptorId);
+
+ if (isClsBuiltin) {
+ continue;
+ }
+
+ if (mergedDescriptorMap.containsKey(clsDescriptorId)) {
+ continue;
+ }
+
+ mergedDescriptorMap.put(clsDescriptorId, messageToMergedClassDescriptor(clsMsg));
+ }
+ }
+
+ /**
+ * Converts {@link ClassDescriptorMessage} to a {@link ClassDescriptor} and merges it with a local {@link ClassDescriptor} of the
+ * same class.
+ *
+ * @param clsMsg ClassDescriptorMessage.
+ * @return Merged class descriptor.
+ */
+ @NotNull
+ private ClassDescriptor messageToMergedClassDescriptor(ClassDescriptorMessage clsMsg) {
+ ClassDescriptor localDescriptor = serializationService.getClassDescriptor(clsMsg.className());
+
+ List<FieldDescriptor> remoteFields = clsMsg.fields().stream().map(fieldMsg -> {
+ int typeDescriptorId = fieldMsg.typeDescriptorId();
+
+ return new FieldDescriptor(fieldMsg.name(), getClass(typeDescriptorId, fieldMsg.className()), typeDescriptorId);
+ }).collect(toList());
+
+ SerializationType serializationType = SerializationType.getByValue(clsMsg.serializationType());
+
+ var serialization = new Serialization(
+ serializationType,
+ clsMsg.hasSerializationOverride(),
+ clsMsg.hasWriteReplace(),
+ clsMsg.hasReadResolve()
+ );
+
+ ClassDescriptor remoteDescriptor = new ClassDescriptor(
+ localDescriptor.clazz(),
+ clsMsg.descriptorId(),
+ remoteFields,
+ serialization
+ );
+
+ return mergeDescriptor(localDescriptor, remoteDescriptor);
+ }
+
+ private ClassDescriptor mergeDescriptor(ClassDescriptor localDescriptor, ClassDescriptor remoteDescriptor) {
+ // TODO: IGNITE-15948 Handle class structure changes
+ return remoteDescriptor;
+ }
+
+ private Class<?> getClass(int descriptorId, String typeName) {
+ if (serializationService.isBuiltIn(descriptorId)) {
+ return serializationService.getClassDescriptor(descriptorId).clazz();
+ } else {
+ return serializationService.getClassDescriptor(typeName).clazz();
+ }
+ }
+
+ @TestOnly
+ Map<Integer, ClassDescriptor> getDescriptorMapView() {
+ return descriptorMapView;
+ }
+}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationResult.java
similarity index 51%
copy from modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
copy to modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationResult.java
index c5cc87d..20d1daa 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationResult.java
@@ -17,26 +17,38 @@
package org.apache.ignite.internal.network.serialization;
+import java.util.List;
+
/**
- * Serialization type.
+ * Class that holds a byte array of the serialized object and a list of type descriptor ids with all the descriptors that were
+ * used during the serialization.
+ * TODO: IGNITE-16081 It's a temporary class that will be removed.
*/
-public enum SerializationType {
- /** Used for predefined descriptors like primitive (or boxed int). See {@link BuiltinType}. */
- BUILTIN(0),
- /** Type for classes that are neither serializable nor externalizable. */
- ARBITRARY(1),
- /** Externalizable. */
- EXTERNALIZABLE(2),
- /** Serializable (but not Externalizable). */
- SERIALIZABLE(3);
+public class SerializationResult {
+ /** Serialized object. */
+ private final byte[] array;
+
+ /** Type descriptors. */
+ private final List<Integer> ids;
- private final int value;
+ /**
+ * Constructor.
+ *
+ * @param array Serialized object.
+ * @param ids Type descriptors.
+ */
+ public SerializationResult(byte[] array, List<Integer> ids) {
+ this.array = array;
+ this.ids = ids;
+ }
- SerializationType(int value) {
- this.value = value;
+ /** Gets serialized object. */
+ public byte[] array() {
+ return array;
}
- public int value() {
- return value;
+ /** Gets descriptor ids. */
+ public List<Integer> ids() {
+ return ids;
}
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationService.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationService.java
new file mode 100644
index 0000000..fdfb86d
--- /dev/null
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationService.java
@@ -0,0 +1,106 @@
+/*
+ * 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.ignite.internal.network.serialization;
+
+import java.util.Map;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.serialization.MessageDeserializer;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.network.serialization.MessageSerializer;
+
+/**
+ * Serialization service implementation.
+ */
+public class SerializationService {
+ /** Message serialization registry. */
+ private final MessageSerializationRegistry messageRegistry;
+
+ /** User object serializer. */
+ private final UserObjectSerializer userObjectSerializer;
+
+ public SerializationService(MessageSerializationRegistry messageRegistry, UserObjectSerializer userObjectSerializer) {
+ this.messageRegistry = messageRegistry;
+ this.userObjectSerializer = userObjectSerializer;
+ }
+
+ /**
+ * Creates a message serializer.
+ *
+ * @see MessageSerializationRegistry#createSerializer(short, short)
+ */
+ public <T extends NetworkMessage> MessageSerializer<T> createSerializer(short groupType, short messageType) {
+ return messageRegistry.createSerializer(groupType, messageType);
+ }
+
+ /**
+ * Creates a message deserializer.
+ *
+ * @see MessageSerializationRegistry#createDeserializer(short, short)
+ */
+ public <T extends NetworkMessage> MessageDeserializer<T> createDeserializer(short groupType, short messageType) {
+ return messageRegistry.createDeserializer(groupType, messageType);
+ }
+
+ /**
+ * Serializes a marshallable object to a byte array.
+ *
+ * @see UserObjectSerializer#write(Object)
+ */
+ public <T> SerializationResult writeMarshallable(T object) {
+ return userObjectSerializer.write(object);
+ }
+
+ /**
+ * Deserializes a marshallable object from a byte array.
+ *
+ * @see UserObjectSerializer#read(Map, byte[])
+ */
+ public <T> T readMarshallable(Map<Integer, ClassDescriptor> descriptor, byte[] array) {
+ return userObjectSerializer.read(descriptor, array);
+ }
+
+ /**
+ * Returns {@code true} if type descriptor id belongs to a built-in type, {@code false} otherwise.
+ *
+ * @param typeDescriptorId Type descriptor id.
+ * @return {@code true} if descriptor is built-in, {@code false} otherwise.
+ */
+ public boolean isBuiltIn(int typeDescriptorId) {
+ return userObjectSerializer.isBuiltIn(typeDescriptorId);
+ }
+
+ /**
+ * Gets a class descriptor by the descriptor id.
+ *
+ * @param typeDescriptorId Type descriptor id.
+ * @return Class descriptor.
+ */
+ public ClassDescriptor getClassDescriptor(int typeDescriptorId) {
+ return userObjectSerializer.getClassDescriptor(typeDescriptorId);
+ }
+
+ /**
+ * Gets a class descriptor for the class specified by a name.
+ *
+ * @param typeName Class' name.
+ * @return Class descriptor.
+ */
+ public ClassDescriptor getClassDescriptor(String typeName) {
+ return userObjectSerializer.getClassDescriptor(typeName);
+ }
+}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
index c5cc87d..8f8b3dd 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.network.serialization;
+import org.apache.ignite.lang.IgniteInternalException;
+
/**
* Serialization type.
*/
@@ -39,4 +41,20 @@ public enum SerializationType {
public int value() {
return value;
}
+
+ /**
+ * Gets SerializationType by {@link #value}.
+ *
+ * @param value Value.
+ * @return Serialization type.
+ */
+ public static SerializationType getByValue(int value) {
+ for (SerializationType serializationType : SerializationType.values()) {
+ if (serializationType.value == value) {
+ return serializationType;
+ }
+ }
+
+ throw new IgniteInternalException("SerializationType by value=" + value + " not found");
+ }
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/UserObjectSerializer.java
similarity index 52%
copy from modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
copy to modules/network/src/main/java/org/apache/ignite/internal/network/serialization/UserObjectSerializer.java
index c5cc87d..b88cf8d 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/UserObjectSerializer.java
@@ -17,26 +17,32 @@
package org.apache.ignite.internal.network.serialization;
-/**
- * Serialization type.
- */
-public enum SerializationType {
- /** Used for predefined descriptors like primitive (or boxed int). See {@link BuiltinType}. */
- BUILTIN(0),
- /** Type for classes that are neither serializable nor externalizable. */
- ARBITRARY(1),
- /** Externalizable. */
- EXTERNALIZABLE(2),
- /** Serializable (but not Externalizable). */
- SERIALIZABLE(3);
+import java.util.Map;
+
+/** User object serializer. */
+public interface UserObjectSerializer {
+ /**
+ * Reads a marshallable object from a byte array using a map of descriptors.
+ *
+ * @param descriptors Descriptor map.
+ * @param array Byte array.
+ * @param <T> Object type.
+ * @return Unmarshalled object.
+ */
+ <T> T read(Map<Integer, ClassDescriptor> descriptors, byte[] array);
+
+ /**
+ * Marshalls object.
+ *
+ * @param object Object.
+ * @param <T> Object's type.
+ * @return {@link SerializationResult}.
+ */
+ <T> SerializationResult write(T object);
- private final int value;
+ ClassDescriptor getClassDescriptor(int typeDescriptorId);
- SerializationType(int value) {
- this.value = value;
- }
+ ClassDescriptor getClassDescriptor(String typeName);
- public int value() {
- return value;
- }
+ boolean isBuiltIn(int typeDescriptorId);
}
diff --git a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java
index 37d7eae..3aa9a83 100644
--- a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java
+++ b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeClusterServiceFactory.java
@@ -34,6 +34,7 @@ import org.apache.ignite.internal.network.NetworkMessagesFactory;
import org.apache.ignite.internal.network.netty.ConnectionManager;
import org.apache.ignite.internal.network.recovery.RecoveryClientHandshakeManager;
import org.apache.ignite.internal.network.recovery.RecoveryServerHandshakeManager;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.network.AbstractClusterService;
import org.apache.ignite.network.ClusterLocalConfiguration;
import org.apache.ignite.network.ClusterService;
@@ -41,7 +42,6 @@ import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NodeFinder;
import org.apache.ignite.network.NodeFinderFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
/**
* Cluster service factory that uses ScaleCube for messaging and topology services.
@@ -74,7 +74,7 @@ public class ScaleCubeClusterServiceFactory {
public void start() {
String consistentId = context.getName();
- MessageSerializationRegistry registry = context.getSerializationRegistry();
+ var serializationService = new SerializationService(context.getSerializationRegistry(), null);
UUID launchId = UUID.randomUUID();
@@ -84,7 +84,7 @@ public class ScaleCubeClusterServiceFactory {
connectionMgr = new ConnectionManager(
configView,
- registry,
+ serializationService,
consistentId,
() -> new RecoveryServerHandshakeManager(launchId, consistentId, messageFactory),
() -> new RecoveryClientHandshakeManager(launchId, consistentId, messageFactory),
diff --git a/modules/network/src/test/java/org/apache/ignite/internal/network/netty/InboundDecoderTest.java b/modules/network/src/test/java/org/apache/ignite/internal/network/netty/InboundDecoderTest.java
index 2f32e49..694274f 100644
--- a/modules/network/src/test/java/org/apache/ignite/internal/network/netty/InboundDecoderTest.java
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/netty/InboundDecoderTest.java
@@ -36,6 +36,8 @@ import org.apache.ignite.internal.network.AllTypesMessage;
import org.apache.ignite.internal.network.AllTypesMessageGenerator;
import org.apache.ignite.internal.network.NestedMessageMessage;
import org.apache.ignite.internal.network.direct.DirectMessageWriter;
+import org.apache.ignite.internal.network.serialization.PerSessionSerializationService;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.TestMessage;
import org.apache.ignite.network.TestMessageSerializationRegistryImpl;
@@ -91,9 +93,11 @@ public class InboundDecoderTest {
* Serializes and then deserializes the given message.
*/
private <T extends NetworkMessage> T sendAndReceive(T msg) {
- var channel = new EmbeddedChannel(new InboundDecoder(registry));
+ var serializationService = new SerializationService(registry, null);
+ var perSessionSerializationService = new PerSessionSerializationService(serializationService);
+ var channel = new EmbeddedChannel(new InboundDecoder(perSessionSerializationService));
- var writer = new DirectMessageWriter(registry, ConnectionManager.DIRECT_PROTOCOL_VERSION);
+ var writer = new DirectMessageWriter(perSessionSerializationService, ConnectionManager.DIRECT_PROTOCOL_VERSION);
MessageSerializer<NetworkMessage> serializer = registry.createSerializer(msg.groupType(), msg.messageType());
@@ -127,7 +131,9 @@ public class InboundDecoderTest {
*/
@Test
public void testPartialHeader() throws Exception {
- var channel = new EmbeddedChannel(new InboundDecoder(registry));
+ var serializationService = new SerializationService(registry, null);
+ var perSessionSerializationService = new PerSessionSerializationService(serializationService);
+ var channel = new EmbeddedChannel(new InboundDecoder(perSessionSerializationService));
ByteBuf buffer = allocator.buffer();
@@ -158,11 +164,13 @@ public class InboundDecoderTest {
Mockito.doReturn(channel).when(ctx).channel();
- final var decoder = new InboundDecoder(registry);
+ var serializationService = new SerializationService(registry, null);
+ var perSessionSerializationService = new PerSessionSerializationService(serializationService);
+ final var decoder = new InboundDecoder(perSessionSerializationService);
final var list = new ArrayList<>();
- var writer = new DirectMessageWriter(registry, ConnectionManager.DIRECT_PROTOCOL_VERSION);
+ var writer = new DirectMessageWriter(perSessionSerializationService, ConnectionManager.DIRECT_PROTOCOL_VERSION);
var msg = new TestMessagesFactory().testMessage().msg("abcdefghijklmn").build();
diff --git a/modules/network/src/test/java/org/apache/ignite/internal/network/netty/NettyServerTest.java b/modules/network/src/test/java/org/apache/ignite/internal/network/netty/NettyServerTest.java
index 1f13e99..c80ec5e 100644
--- a/modules/network/src/test/java/org/apache/ignite/internal/network/netty/NettyServerTest.java
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/netty/NettyServerTest.java
@@ -43,6 +43,7 @@ import org.apache.ignite.internal.configuration.testframework.ConfigurationExten
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.network.handshake.HandshakeAction;
import org.apache.ignite.internal.network.handshake.HandshakeManager;
+import org.apache.ignite.internal.network.serialization.SerializationService;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.network.NetworkMessage;
@@ -176,14 +177,13 @@ public class NettyServerTest {
bootstrapFactory.start();
server = new NettyServer(
- "test",
serverCfg.value(),
() -> handshakeManager,
sender -> {
},
(socketAddress, message) -> {
},
- registry,
+ new SerializationService(registry, null),
bootstrapFactory
);
@@ -244,14 +244,13 @@ public class NettyServerTest {
bootstrapFactory.start();
var server = new NettyServer(
- "test",
serverCfg.value(),
() -> mock(HandshakeManager.class),
null,
null,
null,
bootstrapFactory
- );
+ );
try {
server.start().get(3, TimeUnit.SECONDS);
diff --git a/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/MarshallableTest.java b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/MarshallableTest.java
new file mode 100644
index 0000000..402672d
--- /dev/null
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/MarshallableTest.java
@@ -0,0 +1,229 @@
+/*
+ * 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.ignite.internal.network.serialization;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.embedded.EmbeddedChannel;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.ignite.internal.network.direct.DirectMessageWriter;
+import org.apache.ignite.internal.network.netty.ConnectionManager;
+import org.apache.ignite.internal.network.netty.InboundDecoder;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.TestMessageSerializationRegistryImpl;
+import org.apache.ignite.network.TestMessagesFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.network.serialization.MessageSerializer;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests marshallable serialization.
+ */
+public class MarshallableTest {
+ /** {@link ByteBuf} allocator. */
+ private final UnpooledByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT;
+
+ /** Registry. */
+ private final MessageSerializationRegistry registry = new TestMessageSerializationRegistryImpl();
+
+ private final TestMessagesFactory msgFactory = new TestMessagesFactory();
+
+ /**
+ * Tests that marshallable object can be serialized along with its descriptor.
+ */
+ @Test
+ public void testMarshallable() {
+ // Test map that will be sent as a Marshallable object within the MessageWithMarshallable message
+ Map<String, SimpleSerializableObject> testMap = Map.of("test", new SimpleSerializableObject(10));
+
+ ByteBuffer outBuffer = write(testMap);
+
+ Map<String, SimpleSerializableObject> received = read(outBuffer);
+
+ assertEquals(testMap, received);
+ }
+
+ /** Writes a map to a buffer through the {@link MessageWithMarshallable}. */
+ private ByteBuffer write(Map<String, SimpleSerializableObject> testMap) {
+ var serializers = new Serialization();
+
+ var writer = new DirectMessageWriter(serializers.perSessionSerializationService, ConnectionManager.DIRECT_PROTOCOL_VERSION);
+
+ MessageWithMarshallable msg = msgFactory.messageWithMarshallable().marshallableMap(testMap).build();
+
+ MessageSerializer<NetworkMessage> serializer = registry.createSerializer(msg.groupType(), msg.messageType());
+
+ ByteBuffer nioBuffer = ByteBuffer.allocate(1000);
+
+ writer.setBuffer(nioBuffer);
+
+ // Write a message to the ByteBuffer.
+ boolean fullyWritten = serializer.writeMessage(msg, writer);
+
+ assertTrue(fullyWritten);
+
+ return nioBuffer;
+ }
+
+ /** Reads a {@link MessageWithMarshallable} from the buffer (byte by byte) and checks for the class descriptor merging. */
+ private Map<String, SimpleSerializableObject> read(ByteBuffer outBuffer) {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+
+ var channel = new EmbeddedChannel();
+
+ doReturn(channel).when(ctx).channel();
+
+ var serializers = new Serialization();
+
+ PerSessionSerializationService perSessionSerializationService = serializers.perSessionSerializationService;
+ ClassDescriptor descriptor = serializers.descriptor;
+
+ final var decoder = new InboundDecoder(perSessionSerializationService);
+
+ int size = outBuffer.position();
+
+ outBuffer.flip();
+
+ ByteBuf inBuffer = allocator.buffer();
+
+ // List that holds decoded object
+ final var list = new ArrayList<>();
+
+ for (int i = 0; i < size; i++) {
+ // Write bytes to a decoding buffer one by one
+ inBuffer.writeByte(outBuffer.get());
+
+ decoder.decode(ctx, inBuffer, list);
+
+ if (i < size - 1) {
+ // Any time before the buffer is fully read, message object should not be decoded
+ assertThat(list, is(empty()));
+ }
+ }
+
+ // Buffer is fully read, message object should be decoded
+ assertThat(list, hasSize(1));
+
+ // Check that the descriptor of the SimpleSerializableObject was received
+ Map<Integer, ClassDescriptor> mergedDescriptors = perSessionSerializationService.getDescriptorMapView();
+ assertEquals(1, mergedDescriptors.size());
+
+ ClassDescriptor mergedDescriptor = mergedDescriptors.values().stream().findFirst().get();
+
+ assertEquals(descriptor.className(), mergedDescriptor.className());
+
+ MessageWithMarshallable received = (MessageWithMarshallable) list.get(0);
+
+ return received.marshallableMap();
+ }
+
+ /** Helper class that holds classes needed for serialization. */
+ private class Serialization {
+ private final ClassDescriptorFactoryContext descriptorContext;
+ private final ClassDescriptorFactory factory;
+ private final ClassDescriptor descriptor;
+ private final StubSerializer userObjectSerializer;
+ private final SerializationService serializationService;
+ private final PerSessionSerializationService perSessionSerializationService;
+
+ Serialization() {
+ this.descriptorContext = new ClassDescriptorFactoryContext();
+ this.factory = new ClassDescriptorFactory(descriptorContext);
+
+ // Create descriptor for SimpleSerializableObject
+ this.descriptor = factory.create(SimpleSerializableObject.class);
+
+ this.userObjectSerializer = new StubSerializer(descriptorContext, descriptor);
+
+ this.serializationService = new SerializationService(registry, userObjectSerializer);
+ this.perSessionSerializationService = new PerSessionSerializationService(serializationService);
+ }
+ }
+
+ /** Stub implementation of the serializer, uses standard JDK serializable serialization to actually marshall an object. */
+ private class StubSerializer implements UserObjectSerializer {
+
+ private final ClassDescriptorFactoryContext descriptorContext;
+
+ private final ClassDescriptor descriptor;
+
+ StubSerializer(ClassDescriptorFactoryContext descriptorContext, ClassDescriptor descriptor) {
+ this.descriptorContext = descriptorContext;
+ this.descriptor = descriptor;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public <T> T read(Map<Integer, ClassDescriptor> descriptor, byte[] array) {
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(array); ObjectInputStream ois = new ObjectInputStream(bais)) {
+ return (T) ois.readObject();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public <T> SerializationResult write(T object) {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+ oos.writeObject(object);
+ oos.close();
+ return new SerializationResult(baos.toByteArray(), Collections.singletonList(descriptor.descriptorId()));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public ClassDescriptor getClassDescriptor(int typeDescriptorId) {
+ return descriptorContext.getDescriptor(typeDescriptorId);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public ClassDescriptor getClassDescriptor(String typeName) {
+ assertEquals(descriptor.className(), typeName);
+
+ return descriptor;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean isBuiltIn(int typeDescriptorId) {
+ return ClassDescriptorFactoryContext.isBuiltIn(typeDescriptorId);
+ }
+ }
+}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/MessageWithMarshallable.java
similarity index 61%
copy from modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
copy to modules/network/src/test/java/org/apache/ignite/internal/network/serialization/MessageWithMarshallable.java
index c5cc87d..e3fe5fc 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/MessageWithMarshallable.java
@@ -17,26 +17,15 @@
package org.apache.ignite.internal.network.serialization;
-/**
- * Serialization type.
- */
-public enum SerializationType {
- /** Used for predefined descriptors like primitive (or boxed int). See {@link BuiltinType}. */
- BUILTIN(0),
- /** Type for classes that are neither serializable nor externalizable. */
- ARBITRARY(1),
- /** Externalizable. */
- EXTERNALIZABLE(2),
- /** Serializable (but not Externalizable). */
- SERIALIZABLE(3);
-
- private final int value;
-
- SerializationType(int value) {
- this.value = value;
- }
+import java.util.Map;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.TestMessageTypes;
+import org.apache.ignite.network.annotations.Marshallable;
+import org.apache.ignite.network.annotations.Transferable;
- public int value() {
- return value;
- }
+/** Message with a marshallable field. */
+@Transferable(TestMessageTypes.TEST_MESSAGE_WITH_MARSHALLABLE)
+public interface MessageWithMarshallable extends NetworkMessage {
+ @Marshallable
+ Map<String, SimpleSerializableObject> marshallableMap();
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/SimpleSerializableObject.java
similarity index 54%
copy from modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
copy to modules/network/src/test/java/org/apache/ignite/internal/network/serialization/SimpleSerializableObject.java
index c5cc87d..3b19ea6 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/SerializationType.java
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/SimpleSerializableObject.java
@@ -17,26 +17,33 @@
package org.apache.ignite.internal.network.serialization;
-/**
- * Serialization type.
- */
-public enum SerializationType {
- /** Used for predefined descriptors like primitive (or boxed int). See {@link BuiltinType}. */
- BUILTIN(0),
- /** Type for classes that are neither serializable nor externalizable. */
- ARBITRARY(1),
- /** Externalizable. */
- EXTERNALIZABLE(2),
- /** Serializable (but not Externalizable). */
- SERIALIZABLE(3);
+import java.io.Serializable;
+import java.util.Objects;
+
+/** Serializable class for marshallable test. */
+public class SimpleSerializableObject implements Serializable {
+ private static final long serialVersionUid = 0L;
- private final int value;
+ private final int val;
+
+ public SimpleSerializableObject(int val) {
+ this.val = val;
+ }
- SerializationType(int value) {
- this.value = value;
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SimpleSerializableObject simpleSerializableObject = (SimpleSerializableObject) o;
+ return val == simpleSerializableObject.val;
}
- public int value() {
- return value;
+ @Override
+ public int hashCode() {
+ return Objects.hash(val);
}
}
diff --git a/modules/network/src/test/java/org/apache/ignite/network/TestMessageTypes.java b/modules/network/src/test/java/org/apache/ignite/network/TestMessageTypes.java
index 00a2331..0d036a0 100644
--- a/modules/network/src/test/java/org/apache/ignite/network/TestMessageTypes.java
+++ b/modules/network/src/test/java/org/apache/ignite/network/TestMessageTypes.java
@@ -41,4 +41,6 @@ public class TestMessageTypes {
public static final short TEST = 5;
public static final short NESTED_MESSAGE = 6;
+
+ public static final short TEST_MESSAGE_WITH_MARSHALLABLE = 7;
}
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java
index 4f32e71..2c80c9c 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java
@@ -37,13 +37,11 @@ import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.MessagingService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.client.service.RaftGroupListener;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.utils.ClusterServiceTestUtils;
@@ -62,8 +60,6 @@ public class ItLozaTest {
/** Server port offset. */
private static final int PORT = 20010;
- private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistryImpl();
-
@WorkDirectory
private Path dataPath;
@@ -91,7 +87,6 @@ public class ItLozaTest {
testInfo,
port,
new StaticNodeFinder(srvs),
- SERIALIZATION_REGISTRY,
NETWORK_FACTORY
);
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/client/service/ItAbstractListenerSnapshotTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/client/service/ItAbstractListenerSnapshotTest.java
index 397e836..5bceb02 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/client/service/ItAbstractListenerSnapshotTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/client/service/ItAbstractListenerSnapshotTest.java
@@ -38,11 +38,9 @@ import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
import org.apache.ignite.raft.jraft.rpc.impl.RaftGroupServiceImpl;
@@ -81,8 +79,6 @@ public abstract class ItAbstractListenerSnapshotTest<T extends RaftGroupListener
/** Network factory. */
private static final TestScaleCubeClusterServiceFactory NETWORK_FACTORY = new TestScaleCubeClusterServiceFactory();
- private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistryImpl();
-
@WorkDirectory
private Path workDir;
@@ -353,7 +349,6 @@ public abstract class ItAbstractListenerSnapshotTest<T extends RaftGroupListener
testInfo,
port,
new StaticNodeFinder(List.of(otherPeer)),
- SERIALIZATION_REGISTRY,
NETWORK_FACTORY
);
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItCliServiceTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItCliServiceTest.java
index 58652d8..bc9c640 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItCliServiceTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItCliServiceTest.java
@@ -14,8 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.ignite.raft.jraft.core;
+import static java.lang.Thread.sleep;
+import static java.util.stream.Collectors.toList;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.util.ArrayList;
@@ -35,7 +45,6 @@ import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
@@ -58,15 +67,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.ExtendWith;
-import static java.lang.Thread.sleep;
-import static java.util.stream.Collectors.toList;
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
-
/**
* Jraft cli tests.
*/
@@ -89,7 +89,7 @@ public class ItCliServiceTest {
private ExecutorService clientExecutor;
- /** */
+ /** Executes before each test. */
@BeforeEach
public void setup(TestInfo testInfo, @WorkDirectory Path dataPath) throws Exception {
LOG.info(">>>>>>>>>>>>>>> Start test method: " + testInfo.getDisplayName());
@@ -99,15 +99,18 @@ public class ItCliServiceTest {
LinkedHashSet<PeerId> learners = new LinkedHashSet<>();
// 2 learners
- for (int i = 0; i < 2; i++)
+ for (int i = 0; i < 2; i++) {
learners.add(new PeerId(TestUtils.getLocalAddress(), TestUtils.INIT_PORT + LEARNER_PORT_STEP + i));
+ }
cluster = new TestCluster(groupId, dataPath.toString(), peers, learners, 300, testInfo);
- for (PeerId peer : peers)
+ for (PeerId peer : peers) {
cluster.start(peer.getEndpoint());
+ }
- for (PeerId peer : learners)
+ for (PeerId peer : learners) {
cluster.startLearner(peer);
+ }
cluster.waitLeader();
cluster.ensureLeader(cluster.getLeader());
@@ -120,18 +123,15 @@ public class ItCliServiceTest {
opts.setClientExecutor(clientExecutor);
List<NetworkAddress> addressList = peers.stream()
- .map(PeerId::getEndpoint)
- .map(JRaftUtils::addressFromEndpoint)
- .collect(toList());
-
- var registry = new MessageSerializationRegistryImpl();
+ .map(PeerId::getEndpoint)
+ .map(JRaftUtils::addressFromEndpoint)
+ .collect(toList());
ClusterService clientSvc = ClusterServiceTestUtils.clusterService(
- testInfo,
- TestUtils.INIT_PORT - 1,
- new StaticNodeFinder(addressList),
- registry,
- new TestScaleCubeClusterServiceFactory()
+ testInfo,
+ TestUtils.INIT_PORT - 1,
+ new StaticNodeFinder(addressList),
+ new TestScaleCubeClusterServiceFactory()
);
clientSvc.start();
@@ -198,8 +198,9 @@ public class ItCliServiceTest {
cluster.ensureSame(addr -> addr.equals(learner3.getEndpoint()));
for (MockStateMachine fsm : cluster.getFsms()) {
- if (!fsm.getAddress().equals(learner3.getEndpoint()))
+ if (!fsm.getAddress().equals(learner3.getEndpoint())) {
assertEquals(10, fsm.getLogs().size());
+ }
}
assertEquals(0, cluster.getFsmByPeer(learner3).getLogs().size());
@@ -216,8 +217,9 @@ public class ItCliServiceTest {
cluster.ensureSame();
- for (MockStateMachine fsm : cluster.getFsms())
+ for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(20, fsm.getLogs().size());
+ }
List<PeerId> newLearners = new ArrayList<>(oldLearners);
newLearners.add(learner3);
@@ -231,8 +233,9 @@ public class ItCliServiceTest {
cluster.ensureSame(addr -> addr.equals(learner3.getEndpoint()));
for (MockStateMachine fsm : cluster.getFsms()) {
- if (!fsm.getAddress().equals(learner3.getEndpoint()))
+ if (!fsm.getAddress().equals(learner3.getEndpoint())) {
assertEquals(30, fsm.getLogs().size());
+ }
}
// Latest 10 logs are not replicated to learner3, because it's removed.
@@ -251,10 +254,11 @@ public class ItCliServiceTest {
// Latest 10 logs are not replicated to learner1 and learner2, because they were removed by resetting learners set.
for (MockStateMachine fsm : cluster.getFsms()) {
- if (!oldLearners.contains(new PeerId(fsm.getAddress(), 0)))
+ if (!oldLearners.contains(new PeerId(fsm.getAddress(), 0))) {
assertEquals(40, fsm.getLogs().size());
- else
+ } else {
assertEquals(30, fsm.getLogs().size());
+ }
}
assertEquals(Collections.singletonList(learner3), cliService.getLearners(groupId, conf));
@@ -285,8 +289,9 @@ public class ItCliServiceTest {
cluster.ensureSame();
- for (MockStateMachine fsm : cluster.getFsms())
+ for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(20, fsm.getLogs().size());
+ }
//remove peer3
assertTrue(cliService.removePeer(groupId, conf, peer3).isOk());
@@ -298,10 +303,11 @@ public class ItCliServiceTest {
cluster.ensureSame(addr -> addr.equals(peer3.getEndpoint()));
for (MockStateMachine fsm : cluster.getFsms()) {
- if (fsm.getAddress().equals(peer3.getEndpoint()))
+ if (fsm.getAddress().equals(peer3.getEndpoint())) {
assertEquals(20, fsm.getLogs().size());
- else
+ } else {
assertEquals(30, fsm.getLogs().size());
+ }
}
}
@@ -310,8 +316,9 @@ public class ItCliServiceTest {
public void testChangePeers() throws Exception {
List<PeerId> newPeers = TestUtils.generatePeers(10);
newPeers.removeAll(conf.getPeerSet());
- for (PeerId peer : newPeers)
+ for (PeerId peer : newPeers) {
assertTrue(cluster.start(peer.getEndpoint()));
+ }
cluster.waitLeader();
Node oldLeaderNode = cluster.getLeader();
assertNotNull(oldLeaderNode);
@@ -329,17 +336,21 @@ public class ItCliServiceTest {
public void testSnapshot() throws Exception {
sendTestTaskAndWait(cluster.getLeader(), 0);
assertEquals(5, cluster.getFsms().size());
- for (MockStateMachine fsm : cluster.getFsms())
+ for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(0, fsm.getSaveSnapshotTimes());
+ }
- for (PeerId peer : conf)
+ for (PeerId peer : conf) {
assertTrue(cliService.snapshot(groupId, peer).isOk());
+ }
- for (PeerId peer : conf.getLearners())
+ for (PeerId peer : conf.getLearners()) {
assertTrue(cliService.snapshot(groupId, peer).isOk());
+ }
- for (MockStateMachine fsm : cluster.getFsms())
+ for (MockStateMachine fsm : cluster.getFsms()) {
assertEquals(1, fsm.getSaveSnapshotTimes());
+ }
}
@Test
@@ -365,8 +376,7 @@ public class ItCliServiceTest {
try {
cliService.getPeers(groupId, conf);
fail();
- }
- catch (IllegalStateException e) {
+ } catch (IllegalStateException e) {
assertTrue(e.getMessage().startsWith("Fail to get leader of group " + groupId), e.getMessage());
}
}
@@ -397,8 +407,7 @@ public class ItCliServiceTest {
try {
cliService.getAlivePeers(groupId, conf);
fail();
- }
- catch (IllegalStateException e) {
+ } catch (IllegalStateException e) {
assertTrue(e.getMessage().startsWith("Fail to get leader of group " + groupId), e.getMessage());
}
}
@@ -427,8 +436,9 @@ public class ItCliServiceTest {
assertEquals(groupIds.size(), rebalancedLeaderIds.size());
Map<PeerId, Integer> ret = new HashMap<>();
- for (Map.Entry<String, PeerId> entry : rebalancedLeaderIds.entrySet())
+ for (Map.Entry<String, PeerId> entry : rebalancedLeaderIds.entrySet()) {
ret.compute(entry.getValue(), (ignored, num) -> num == null ? 1 : num + 1);
+ }
int expectedAvgLeaderNum = (int) Math.ceil((double) groupIds.size() / conf.size());
for (Map.Entry<PeerId, Integer> entry : ret.entrySet()) {
System.out.println(entry);
@@ -473,15 +483,16 @@ public class ItCliServiceTest {
Map<String, PeerId> rebalancedLeaderIds = new HashMap<>();
CliService cliService = new MockTransferLeaderFailCliService(rebalancedLeaderIds,
- new PeerId("host_1", 8080));
+ new PeerId("host_1", 8080));
assertEquals("Fail to transfer leader",
- cliService.rebalance(groupIds, conf, rebalancedLeaderIds).getErrorMsg());
+ cliService.rebalance(groupIds, conf, rebalancedLeaderIds).getErrorMsg());
assertTrue(groupIds.size() >= rebalancedLeaderIds.size());
Map<PeerId, Integer> ret = new HashMap<>();
- for (Map.Entry<String, PeerId> entry : rebalancedLeaderIds.entrySet())
+ for (Map.Entry<String, PeerId> entry : rebalancedLeaderIds.entrySet()) {
ret.compute(entry.getValue(), (ignored, num) -> num == null ? 1 : num + 1);
+ }
for (Map.Entry<PeerId, Integer> entry : ret.entrySet()) {
LOG.info(entry.toString());
assertEquals(new PeerId("host_1", 8080), entry.getKey());
@@ -501,10 +512,11 @@ public class ItCliServiceTest {
@Override
public Status getLeader(String groupId, Configuration conf, PeerId leaderId) {
PeerId ret = rebalancedLeaderIds.get(groupId);
- if (ret != null)
+ if (ret != null) {
leaderId.parse(ret.toString());
- else
+ } else {
leaderId.parse(initialLeaderId.toString());
+ }
return Status.OK();
}
@@ -546,21 +558,24 @@ public class ItCliServiceTest {
}
/**
+ * Waits for the condition.
+ *
* @param cond The condition.
* @param timeout The timeout.
* @return {@code True} if condition has happened within the timeout.
*/
- @SuppressWarnings("BusyWait") protected static boolean waitForCondition(BooleanSupplier cond, long timeout) {
+ @SuppressWarnings("BusyWait")
+ protected static boolean waitForCondition(BooleanSupplier cond, long timeout) {
long stop = System.currentTimeMillis() + timeout;
while (System.currentTimeMillis() < stop) {
- if (cond.getAsBoolean())
+ if (cond.getAsBoolean()) {
return true;
+ }
try {
sleep(50);
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
return false;
}
}
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
index 6c726d2..2fd9392 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
@@ -47,7 +47,6 @@ import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NodeFinder;
import org.apache.ignite.network.StaticNodeFinder;
-import org.apache.ignite.network.TestMessageSerializationRegistryImpl;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
import org.apache.ignite.raft.jraft.Iterator;
import org.apache.ignite.raft.jraft.JRaftUtils;
@@ -3541,11 +3540,10 @@ public class ItNodeTest {
var nodeManager = new NodeManager();
ClusterService clusterService = ClusterServiceTestUtils.clusterService(
- testInfo,
- peerId.getEndpoint().getPort(),
- new StaticNodeFinder(addressList),
- new TestMessageSerializationRegistryImpl(),
- new TestScaleCubeClusterServiceFactory()
+ testInfo,
+ peerId.getEndpoint().getPort(),
+ new StaticNodeFinder(addressList),
+ new TestScaleCubeClusterServiceFactory()
);
ExecutorService requestExecutor = JRaftUtils.createRequestExecutor(nodeOptions);
@@ -3578,11 +3576,10 @@ public class ItNodeTest {
*/
private ClusterService createClusterService(Endpoint endpoint, NodeFinder nodeFinder) {
return ClusterServiceTestUtils.clusterService(
- testInfo,
- endpoint.getPort(),
- nodeFinder,
- new TestMessageSerializationRegistryImpl(),
- new TestScaleCubeClusterServiceFactory()
+ testInfo,
+ endpoint.getPort(),
+ nodeFinder,
+ new TestScaleCubeClusterServiceFactory()
);
}
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/RaftServerAbstractTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/RaftServerAbstractTest.java
index 4b28d23..6df91cd 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/RaftServerAbstractTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/RaftServerAbstractTest.java
@@ -21,11 +21,9 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
import org.apache.ignite.utils.ClusterServiceTestUtils;
import org.junit.jupiter.api.AfterEach;
@@ -48,8 +46,6 @@ abstract class RaftServerAbstractTest {
*/
protected static final int PORT = 20010;
- private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistryImpl();
-
/** Test info. */
TestInfo testInfo;
@@ -77,7 +73,6 @@ abstract class RaftServerAbstractTest {
testInfo,
port,
new StaticNodeFinder(servers),
- SERIALIZATION_REGISTRY,
NETWORK_FACTORY
);
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java
index 5bedd44..7e07c49 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/core/TestCluster.java
@@ -16,6 +16,11 @@
*/
package org.apache.ignite.raft.jraft.core;
+import static java.util.stream.Collectors.toList;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -38,7 +43,6 @@ import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
-import org.apache.ignite.network.TestMessageSerializationRegistryImpl;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
import org.apache.ignite.raft.jraft.JRaftServiceFactory;
import org.apache.ignite.raft.jraft.JRaftUtils;
@@ -60,11 +64,6 @@ import org.apache.ignite.utils.ClusterServiceTestUtils;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.TestInfo;
-import static java.util.stream.Collectors.toList;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertSame;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
/**
* Test cluster for NodeTest
*/
@@ -249,11 +248,10 @@ public class TestCluster {
NodeManager nodeManager = new NodeManager();
ClusterService clusterService = ClusterServiceTestUtils.clusterService(
- testInfo,
- listenAddr.getPort(),
- new StaticNodeFinder(addressList),
- new TestMessageSerializationRegistryImpl(),
- new TestScaleCubeClusterServiceFactory()
+ testInfo,
+ listenAddr.getPort(),
+ new StaticNodeFinder(addressList),
+ new TestScaleCubeClusterServiceFactory()
);
var rpcClient = new IgniteRpcClient(clusterService);
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java
index 56ae92d..1666ef7 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/rpc/IgniteRpcTest.java
@@ -17,12 +17,13 @@
package org.apache.ignite.raft.jraft.rpc;
+import static org.apache.ignite.raft.jraft.JRaftUtils.addressFromEndpoint;
+
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
import org.apache.ignite.raft.jraft.JRaftUtils;
@@ -35,10 +36,8 @@ import org.apache.ignite.utils.ClusterServiceTestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.TestInfo;
-import static org.apache.ignite.raft.jraft.JRaftUtils.addressFromEndpoint;
-
/**
- *
+ * Ignite RPC test.
*/
public class IgniteRpcTest extends AbstractRpcTest {
/** The counter. */
@@ -50,7 +49,7 @@ public class IgniteRpcTest extends AbstractRpcTest {
/** Test info. */
private final TestInfo testInfo;
- /** */
+ /** Constructor. */
public IgniteRpcTest(TestInfo testInfo) {
this.testInfo = testInfo;
}
@@ -66,11 +65,10 @@ public class IgniteRpcTest extends AbstractRpcTest {
/** {@inheritDoc} */
@Override public RpcServer<?> createServer(Endpoint endpoint) {
ClusterService service = ClusterServiceTestUtils.clusterService(
- testInfo,
- endpoint.getPort(),
- new StaticNodeFinder(Collections.emptyList()),
- new MessageSerializationRegistryImpl(),
- new TestScaleCubeClusterServiceFactory()
+ testInfo,
+ endpoint.getPort(),
+ new StaticNodeFinder(Collections.emptyList()),
+ new TestScaleCubeClusterServiceFactory()
);
NodeOptions nodeOptions = new NodeOptions();
@@ -95,11 +93,10 @@ public class IgniteRpcTest extends AbstractRpcTest {
int i = cntr.incrementAndGet();
ClusterService service = ClusterServiceTestUtils.clusterService(
- testInfo,
- endpoint.getPort() - i,
- new StaticNodeFinder(List.of(addressFromEndpoint(endpoint))),
- new MessageSerializationRegistryImpl(),
- new TestScaleCubeClusterServiceFactory()
+ testInfo,
+ endpoint.getPort() - i,
+ new StaticNodeFinder(List.of(addressFromEndpoint(endpoint))),
+ new TestScaleCubeClusterServiceFactory()
);
IgniteRpcClient client = new IgniteRpcClient(service) {
@@ -127,6 +124,8 @@ public class IgniteRpcTest extends AbstractRpcTest {
}
/**
+ * Waits for the cluster's topology to have the expected count of nodes.
+ *
* @param service The service.
* @param expected Expected count.
* @param timeout The timeout.
@@ -136,13 +135,13 @@ public class IgniteRpcTest extends AbstractRpcTest {
long stop = System.currentTimeMillis() + timeout;
while (System.currentTimeMillis() < stop) {
- if (service.topologyService().allMembers().size() == expected)
+ if (service.topologyService().allMembers().size() == expected) {
return true;
+ }
try {
Thread.sleep(50);
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
return false;
}
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
index 7955371..de684e6 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/ItDistributedConfigurationPropertiesTest.java
@@ -53,7 +53,6 @@ import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
@@ -116,7 +115,6 @@ public class ItDistributedConfigurationPropertiesTest {
testInfo,
addr.port(),
new StaticNodeFinder(memberAddrs),
- new MessageSerializationRegistryImpl(),
new TestScaleCubeClusterServiceFactory()
);
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
index c435400..58b4ea4 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItDistributedConfigurationStorageTest.java
@@ -47,7 +47,6 @@ import org.apache.ignite.internal.tx.impl.HeapLockManager;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.persistence.PersistentVaultService;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
@@ -97,7 +96,6 @@ public class ItDistributedConfigurationStorageTest {
testInfo,
addr.port(),
new StaticNodeFinder(List.of(addr)),
- new MessageSerializationRegistryImpl(),
new TestScaleCubeClusterServiceFactory()
);
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java
index cf0d41e..1eddc47 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java
@@ -71,11 +71,9 @@ import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.Pair;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
@@ -98,8 +96,6 @@ import org.mockito.junit.jupiter.MockitoExtension;
public class ItInternalTableScanTest {
private static final TestScaleCubeClusterServiceFactory NETWORK_FACTORY = new TestScaleCubeClusterServiceFactory();
- private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistryImpl();
-
private static final RaftMessagesFactory FACTORY = new RaftMessagesFactory();
private static final String TEST_TABLE_NAME = "testTbl";
@@ -143,7 +139,6 @@ public class ItInternalTableScanTest {
testInfo,
20_000,
new StaticNodeFinder(List.of(nodeNetworkAddress)),
- SERIALIZATION_REGISTRY,
NETWORK_FACTORY
);
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
index efa4505..46c22de 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
@@ -54,12 +54,10 @@ import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.lang.IgniteUuidGenerator;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
-import org.apache.ignite.network.MessageSerializationRegistryImpl;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NodeFinder;
import org.apache.ignite.network.StaticNodeFinder;
import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
-import org.apache.ignite.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
@@ -81,8 +79,6 @@ public class ItTxDistributedTestSingleNode extends TxAbstractTest {
private static final TestScaleCubeClusterServiceFactory NETWORK_FACTORY = new TestScaleCubeClusterServiceFactory();
- private static final MessageSerializationRegistry SERIALIZATION_REGISTRY = new MessageSerializationRegistryImpl();
-
private ClusterService client;
protected Map<ClusterNode, Loza> raftServers;
@@ -381,7 +377,6 @@ public class ItTxDistributedTestSingleNode extends TxAbstractTest {
testInfo,
port,
nodeFinder,
- SERIALIZATION_REGISTRY,
NETWORK_FACTORY
);