shishkovilja commented on code in PR #12860: URL: https://github.com/apache/ignite/pull/12860#discussion_r2913011725
########## modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/DiscoveryMessageParser.java: ########## @@ -0,0 +1,168 @@ +/* + * 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.zk.internal; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.util.zip.DeflaterOutputStream; +import java.util.zip.InflaterInputStream; +import org.apache.ignite.internal.direct.DirectMessageReader; +import org.apache.ignite.internal.direct.DirectMessageWriter; +import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl; +import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage; +import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.marshaller.Marshaller; +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.MessageFactoryProvider; +import org.apache.ignite.plugin.extensions.communication.MessageSerializer; +import org.apache.ignite.spi.IgniteSpiException; + +import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.makeMessageType; + +/** + * Class is responsible for serializing discovery messages using RU-ready {@link MessageSerializer} mechanism. + */ +public class DiscoveryMessageParser { + /** Leading byte for messages use {@link JdkMarshaller} for serialization. */ + // TODO: remove these flags after refactoring all discovery messages. + private static final byte JAVA_SERIALIZATION = (byte)1; + + /** Leading byte for messages use {@link MessageSerializer} for serialization. */ + private static final byte MESSAGE_SERIALIZATION = (byte)2; + + /** Size for an intermediate buffer for serializing discovery messages. */ + private static final int MSG_BUFFER_SIZE = 100; + + /** */ + private final MessageFactory msgFactory; + + /** */ + private final Marshaller marsh; + + /** */ + public DiscoveryMessageParser(Marshaller marsh) { + this.marsh = marsh; + this.msgFactory = new IgniteMessageFactoryImpl( + new MessageFactoryProvider[] { new DiscoveryMessageFactory() }); + } + + /** Marshals discovery message to bytes array. */ + public byte[] marshalZip(DiscoveryCustomMessage msg) { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + + try (DeflaterOutputStream out = new DeflaterOutputStream(baos)) { + if (msg instanceof Message) { + out.write(MESSAGE_SERIALIZATION); + + serializeMessage((Message)msg, out); + } + else { + out.write(JAVA_SERIALIZATION); + + U.marshal(marsh, msg, out); + } + } + catch (Exception e) { + throw new IgniteSpiException("Failed to serialize message: " + msg, e); + } + + return baos.toByteArray(); + } + + /** Unmarshals discovery message from bytes array. */ + public DiscoveryCustomMessage unmarshalZip(byte[] bytes) { + try ( + ByteArrayInputStream bais = new ByteArrayInputStream(bytes); + InflaterInputStream in = new InflaterInputStream(bais) + ) { + byte mode = (byte)in.read(); + + if (mode == JAVA_SERIALIZATION) + return U.unmarshal(marsh, in, U.gridClassLoader()); + + if (MESSAGE_SERIALIZATION != mode) + throw new IOException("Received unexpected byte while reading discovery message: " + mode); + + return (DiscoveryCustomMessage)deserializeMessage(in); + } + catch (Exception e) { + throw new IgniteSpiException("Failed to deserialize message.", e); + } + } + + /** */ + private void serializeMessage(Message m, OutputStream out) throws IOException { + DirectMessageWriter msgWriter = new DirectMessageWriter(msgFactory); + ByteBuffer msgBuf = ByteBuffer.allocate(MSG_BUFFER_SIZE); + + msgWriter.setBuffer(msgBuf); + + MessageSerializer msgSer = msgFactory.serializer(m.directType()); + + boolean finished; + + do { + // Should be cleared before first operation. Review Comment: ```suggestion ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
