This is an automated email from the ASF dual-hosted git repository.
petrov-mg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 84eaa722f38 IGNITE-29006 Refactored TcpDiscoveryMessageSerializer to
handle serialization only (#13514)
84eaa722f38 is described below
commit 84eaa722f38dd61fe4516b2d318a235d87d7b361
Author: Mikhail Petrov <[email protected]>
AuthorDate: Tue Sep 22 14:39:14 2026 +0300
IGNITE-29006 Refactored TcpDiscoveryMessageSerializer to handle
serialization only (#13514)
---
.../ignite/spi/discovery/tcp/ServerImpl.java | 134 +++++++++------------
.../spi/discovery/tcp/TcpDiscoveryIoSession.java | 70 ++++-------
.../tcp/TcpDiscoveryMessageSerializer.java | 68 -----------
.../tcp/internal/ClientMessageHolder.java | 59 +++++++++
.../internal/TcpDiscoveryMessageSerializer.java | 96 +++++++++++++++
5 files changed, 239 insertions(+), 188 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 5974e617807..1356c3d04d2 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -99,7 +99,6 @@ import
org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.P1;
-import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -122,8 +121,10 @@ import
org.apache.ignite.spi.discovery.DiscoveryNotification;
import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
import org.apache.ignite.spi.discovery.DiscoverySpiListener;
import org.apache.ignite.spi.discovery.IgniteDiscoveryThread;
+import org.apache.ignite.spi.discovery.tcp.internal.ClientMessageHolder;
import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket;
import org.apache.ignite.spi.discovery.tcp.internal.FutureTask;
+import
org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryMessageSerializer;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNodesRing;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoverySpiState;
@@ -2851,15 +2852,11 @@ class ServerImpl extends TcpDiscoveryImpl {
/** Force pending messages send. */
private boolean forceSndPending;
- // This serializer is used exclusively for serializing messages sent
to clients,
- // as it represents a special case within the RingMessageWorker
workflow.
- // Generally, both serialization and deserialization of messages
should be handled by TcpDiscoveryIoSession.
- // However, there are scenarios where the session is not available,
yet messages still need to be sent to clients.
- // A typical example is a single server with one or more connected
clients.
- // To address this, we use TcpDiscoveryMessageSerializer, which
includes some code copied from TcpDiscoveryIoSession
- // and can be instantiated independently of any active session.
- /** */
- private final TcpDiscoveryMessageSerializer clientMsgSer = new
TcpDiscoveryMessageSerializer(ctx);
+ /**
+ * This serializer is used exclusively for pre-serializing messages
sent to clients. Pre-serialization is performed
+ * once for each message, after which the serialized message is reused
for sending to all connected clients.
+ */
+ private final TcpDiscoveryMessageSerializer cliMsgSer = new
TcpDiscoveryMessageSerializer(ctx);
/** IO session. */
private TcpDiscoveryIoSession ses;
@@ -3229,50 +3226,52 @@ class ServerImpl extends TcpDiscoveryImpl {
* @param msg Message.
*/
private void sendMessageToClients(TcpDiscoveryAbstractMessage msg) {
- if (redirectToClients(msg)) {
- if (spi.ensured(msg))
- msgHist.add(msg);
+ if (!redirectToClients(msg))
+ return;
- if (clientMsgWorkers.isEmpty())
- return;
+ if (spi.ensured(msg))
+ msgHist.add(msg);
+
+ if (clientMsgWorkers.isEmpty())
+ return;
+
+ ClientMessageHolder sharedMsgHolder = new ClientMessageHolder(msg);
+
+ for (ClientMessageWorker worker : clientMsgWorkers.values()) {
+ TcpDiscoveryAbstractMessage rebuiltMsg = rebuildForClient(msg,
worker.clientNodeId);
- byte[] msgBytes;
+ ClientMessageHolder msgToSend = rebuiltMsg == msg
+ ? sharedMsgHolder
+ : new ClientMessageHolder(rebuiltMsg);
try {
- msgBytes = clientMsgSer.serializeMessage(msg);
+ msgToSend.serialize(cliMsgSer);
}
- catch (IgniteCheckedException | IOException e) {
- U.error(log, "Failed to serialize message: " + msg, e);
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to serialize message: " + msgToSend,
e);
return;
}
- for (ClientMessageWorker clientMsgWorker :
clientMsgWorkers.values()) {
- TcpDiscoveryAbstractMessage msg0 = msg;
- byte[] msgBytes0 = msgBytes;
+ worker.addMessage(msgToSend);
+ }
+ }
- if (msg instanceof TcpDiscoveryNodeAddedMessage) {
- TcpDiscoveryNodeAddedMessage nodeAddedMsg =
(TcpDiscoveryNodeAddedMessage)msg;
+ /** */
+ private TcpDiscoveryAbstractMessage
rebuildForClient(TcpDiscoveryAbstractMessage msg, UUID clientNodeId) {
+ if (!(msg instanceof TcpDiscoveryNodeAddedMessage))
+ return msg;
- if
(clientMsgWorker.clientNodeId.equals(nodeAddedMsg.node().id())) {
- msg0 = new
TcpDiscoveryNodeAddedMessage(nodeAddedMsg);
+ TcpDiscoveryNodeAddedMessage nodeAddedMsg =
(TcpDiscoveryNodeAddedMessage)msg;
- prepareNodeAddedMessage(msg0,
clientMsgWorker.clientNodeId, null);
+ if (!clientNodeId.equals(nodeAddedMsg.node().id()))
+ return msg;
- try {
- msgBytes0 =
clientMsgSer.serializeMessage(msg0);
- }
- catch (IgniteCheckedException | IOException e) {
- U.error(log, "Failed to serialize message: " +
msg0, e);
+ TcpDiscoveryNodeAddedMessage res = new
TcpDiscoveryNodeAddedMessage(nodeAddedMsg);
- return;
- }
- }
- }
+ prepareNodeAddedMessage(res, clientNodeId, null);
- clientMsgWorker.addMessage(msg0, msgBytes0);
- }
- }
+ return res;
}
/**
@@ -7469,21 +7468,10 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/** */
- private class ClientMessageWorker extends
MessageWorker<T2<TcpDiscoveryAbstractMessage, byte[]>> {
+ private class ClientMessageWorker extends
MessageWorker<ClientMessageHolder> {
/** Node ID. */
private final UUID clientNodeId;
- // The code responsible for sending and receiving messages to and from
client nodes represents a special case in ServerImpl,
- // as it is split into two separate components.
- // One part, ClientMessageWorker, handles only message sending to
clients and does not process responses.
- // The other part, which reads messages from clients, is implemented
in SocketReader.
- // Due to this separation, we don't require a full
TcpDiscoveryIoSession here
- // and can instead extract just the message-writing functionality.
- // At the same time, we aim to keep both reading and writing logic
encapsulated within TcpDiscoveryIoSession.
- // As a result, we need to copy some code from TcpDiscoveryIoSession
into the new class, TcpDiscoveryMessageSerializer.
- /** */
- private final TcpDiscoveryMessageSerializer clientMsgSer;
-
/** Session shared with the socket reader serving the same client
connection. */
private final TcpDiscoveryIoSession ses;
@@ -7518,8 +7506,6 @@ class ServerImpl extends TcpDiscoveryImpl {
this.ses = ses;
this.clientNodeId = clientNodeId;
- clientMsgSer = new TcpDiscoveryMessageSerializer(ctx);
-
lastMetricsUpdateMsgTimeNanos = System.nanoTime();
}
@@ -7546,24 +7532,19 @@ class ServerImpl extends TcpDiscoveryImpl {
this.metrics = metrics;
}
- /**
- * @param msg Message.
- */
+ /** @param msg Discovery Message. */
void addMessage(TcpDiscoveryAbstractMessage msg) {
- addMessage(msg, null);
+ addMessage(new ClientMessageHolder(msg));
}
- /**
- * @param msg Message.
- * @param msgBytes Optional message bytes.
- */
- void addMessage(TcpDiscoveryAbstractMessage msg, @Nullable byte[]
msgBytes) {
- T2<TcpDiscoveryAbstractMessage, byte[]> t = new T2<>(msg,
msgBytes);
+ /** @param msgHolder Holder of a Discovery Message to send to the
client. */
+ void addMessage(ClientMessageHolder msgHolder) {
+ TcpDiscoveryAbstractMessage msg = msgHolder.message();
if (msg.highPriority())
- queue.addFirst(t);
+ queue.addFirst(msgHolder);
else
- queue.add(t);
+ queue.add(msgHolder);
DebugLogger log = messageLogger(msg);
@@ -7572,10 +7553,10 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/** {@inheritDoc} */
- @Override protected void
processMessage(T2<TcpDiscoveryAbstractMessage, byte[]> msgT) {
+ @Override protected void processMessage(ClientMessageHolder msgHolder)
{
boolean success = false;
- TcpDiscoveryAbstractMessage msg = msgT.get1();
+ TcpDiscoveryAbstractMessage msg = msgHolder.message();
try {
assert msg.verified() : msg;
@@ -7601,8 +7582,9 @@ class ServerImpl extends TcpDiscoveryImpl {
+ getLocalNodeId() + ", rmtNodeId=" +
clientNodeId + ", msg=" + msg + ']');
}
- writeToSocket(msgT,
spi.failureDetectionTimeoutEnabled() ? spi.clientFailureDetectionTimeout() :
- spi.getSocketTimeout());
+ long timeout = spi.failureDetectionTimeoutEnabled() ?
spi.clientFailureDetectionTimeout() : spi.getSocketTimeout();
+
+ writeMessage(msgHolder, timeout);
}
}
else {
@@ -7613,7 +7595,7 @@ class ServerImpl extends TcpDiscoveryImpl {
assert topologyInitialized(msg) : msg;
- writeToSocket(msgT, spi.getEffectiveSocketTimeout(false));
+ writeMessage(msgHolder,
spi.getEffectiveSocketTimeout(false));
}
boolean clientFailed = msg instanceof
TcpDiscoveryNodeFailedMessage &&
@@ -7643,14 +7625,16 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/**
- * @param msgT Message tuple.
+ * @param msgHolder Message holder.
* @param timeout Timeout.
*/
- private void writeToSocket(T2<TcpDiscoveryAbstractMessage, byte[]>
msgT, long timeout)
- throws IgniteCheckedException, IOException {
- byte[] msgBytes = msgT.get2() == null ?
clientMsgSer.serializeMessage(msgT.get1()) : msgT.get2();
+ private void writeMessage(ClientMessageHolder msgHolder, long timeout)
throws IgniteCheckedException, IOException {
+ byte[] msgBytes = msgHolder.messageBytes();
- spi.write(ses, msgBytes, timeout);
+ if (msgBytes != null)
+ spi.write(ses, msgBytes, timeout);
+ else
+ spi.writeMessage(ses, msgHolder.message(), timeout);
}
/**
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
index 0251712fc95..4b81a151fd7 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
@@ -36,7 +36,6 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.direct.DirectMessageReader;
-import org.apache.ignite.internal.direct.DirectMessageWriter;
import org.apache.ignite.internal.managers.communication.DiscoveryMarshalling;
import
org.apache.ignite.internal.managers.communication.UnknownMessageException;
import org.apache.ignite.internal.util.CommonUtils;
@@ -47,6 +46,7 @@ import org.apache.ignite.marshaller.jdk.JdkMarshaller;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
+import
org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryMessageSerializer;
import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -60,14 +60,21 @@ import org.jetbrains.annotations.Nullable;
* <li>Using {@link MessageSerializer} for messages implementing the
{@link Message} interface.</li>
* <li>Deprecated: Using {@link JdkMarshaller} for messages that have not
yet been refactored.</li>
* </ul>
- * A leading byte is used to distinguish between the modes. The byte will be
removed in future.
+ * A leading byte is used to distinguish between the modes. The byte will be
removed in the future.
+ * <p>
+ * <b>NOTE:</b> This class is designed with the following access rules in
mind. Socket read operations must be performed
+ * by a single thread at a time, while socket write operations may be
performed concurrently by multiple threads. Because
+ * {@link #writeMessage(TcpDiscoveryAbstractMessage)} writes messages in
batches, all session write methods must be
+ * blocking. Currently, {@link TcpDiscoveryIoSession} may be accessed
concurrently for writing by the
+ * ServerImpl.ClientMessageWorker and ServerImpl.SocketReader threads.
+ * </p>
*/
public class TcpDiscoveryIoSession implements AutoCloseable {
/** Default size of buffer used for buffering socket in/out. */
private static final int DFLT_SOCK_BUFFER_SIZE = 8192;
- /** Size for an intermediate buffer for serializing discovery messages. */
- private static final int MSG_BUFFER_SIZE = 100;
+ /** Size of the intermediate buffer a message is deserialized through. */
+ private static final int READ_BUFFER_SIZE = 100;
/** */
private final GridKernalContext ctx;
@@ -82,23 +89,20 @@ public class TcpDiscoveryIoSession implements AutoCloseable
{
private final Socket sock;
/** */
- private final DirectMessageWriter msgWriter;
+ private final TcpDiscoveryMessageSerializer msgSer;
/** */
private final DirectMessageReader msgReader;
+ /** */
+ private final ByteBuffer readBuf;
+
/** Buffered socket output stream. */
private final OutputStream out;
/** Buffered socket input stream. */
private final CompositeInputStream in;
- /** */
- private final ByteBuffer readBuf;
-
- /** */
- private final ByteBuffer writeBuf;
-
/**
* Creates a new discovery I/O session bound to the given socket.
*
@@ -112,12 +116,11 @@ public class TcpDiscoveryIoSession implements
AutoCloseable {
this.msgFactory = ctx.messageFactory();
this.log = ctx.log(getClass());
- readBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE);
- writeBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE);
-
- msgWriter = new DirectMessageWriter(msgFactory);
+ readBuf = ByteBuffer.allocate(READ_BUFFER_SIZE);
msgReader = new DirectMessageReader(msgFactory, null);
+ msgSer = new TcpDiscoveryMessageSerializer(ctx);
+
try {
int sendBufSize = sock.getSendBufferSize() > 0 ?
sock.getSendBufferSize() : DFLT_SOCK_BUFFER_SIZE;
int rcvBufSize = sock.getReceiveBufferSize() > 0 ?
sock.getReceiveBufferSize() : DFLT_SOCK_BUFFER_SIZE;
@@ -132,13 +135,14 @@ public class TcpDiscoveryIoSession implements
AutoCloseable {
/**
* Writes a discovery message to the underlying socket output stream.
+ * Refer to the class description for the rationale behind synchronized
access.
*
* @param msg Message to send to the remote node.
* @throws IgniteCheckedException If serialization fails.
*/
- void writeMessage(TcpDiscoveryAbstractMessage msg) throws
IgniteCheckedException, IOException {
+ synchronized void writeMessage(TcpDiscoveryAbstractMessage msg) throws
IgniteCheckedException, IOException {
try {
- serializeMessage((Message)msg, out);
+ msgSer.writeTo(msg, out);
out.flush();
}
@@ -262,39 +266,14 @@ public class TcpDiscoveryIoSession implements
AutoCloseable {
return sock;
}
- /**
- * Serializes a discovery message into given output stream.
- *
- * @param m Discovery message to serialize.
- * @param out Output stream to write serialized message.
- * @throws IOException If serialization fails.
- */
- void serializeMessage(Message m, OutputStream out) throws IOException,
IgniteCheckedException {
- DiscoveryMarshalling.marshal(m, ctx, null);
-
- msgWriter.reset();
- msgWriter.setBuffer(writeBuf);
-
- boolean finished;
-
- do {
- // Should be cleared before first operation.
- writeBuf.clear();
-
- finished = MessageSerialization.writeTo(msgFactory, m, msgWriter);
-
- out.write(writeBuf.array(), 0, writeBuf.position());
- }
- while (!finished);
- }
-
/**
* Writes raw data to the underlying socket output stream.
+ * Refer to the class description for the rationale behind synchronized
access.
*
* @param data Raw data to write.
* @throws IOException If failed.
*/
- void write(byte[] data) throws IOException {
+ synchronized void write(byte[] data) throws IOException {
out.write(data);
out.flush();
@@ -302,11 +281,12 @@ public class TcpDiscoveryIoSession implements
AutoCloseable {
/**
* Writes a single byte response to the underlying socket output stream.
+ * Refer to the class description for the rationale behind synchronized
access.
*
* @param b Integer response.
* @throws IOException If failed.
*/
- void write(int b) throws IOException {
+ synchronized void write(int b) throws IOException {
out.write(b);
out.flush();
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java
deleted file mode 100644
index ec7cdc569f0..00000000000
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryMessageSerializer.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.ignite.spi.discovery.tcp;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.Socket;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
-import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
-
-/**
- * Class is responsible for serializing discovery messages using RU-ready
{@link MessageSerializer} mechanism.
- * <p>
- * It is used in a special case: when server wants to send discovery messages
to clients, it may not have a {@link TcpDiscoveryIoSession}
- * to serialize the messages.
- * This class enables server to serialize discovery messages anyway,
duplicating serialization code from {@link TcpDiscoveryIoSession}.
- */
-class TcpDiscoveryMessageSerializer extends TcpDiscoveryIoSession {
- /**
- * @param ctx Kernal context.
- */
- public TcpDiscoveryMessageSerializer(GridKernalContext ctx) {
- super(ctx, new Socket() {
- @Override public OutputStream getOutputStream() throws IOException
{
- return null;
- }
-
- @Override public InputStream getInputStream() throws IOException {
- return null;
- }
- });
- }
-
- /**
- * Serializes a discovery message into a byte array.
- *
- * @param msg Discovery message to serialize.
- * @return Serialized byte array containing the message data.
- * @throws IgniteCheckedException If serialization fails.
- * @throws IOException If serialization fails.
- */
- byte[] serializeMessage(TcpDiscoveryAbstractMessage msg) throws
IgniteCheckedException, IOException {
- try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
- serializeMessage((Message)msg, out);
-
- return out.toByteArray();
- }
- }
-}
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java
new file mode 100644
index 00000000000..1652824dd15
--- /dev/null
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/ClientMessageHolder.java
@@ -0,0 +1,59 @@
+/*
+ * 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.spi.discovery.tcp.internal;
+
+import org.apache.ignite.IgniteCheckedException;
+import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+public class ClientMessageHolder {
+ /** */
+ private final TcpDiscoveryAbstractMessage msg;
+
+ /** */
+ private volatile byte[] msgBytes;
+
+ /** */
+ public ClientMessageHolder(TcpDiscoveryAbstractMessage msg) {
+ assert msg != null;
+
+ this.msg = msg;
+ }
+
+ /** */
+ public TcpDiscoveryAbstractMessage message() {
+ return msg;
+ }
+
+ /** */
+ public byte @Nullable [] messageBytes() {
+ return msgBytes;
+ }
+
+ /** */
+ public void serialize(TcpDiscoveryMessageSerializer ser) throws
IgniteCheckedException {
+ if (msgBytes == null)
+ msgBytes = ser.serialize(msg);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return msg.toString();
+ }
+}
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java
new file mode 100644
index 00000000000..45aa1d9b767
--- /dev/null
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryMessageSerializer.java
@@ -0,0 +1,96 @@
+/*
+ * 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.spi.discovery.tcp.internal;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.direct.DirectMessageWriter;
+import org.apache.ignite.internal.managers.communication.DiscoveryMarshalling;
+import org.apache.ignite.internal.util.io.GridByteArrayOutputStream;
+import org.apache.ignite.internal.util.nio.MessageSerialization;
+import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+
+/** */
+public class TcpDiscoveryMessageSerializer {
+ /** Size of the intermediate buffer a message is serialized through. */
+ private static final int BUFFER_SIZE = 100;
+
+ /** */
+ private final GridKernalContext ctx;
+
+ /** */
+ private final DirectMessageWriter writer;
+
+ /** */
+ private final ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE);
+
+ /** @param ctx Kernal context. */
+ public TcpDiscoveryMessageSerializer(GridKernalContext ctx) {
+ this.ctx = ctx;
+
+ writer = new DirectMessageWriter(ctx.messageFactory());
+ }
+
+ /**
+ * Serializes a discovery message into given output stream.
+ *
+ * @param msg Discovery message to serialize.
+ * @param out Output stream to write serialized message.
+ * @throws IgniteCheckedException If serialization fails.
+ * @throws IOException If serialization fails.
+ */
+ public void writeTo(TcpDiscoveryAbstractMessage msg, OutputStream out)
throws IgniteCheckedException, IOException {
+ DiscoveryMarshalling.marshal(msg, ctx, null);
+
+ writer.reset();
+ writer.setBuffer(buf);
+
+ boolean finished;
+
+ do {
+ // Should be cleared before first operation.
+ buf.clear();
+
+ finished = MessageSerialization.writeTo(ctx.messageFactory(), msg,
writer);
+
+ out.write(buf.array(), 0, buf.position());
+ }
+ while (!finished);
+ }
+
+ /**
+ * Serializes a discovery message into a byte array.
+ *
+ * @param msg Discovery message to serialize.
+ * @return Serialized byte array containing the message data.
+ * @throws IgniteCheckedException If serialization fails.
+ */
+ public byte[] serialize(TcpDiscoveryAbstractMessage msg) throws
IgniteCheckedException {
+ try (GridByteArrayOutputStream out = new GridByteArrayOutputStream()) {
+ writeTo(msg, out);
+
+ return out.toByteArray();
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException("Failed to serialize a discovery
message: " + msg, e);
+ }
+ }
+}
