Skip to content

IGNITE-27953 Use message serializer for CacheAffinityChangeMessage #12867

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anton-vinogradov wants to merge 9 commits into
base: master
Choose a base branch
from
Open

IGNITE-27953 Use message serializer for CacheAffinityChangeMessage #12867

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
27aed64
WIP
anton-vinogradov Mar 6, 2026
b428ffe
WIP
anton-vinogradov Mar 6, 2026
02c34ef
WIP
anton-vinogradov Mar 6, 2026
7a18860
WIP
anton-vinogradov Mar 6, 2026
84dc5a7
WIP
anton-vinogradov Mar 6, 2026
782475d
Merge remote-tracking branch 'origin/master' into ignite-27953
anton-vinogradov Mar 6, 2026
6c113ee
WIP
anton-vinogradov Mar 6, 2026
3c1a21a
WIP
anton-vinogradov Mar 6, 2026
a56fc7b
WIP
anton-vinogradov Mar 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
124 changes: 73 additions & 51 deletions modules/codegen/src/main/java/org/apache/ignite/internal/MessageSerializerGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -349,10 +350,7 @@ private void returnFalseIfWriteFailed(VariableElement field) throws Exception {
return;
}

imports.add("org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType");

returnFalseIfWriteFailed(write, field, "writer.writeObjectArray", getExpr,
"MessageCollectionItemType." + messageCollectionItemType(componentType));
returnFalseIfWriteFailed(write, field, "writer.writeObjectArray", getExpr, messageCollectionItemTypes(type));

return;
}
Expand All @@ -374,16 +372,10 @@ else if (sameType(type, "org.apache.ignite.internal.processors.affinity.Affinity
returnFalseIfWriteFailed(write, field, "writer.writeAffinityTopologyVersion", getExpr);

else if (assignableFrom(erasedType(type), type(Map.class.getName()))) {
List<? extends TypeMirror> typeArgs = ((DeclaredType)type).getTypeArguments();

assert typeArgs.size() == 2;

imports.add("org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType");

List<String> args = new ArrayList<>();

args.add(getExpr);
args.add("MessageCollectionItemType." + messageCollectionItemType(typeArgs.get(0)));
args.add("MessageCollectionItemType." + messageCollectionItemType(typeArgs.get(1)));
args.add(messageCollectionItemTypes(type));

if (compress)
args.add("true"); // the value of the compress argument in the MessageWriter#writeMap method
Expand All @@ -410,20 +402,8 @@ else if (assignableFrom(type, type(MESSAGE_INTERFACE))) {
returnFalseIfWriteFailed(write, field, "writer.writeMessage", getExpr);
}

else if (assignableFrom(erasedType(type), type(Collection.class.getName()))) {
List<? extends TypeMirror> typeArgs = ((DeclaredType)type).getTypeArguments();

assert typeArgs.size() == 1;

imports.add("org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType");

String collectionWriter = assignableFrom(erasedType(type), type(Set.class.getName()))
? "writer.writeSet"
: "writer.writeCollection";

returnFalseIfWriteFailed(write, field, collectionWriter, getExpr,
"MessageCollectionItemType." + messageCollectionItemType(typeArgs.get(0)));
}
else if (assignableFrom(erasedType(type), type(Collection.class.getName())))
returnFalseIfWriteFailed(write, field, "writer.writeCollection", getExpr, messageCollectionItemTypes(type));

else if (enumType(env, type)) {
Element element = env.getTypeUtils().asElement(type);
Expand Down Expand Up @@ -578,25 +558,15 @@ private void returnFalseIfReadFailed(VariableElement field) throws Exception {
}

if (componentType.getKind() == TypeKind.ARRAY) {
TypeMirror ctype = ((ArrayType)componentType).getComponentType();

assert ctype.getKind().isPrimitive();

returnFalseIfReadFailed(field, "reader.readObjectArray",
"MessageCollectionItemType." + messageCollectionItemType(ctype),
ctype.getKind().name().toLowerCase() + "[].class");
returnFalseIfReadFailed(field, "reader.readObjectArray", messageCollectionItemTypes(type));

return;
}

if (componentType.getKind() == TypeKind.DECLARED) {
Element componentElement = ((DeclaredType)componentType).asElement();

String cls = componentElement.getSimpleName().toString();

returnFalseIfReadFailed(field, "reader.readObjectArray",
"MessageCollectionItemType." + messageCollectionItemType(componentType),
cls + ".class");
returnFalseIfReadFailed(field, "reader.readObjectArray", messageCollectionItemTypes(type));

if (!"java.lang".equals(env.getElementUtils().getPackageOf(componentElement).getQualifiedName().toString())) {
String importCls = ((QualifiedNameable)componentElement).getQualifiedName().toString();
Expand Down Expand Up @@ -630,9 +600,8 @@ else if (assignableFrom(erasedType(type), type(Map.class.getName()))) {
assert typeArgs.size() == 2;

List<String> args = new ArrayList<>();
args.add("MessageCollectionItemType." + messageCollectionItemType(typeArgs.get(0)));
args.add("MessageCollectionItemType." + messageCollectionItemType(typeArgs.get(1)));
args.add("false"); // the value of the linked argument in the MessageReader#readMap method

args.add(messageCollectionItemTypes(type));

if (compress)
args.add("true"); // the value of the compress argument in the MessageReader#readMap method
Expand Down Expand Up @@ -660,16 +629,7 @@ else if (assignableFrom(type, type(MESSAGE_INTERFACE))) {
}

else if (assignableFrom(erasedType(type), type(Collection.class.getName()))) {
List<? extends TypeMirror> typeArgs = ((DeclaredType)type).getTypeArguments();

assert typeArgs.size() == 1;

String collectionReader = assignableFrom(erasedType(type), type(Set.class.getName()))
? "reader.readSet"
: "reader.readCollection";

returnFalseIfReadFailed(field, collectionReader,
"MessageCollectionItemType." + messageCollectionItemType(typeArgs.get(0)));
returnFalseIfReadFailed(field, "reader.readCollection", messageCollectionItemTypes(type));
}
else if (enumType(env, type)) {
String fieldPrefix = typeNameToFieldName(env.getTypeUtils().asElement(type).getSimpleName().toString());
Expand All @@ -691,6 +651,68 @@ else if (enumType(env, type)) {
throw new IllegalArgumentException("Unsupported type kind: " + type.getKind());
}

/** */
private String messageCollectionItemTypes(TypeMirror type) throws Exception {
imports.add("org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType");

if (type.getKind() == TypeKind.ARRAY) {
ArrayType arrType = (ArrayType)type;
TypeMirror componentType = arrType.getComponentType();

String clazz;

if (componentType.getKind() == TypeKind.ARRAY) {
TypeMirror ctype = ((ArrayType)componentType).getComponentType();

clazz = ctype.getKind().name().toLowerCase() + "[].class";
}
else if (componentType.getKind() == TypeKind.DECLARED) {
Element componentElement = ((DeclaredType)componentType).asElement();

clazz = componentElement.getSimpleName() + ".class";
}
else {
assert componentType.getKind().isPrimitive();

imports.add("org.apache.ignite.plugin.extensions.communication.MessageItemType");

return "new MessageItemType(MessageCollectionItemType." + messageCollectionItemType(componentType) + "_ARR)";
}

imports.add("org.apache.ignite.plugin.extensions.communication.MessageArrayType");

return "new MessageArrayType(" + messageCollectionItemTypes(componentType) + ", " + clazz + ")";
}
else if (assignableFrom(erasedType(type), type(Map.class.getName()))) {
imports.add("org.apache.ignite.plugin.extensions.communication.MessageMapType");

List<? extends TypeMirror> typeArgs = ((DeclaredType)type).getTypeArguments();

assert typeArgs.size() == 2;

return "new MessageMapType(" +
messageCollectionItemTypes(typeArgs.get(0)) + ", " +
messageCollectionItemTypes(typeArgs.get(1)) + ", " +
assignableFrom(erasedType(type), type(LinkedHashMap.class.getName())) + ")";
}
else if (assignableFrom(erasedType(type), type(Collection.class.getName()))) {
imports.add("org.apache.ignite.plugin.extensions.communication.MessageCollectionType");

List<? extends TypeMirror> typeArgs = ((DeclaredType)type).getTypeArguments();

assert typeArgs.size() == 1;

return "new MessageCollectionType(" +
messageCollectionItemTypes(typeArgs.get(0)) + ", " +
assignableFrom(erasedType(type), type(Set.class.getName())) + ")";
}
else {
imports.add("org.apache.ignite.plugin.extensions.communication.MessageItemType");

return "new MessageItemType(MessageCollectionItemType." + messageCollectionItemType(type) + ")";
}
}

/**
* Find MessageCollectionItemType for a given type.
* <p>
Expand Down
31 changes: 10 additions & 21 deletions modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import java.util.BitSet;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.function.Function;
import org.apache.ignite.internal.direct.state.DirectMessageState;
Expand All @@ -38,8 +37,10 @@
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageArrayType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionType;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageMapType;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.jetbrains.annotations.Nullable;

Expand Down Expand Up @@ -379,52 +380,40 @@ public ByteBuffer getBuffer() {
}

/** {@inheritDoc} */
@Override public <T> T[] readObjectArray(MessageCollectionItemType itemType, Class<T> itemCls) {
@Override public <T> T[] readObjectArray(MessageArrayType type) {
DirectByteBufferStream stream = state.item().stream;

T[] msg = stream.readObjectArray(itemType, itemCls, this);
T[] msg = stream.readObjectArray(type, this);

lastRead = stream.lastFinished();

return msg;
}

/** {@inheritDoc} */
@Override public <C extends Collection<?>> C readCollection(MessageCollectionItemType itemType) {
@Override public <C extends Collection<?>> C readCollection(MessageCollectionType type) {
DirectByteBufferStream stream = state.item().stream;

C col = stream.readList(itemType, this);
C col = stream.readCollection(type, this);

lastRead = stream.lastFinished();

return col;
}

/** {@inheritDoc} */
@Override public <SET extends Set<?>> SET readSet(MessageCollectionItemType itemType) {
DirectByteBufferStream stream = state.item().stream;

SET set = stream.readSet(itemType, this);

lastRead = stream.lastFinished();

return set;
}

/** {@inheritDoc} */
@Override public <M extends Map<?, ?>> M readMap(MessageCollectionItemType keyType,
MessageCollectionItemType valType, boolean linked, boolean compress) {
@Override public <M extends Map<?, ?>> M readMap(MessageMapType type, boolean compress) {
DirectByteBufferStream stream = state.item().stream;

M map;

if (compress)
map = readCompressedMessageAndDeserialize(
stream,
tmpReader -> tmpReader.state.item().stream.readMap(keyType, valType, linked, tmpReader)
tmpReader -> tmpReader.state.item().stream.readMap(type, tmpReader)
);
else {
map = stream.readMap(keyType, valType, linked, this);
map = stream.readMap(type, this);

lastRead = stream.lastFinished();
}
Expand Down
25 changes: 10 additions & 15 deletions modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import java.util.BitSet;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.function.Consumer;
import org.apache.ignite.internal.direct.state.DirectMessageState;
Expand All @@ -37,8 +36,10 @@
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageArrayType;
import org.apache.ignite.plugin.extensions.communication.MessageCollectionType;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageMapType;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.Nullable;

Expand Down Expand Up @@ -360,41 +361,35 @@ public ByteBuffer getBuffer() {
}

/** {@inheritDoc} */
@Override public <T> boolean writeObjectArray(T[] arr, MessageCollectionItemType itemType) {
@Override public <T> boolean writeObjectArray(T[] arr, MessageArrayType type) {
DirectByteBufferStream stream = state.item().stream;

stream.writeObjectArray(arr, itemType, this);
stream.writeObjectArray(arr, type, this);

return stream.lastFinished();
}

/** {@inheritDoc} */
@Override public <T> boolean writeCollection(Collection<T> col, MessageCollectionItemType itemType) {
@Override public <T> boolean writeCollection(Collection<T> col, MessageCollectionType type) {
DirectByteBufferStream stream = state.item().stream;

stream.writeCollection(col, itemType, this);
stream.writeCollection(col, type, this);

return stream.lastFinished();
}

/** {@inheritDoc} */
@Override public <T> boolean writeSet(Set<T> set, MessageCollectionItemType itemType) {
return writeCollection(set, itemType);
}

/** {@inheritDoc} */
@Override public <K, V> boolean writeMap(Map<K, V> map, MessageCollectionItemType keyType,
MessageCollectionItemType valType, boolean compress) {
@Override public <K, V> boolean writeMap(Map<K, V> map, MessageMapType type, boolean compress) {
DirectByteBufferStream stream = state.item().stream;

if (compress)
writeCompressedMessage(
tmpWriter -> tmpWriter.state.item().stream.writeMap(map, keyType, valType, tmpWriter),
tmpWriter -> tmpWriter.state.item().stream.writeMap(map, type, tmpWriter),
map == null,
stream
);
else
stream.writeMap(map, keyType, valType, this);
stream.writeMap(map, type, this);

return stream.lastFinished();
}
Expand Down
Loading
Loading