You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vk...@apache.org on 2015/11/21 01:25:52 UTC
[1/3] ignite git commit: Direct marshalling optimizations
Repository: ignite
Updated Branches:
refs/heads/ignite-1.5 281b27755 -> 2de3b19b3
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
index 987090d..aa88808 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
@@ -23,7 +23,6 @@ import java.nio.ByteBuffer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
-import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.jetbrains.annotations.Nullable;
@@ -41,46 +40,48 @@ public class GridDirectParser implements GridNioParser {
private final MessageFactory msgFactory;
/** */
- private final MessageFormatter formatter;
+ private final GridNioMessageReaderFactory readerFactory;
/**
* @param msgFactory Message factory.
- * @param formatter Formatter.
+ * @param readerFactory Message reader factory.
*/
- public GridDirectParser(MessageFactory msgFactory, MessageFormatter formatter) {
+ public GridDirectParser(MessageFactory msgFactory, GridNioMessageReaderFactory readerFactory) {
assert msgFactory != null;
- assert formatter != null;
+ assert readerFactory != null;
this.msgFactory = msgFactory;
- this.formatter = formatter;
+ this.readerFactory = readerFactory;
}
/** {@inheritDoc} */
@Nullable @Override public Object decode(GridNioSession ses, ByteBuffer buf)
throws IOException, IgniteCheckedException {
- Message msg = ses.removeMeta(MSG_META_KEY);
+ MessageReader reader = ses.meta(READER_META_KEY);
- MessageReader reader = null;
+ if (reader == null)
+ ses.addMeta(READER_META_KEY, reader = readerFactory.reader(ses, msgFactory));
- if (msg == null && buf.hasRemaining()) {
- msg = msgFactory.create(buf.get());
+ Message msg = ses.removeMeta(MSG_META_KEY);
- ses.addMeta(READER_META_KEY, reader = formatter.reader(msgFactory, msg.getClass()));
- }
+ if (msg == null && buf.hasRemaining())
+ msg = msgFactory.create(buf.get());
boolean finished = false;
if (buf.hasRemaining()) {
- if (reader == null)
- reader = ses.meta(READER_META_KEY);
-
- assert reader != null;
+ if (reader != null)
+ reader.setCurrentReadClass(msg.getClass());
finished = msg.readFrom(buf, reader);
}
- if (finished)
+ if (finished) {
+ if (reader != null)
+ reader.reset();
+
return msg;
+ }
else {
ses.addMeta(MSG_META_KEY, msg);
@@ -93,4 +94,4 @@ public class GridDirectParser implements GridNioParser {
// No encoding needed for direct messages.
throw new UnsupportedEncodingException();
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageReaderFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageReaderFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageReaderFactory.java
new file mode 100644
index 0000000..cb7e62d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageReaderFactory.java
@@ -0,0 +1,37 @@
+/*
+ * 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.util.nio;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+
+/**
+ * Message reader factory.
+ */
+public interface GridNioMessageReaderFactory {
+ /**
+ * Creates new reader.
+ *
+ * @param ses Current session.
+ * @param msgFactory Message factory.
+ * @return Reader.
+ * @throws IgniteCheckedException In case of error.
+ */
+ public MessageReader reader(GridNioSession ses, MessageFactory msgFactory) throws IgniteCheckedException;
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageWriterFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageWriterFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageWriterFactory.java
new file mode 100644
index 0000000..6c4dee4
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMessageWriterFactory.java
@@ -0,0 +1,35 @@
+/*
+ * 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.util.nio;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * Message writer factory.
+ */
+public interface GridNioMessageWriterFactory {
+ /**
+ * Creates new writer.
+ *
+ * @param ses Current session.
+ * @return Writer.
+ * @throws IgniteCheckedException In case of error.
+ */
+ public MessageWriter writer(GridNioSession ses) throws IgniteCheckedException;
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
index 5bd08e7..384ee1a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
@@ -62,7 +62,6 @@ import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.apache.ignite.thread.IgniteThread;
import org.jetbrains.annotations.Nullable;
@@ -176,7 +175,7 @@ public class GridNioServer<T> {
/** */
@GridToStringExclude
- private MessageFormatter formatter;
+ private GridNioMessageWriterFactory writerFactory;
/** */
@GridToStringExclude
@@ -212,7 +211,7 @@ public class GridNioServer<T> {
* @param directMode Whether direct mode is used.
* @param daemon Daemon flag to create threads.
* @param metricsLsnr Metrics listener.
- * @param formatter Message formatter.
+ * @param writerFactory Writer factory.
* @param skipRecoveryPred Skip recovery predicate.
* @param msgQueueLsnr Message queue size listener.
* @param filters Filters for this server.
@@ -234,7 +233,7 @@ public class GridNioServer<T> {
boolean directMode,
boolean daemon,
GridNioMetricsListener metricsLsnr,
- MessageFormatter formatter,
+ GridNioMessageWriterFactory writerFactory,
IgnitePredicate<Message> skipRecoveryPred,
IgniteBiInClosure<GridNioSession, Integer> msgQueueLsnr,
GridNioFilter... filters
@@ -304,7 +303,7 @@ public class GridNioServer<T> {
this.directMode = directMode;
this.metricsLsnr = metricsLsnr;
- this.formatter = formatter;
+ this.writerFactory = writerFactory;
this.skipRecoveryPred = skipRecoveryPred != null ? skipRecoveryPred : F.<Message>alwaysFalse();
}
@@ -934,8 +933,14 @@ public class GridNioServer<T> {
MessageWriter writer = ses.meta(MSG_WRITER.ordinal());
- if (writer == null)
- ses.addMeta(MSG_WRITER.ordinal(), writer = formatter.writer());
+ if (writer == null) {
+ try {
+ ses.addMeta(MSG_WRITER.ordinal(), writer = writerFactory.writer(ses));
+ }
+ catch (IgniteCheckedException e) {
+ throw new IOException("Failed to create message writer.", e);
+ }
+ }
boolean handshakeFinished = sslFilter.lock(ses);
@@ -1129,8 +1134,14 @@ public class GridNioServer<T> {
MessageWriter writer = ses.meta(MSG_WRITER.ordinal());
- if (writer == null)
- ses.addMeta(MSG_WRITER.ordinal(), writer = formatter.writer());
+ if (writer == null) {
+ try {
+ ses.addMeta(MSG_WRITER.ordinal(), writer = writerFactory.writer(ses));
+ }
+ catch (IgniteCheckedException e) {
+ throw new IOException("Failed to create message writer.", e);
+ }
+ }
if (req == null) {
req = (NioOperationFuture<?>)ses.pollFuture();
@@ -1152,7 +1163,7 @@ public class GridNioServer<T> {
finished = msg.writeTo(buf, writer);
- if (finished)
+ if (finished && writer != null)
writer.reset();
}
@@ -1176,7 +1187,7 @@ public class GridNioServer<T> {
finished = msg.writeTo(buf, writer);
- if (finished)
+ if (finished && writer != null)
writer.reset();
}
@@ -2221,8 +2232,8 @@ public class GridNioServer<T> {
/** Daemon flag. */
private boolean daemon;
- /** Message formatter. */
- private MessageFormatter formatter;
+ /** Writer factory. */
+ private GridNioMessageWriterFactory writerFactory;
/** Skip recovery predicate. */
private IgnitePredicate<Message> skipRecoveryPred;
@@ -2253,7 +2264,7 @@ public class GridNioServer<T> {
directMode,
daemon,
metricsLsnr,
- formatter,
+ writerFactory,
skipRecoveryPred,
msgQueueLsnr,
filters != null ? Arrays.copyOf(filters, filters.length) : EMPTY_FILTERS
@@ -2450,11 +2461,11 @@ public class GridNioServer<T> {
}
/**
- * @param formatter Message formatter.
+ * @param writerFactory Writer factory.
* @return This for chaining.
*/
- public Builder<T> messageFormatter(MessageFormatter formatter) {
- this.formatter = formatter;
+ public Builder<T> writerFactory(GridNioMessageWriterFactory writerFactory) {
+ this.writerFactory = writerFactory;
return this;
}
@@ -2479,4 +2490,4 @@ public class GridNioServer<T> {
return this;
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
index 93e789d..ebe86fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
@@ -33,7 +33,6 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
-import org.jetbrains.annotations.Nullable;
/**
*
@@ -120,16 +119,17 @@ public class GridShmemCommunicationClient extends GridAbstractCommunicationClien
}
/** {@inheritDoc} */
- @Override public synchronized boolean sendMessage(@Nullable UUID nodeId, Message msg,
- IgniteInClosure<IgniteException> closure)
- throws IgniteCheckedException {
+ @Override public synchronized boolean sendMessage(UUID nodeId, Message msg,
+ IgniteInClosure<IgniteException> closure) throws IgniteCheckedException {
+ assert nodeId != null;
+
if (closed())
throw new IgniteCheckedException("Communication client was closed: " + this);
assert writeBuf.hasArray();
try {
- int cnt = U.writeMessageFully(msg, shmem.outputStream(), writeBuf, formatter.writer());
+ int cnt = U.writeMessageFully(msg, shmem.outputStream(), writeBuf, formatter.writer(nodeId));
metricsLsnr.onBytesSent(cnt);
}
@@ -154,4 +154,4 @@ public class GridShmemCommunicationClient extends GridAbstractCommunicationClien
@Override public String toString() {
return S.toString(GridShmemCommunicationClient.class, this, super.toString());
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageFormatter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageFormatter.java b/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageFormatter.java
index 4c9e0ee..7bc2a53 100644
--- a/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageFormatter.java
+++ b/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageFormatter.java
@@ -17,6 +17,8 @@
package org.apache.ignite.plugin.extensions.communication;
+import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.plugin.Extension;
/**
@@ -33,16 +35,19 @@ public interface MessageFormatter extends Extension {
/**
* Creates new message writer instance.
*
+ * @param rmtNodeId Remote node ID.
* @return Message writer.
+ * @throws IgniteCheckedException In case of error.
*/
- public MessageWriter writer();
+ public MessageWriter writer(UUID rmtNodeId) throws IgniteCheckedException;
/**
* Creates new message reader instance.
*
- * @param factory Message factory.
- * @param msgCls Message class to read.
+ * @param rmtNodeId Remote node ID.
+ * @param msgFactory Message factory.
* @return Message reader.
+ * @throws IgniteCheckedException In case of error.
*/
- public MessageReader reader(MessageFactory factory, Class<? extends Message> msgCls);
-}
\ No newline at end of file
+ public MessageReader reader(UUID rmtNodeId, MessageFactory msgFactory) throws IgniteCheckedException;
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java b/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java
index d40a384..502c69f 100644
--- a/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java
+++ b/modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/MessageReader.java
@@ -39,6 +39,13 @@ public interface MessageReader {
public void setBuffer(ByteBuffer buf);
/**
+ * Sets type of message currently read.
+ *
+ * @param msgCls Message type.
+ */
+ public void setCurrentReadClass(Class<? extends Message> msgCls);
+
+ /**
* Callback that must be invoked by a message implementation before message body started decoding.
*
* @return {@code True} if reading can proceed, {@code false} otherwise.
@@ -272,4 +279,21 @@ public interface MessageReader {
* Increments read state.
*/
public void incrementState();
-}
\ No newline at end of file
+
+ /**
+ * Callback called before inner message is read.
+ */
+ public void beforeInnerMessageRead();
+
+ /**
+ * Callback called after inner message is read.
+ *
+ * @param finished Whether message was fully read.
+ */
+ public void afterInnerMessageRead(boolean finished);
+
+ /**
+ * Resets this reader.
+ */
+ public void reset();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
index cd46de4..1cb202c 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
@@ -696,11 +696,11 @@ public abstract class IgniteSpiAdapter implements IgniteSpi, IgniteSpiManagement
if (msgFormatter0 == null) {
msgFormatter0 = new MessageFormatter() {
- @Override public MessageWriter writer() {
+ @Override public MessageWriter writer(UUID rmtNodeId) {
throw new IgniteException("Failed to write message, node is not started.");
}
- @Override public MessageReader reader(MessageFactory factory, Class<? extends Message> msgCls) {
+ @Override public MessageReader reader(UUID rmtNodeId, MessageFactory msgFactory) {
throw new IgniteException("Failed to read message, node is not started.");
}
};
@@ -880,4 +880,4 @@ public abstract class IgniteSpiAdapter implements IgniteSpi, IgniteSpiManagement
((IgniteKernal)ignite0).context().timeout().removeTimeoutObject(new GridSpiTimeoutObject(obj));
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index e8bd8a1..68e2f43 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -76,7 +76,9 @@ import org.apache.ignite.internal.util.nio.GridConnectionBytesVerifyFilter;
import org.apache.ignite.internal.util.nio.GridDirectParser;
import org.apache.ignite.internal.util.nio.GridNioCodecFilter;
import org.apache.ignite.internal.util.nio.GridNioFilter;
+import org.apache.ignite.internal.util.nio.GridNioMessageReaderFactory;
import org.apache.ignite.internal.util.nio.GridNioMessageTracker;
+import org.apache.ignite.internal.util.nio.GridNioMessageWriterFactory;
import org.apache.ignite.internal.util.nio.GridNioMetricsListener;
import org.apache.ignite.internal.util.nio.GridNioRecoveryDescriptor;
import org.apache.ignite.internal.util.nio.GridNioServer;
@@ -1575,29 +1577,38 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
}
};
- MessageFormatter msgFormatter = new MessageFormatter() {
- private MessageFormatter impl;
+ GridNioMessageReaderFactory readerFactory = new GridNioMessageReaderFactory() {
+ private MessageFormatter formatter;
- @Override public MessageWriter writer() {
- if (impl == null)
- impl = getSpiContext().messageFormatter();
+ @Override public MessageReader reader(GridNioSession ses, MessageFactory msgFactory)
+ throws IgniteCheckedException {
+ if (formatter == null)
+ formatter = getSpiContext().messageFormatter();
- assert impl != null;
+ assert formatter != null;
- return impl.writer();
+ UUID rmtNodeId = ses.meta(NODE_ID_META);
+
+ return rmtNodeId != null ? formatter.reader(rmtNodeId, msgFactory) : null;
}
+ };
- @Override public MessageReader reader(MessageFactory factory, Class<? extends Message> msgCls) {
- if (impl == null)
- impl = getSpiContext().messageFormatter();
+ GridNioMessageWriterFactory writerFactory = new GridNioMessageWriterFactory() {
+ private MessageFormatter formatter;
- assert impl != null;
+ @Override public MessageWriter writer(GridNioSession ses) throws IgniteCheckedException {
+ if (formatter == null)
+ formatter = getSpiContext().messageFormatter();
+
+ assert formatter != null;
+
+ UUID rmtNodeId = ses.meta(NODE_ID_META);
- return impl.reader(factory, msgCls);
+ return rmtNodeId != null ? formatter.writer(rmtNodeId) : null;
}
};
- GridDirectParser parser = new GridDirectParser(msgFactory, msgFormatter);
+ GridDirectParser parser = new GridDirectParser(msgFactory, readerFactory);
IgnitePredicate<Message> skipRecoveryPred = new IgnitePredicate<Message>() {
@Override public boolean apply(Message msg) {
@@ -1658,7 +1669,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
.metricsListener(metricsLsnr)
.writeTimeout(sockWriteTimeout)
.filters(filters)
- .messageFormatter(msgFormatter)
+ .writerFactory(writerFactory)
.skipRecoveryPredicate(skipRecoveryPred)
.messageQueueSizeListener(queueSizeMonitor)
.build();
@@ -1918,7 +1929,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
UUID nodeId = null;
- if (!client.async() && !locNode.version().equals(node.version()))
+ if (!client.async())
nodeId = node.id();
retry = client.sendMessage(nodeId, msg, ackC);
@@ -2591,7 +2602,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
buf.order(ByteOrder.nativeOrder());
- boolean written = msg.writeTo(buf, getSpiContext().messageFormatter().writer());
+ boolean written = msg.writeTo(buf, null);
assert written;
@@ -2932,25 +2943,34 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
}
};
- MessageFormatter msgFormatter = new MessageFormatter() {
- private MessageFormatter impl;
+ GridNioMessageWriterFactory writerFactory = new GridNioMessageWriterFactory() {
+ private MessageFormatter formatter;
- @Override public MessageWriter writer() {
- if (impl == null)
- impl = getSpiContext().messageFormatter();
+ @Override public MessageWriter writer(GridNioSession ses) throws IgniteCheckedException {
+ if (formatter == null)
+ formatter = getSpiContext().messageFormatter();
- assert impl != null;
+ assert formatter != null;
- return impl.writer();
+ UUID rmtNodeId = ses.meta(NODE_ID_META);
+
+ return rmtNodeId != null ? formatter.writer(rmtNodeId) : null;
}
+ };
- @Override public MessageReader reader(MessageFactory factory, Class<? extends Message> msgCls) {
- if (impl == null)
- impl = getSpiContext().messageFormatter();
+ GridNioMessageReaderFactory readerFactory = new GridNioMessageReaderFactory() {
+ private MessageFormatter formatter;
- assert impl != null;
+ @Override public MessageReader reader(GridNioSession ses, MessageFactory msgFactory)
+ throws IgniteCheckedException {
+ if (formatter == null)
+ formatter = getSpiContext().messageFormatter();
+
+ assert formatter != null;
+
+ UUID rmtNodeId = ses.meta(NODE_ID_META);
- return impl.reader(factory, msgCls);
+ return rmtNodeId != null ? formatter.reader(rmtNodeId, msgFactory) : null;
}
};
@@ -2959,8 +2979,8 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
log,
endpoint,
srvLsnr,
- msgFormatter,
- new GridNioCodecFilter(new GridDirectParser(msgFactory, msgFormatter), log, true),
+ writerFactory,
+ new GridNioCodecFilter(new GridDirectParser(msgFactory, readerFactory), log, true),
new GridConnectionBytesVerifyFilter(log)
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java b/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
index f0f41bf..176973e 100644
--- a/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
+++ b/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
@@ -29,6 +29,11 @@ import org.apache.ignite.cache.CacheEntryProcessor;
/**
* Convenience adapter to transform update existing values in streaming cache
* based on the previously cached value.
+ * <p>
+ * This transformer implement {@link EntryProcessor} and internally will call
+ * {@link IgniteCache#invoke(Object, EntryProcessor, Object...)} method. Note
+ * that the value received from the data streamer will be passed to the entry
+ * processor as an argument.
*/
public abstract class StreamTransformer<K, V> implements StreamReceiver<K, V>, EntryProcessor<K, V, Object> {
/** */
@@ -37,7 +42,7 @@ public abstract class StreamTransformer<K, V> implements StreamReceiver<K, V>, E
/** {@inheritDoc} */
@Override public void receive(IgniteCache<K, V> cache, Collection<Map.Entry<K, V>> entries) throws IgniteException {
for (Map.Entry<K, V> entry : entries)
- cache.invoke(entry.getKey(), this);
+ cache.invoke(entry.getKey(), this, entry.getValue());
}
/**
@@ -53,4 +58,4 @@ public abstract class StreamTransformer<K, V> implements StreamReceiver<K, V>, E
}
};
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/test/java/org/apache/ignite/testframework/GridSpiTestContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridSpiTestContext.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridSpiTestContext.java
index 906a050..e257a97 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/GridSpiTestContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/GridSpiTestContext.java
@@ -37,12 +37,12 @@ import org.apache.ignite.events.TaskEvent;
import org.apache.ignite.internal.ClusterMetricsSnapshot;
import org.apache.ignite.internal.direct.DirectMessageReader;
import org.apache.ignite.internal.direct.DirectMessageWriter;
+import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
@@ -492,12 +492,12 @@ public class GridSpiTestContext implements IgniteSpiContext {
@Override public MessageFormatter messageFormatter() {
if (formatter == null) {
formatter = new MessageFormatter() {
- @Override public MessageWriter writer() {
- return new DirectMessageWriter();
+ @Override public MessageWriter writer(UUID rmtNodeId) {
+ return new DirectMessageWriter(GridIoManager.DIRECT_PROTO_VER);
}
- @Override public MessageReader reader(MessageFactory factory, Class<? extends Message> msgCls) {
- return new DirectMessageReader(factory, this);
+ @Override public MessageReader reader(UUID rmtNodeId, MessageFactory msgFactory) {
+ return new DirectMessageReader(msgFactory, GridIoManager.DIRECT_PROTO_VER);
}
};
}
@@ -573,4 +573,4 @@ public class GridSpiTestContext implements IgniteSpiContext {
this.obj = obj;
}
}
-}
\ No newline at end of file
+}
[2/3] ignite git commit: Direct marshalling optimizations
Posted by vk...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java
new file mode 100644
index 0000000..ad8671d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java
@@ -0,0 +1,1360 @@
+/*
+ * 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.direct.stream.v1;
+
+import java.lang.reflect.Array;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.UUID;
+import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
+import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.internal.util.typedef.internal.U;
+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.MessageFactory;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
+
+/**
+ * Direct marshalling I/O stream (version 1).
+ */
+public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
+ /** */
+ private static final Unsafe UNSAFE = GridUnsafe.unsafe();
+
+ /** */
+ private static final long BYTE_ARR_OFF = UNSAFE.arrayBaseOffset(byte[].class);
+
+ /** */
+ private static final long SHORT_ARR_OFF = UNSAFE.arrayBaseOffset(short[].class);
+
+ /** */
+ private static final long INT_ARR_OFF = UNSAFE.arrayBaseOffset(int[].class);
+
+ /** */
+ private static final long LONG_ARR_OFF = UNSAFE.arrayBaseOffset(long[].class);
+
+ /** */
+ private static final long FLOAT_ARR_OFF = UNSAFE.arrayBaseOffset(float[].class);
+
+ /** */
+ private static final long DOUBLE_ARR_OFF = UNSAFE.arrayBaseOffset(double[].class);
+
+ /** */
+ private static final long CHAR_ARR_OFF = UNSAFE.arrayBaseOffset(char[].class);
+
+ /** */
+ private static final long BOOLEAN_ARR_OFF = UNSAFE.arrayBaseOffset(boolean[].class);
+
+ /** */
+ private static final byte[] BYTE_ARR_EMPTY = new byte[0];
+
+ /** */
+ private static final short[] SHORT_ARR_EMPTY = new short[0];
+
+ /** */
+ private static final int[] INT_ARR_EMPTY = U.EMPTY_INTS;
+
+ /** */
+ private static final long[] LONG_ARR_EMPTY = U.EMPTY_LONGS;
+
+ /** */
+ private static final float[] FLOAT_ARR_EMPTY = new float[0];
+
+ /** */
+ private static final double[] DOUBLE_ARR_EMPTY = new double[0];
+
+ /** */
+ private static final char[] CHAR_ARR_EMPTY = new char[0];
+
+ /** */
+ private static final boolean[] BOOLEAN_ARR_EMPTY = new boolean[0];
+
+ /** */
+ private static final ArrayCreator<byte[]> BYTE_ARR_CREATOR = new ArrayCreator<byte[]>() {
+ @Override public byte[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return BYTE_ARR_EMPTY;
+
+ default:
+ return new byte[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<short[]> SHORT_ARR_CREATOR = new ArrayCreator<short[]>() {
+ @Override public short[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return SHORT_ARR_EMPTY;
+
+ default:
+ return new short[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<int[]> INT_ARR_CREATOR = new ArrayCreator<int[]>() {
+ @Override public int[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return INT_ARR_EMPTY;
+
+ default:
+ return new int[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<long[]> LONG_ARR_CREATOR = new ArrayCreator<long[]>() {
+ @Override public long[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return LONG_ARR_EMPTY;
+
+ default:
+ return new long[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<float[]> FLOAT_ARR_CREATOR = new ArrayCreator<float[]>() {
+ @Override public float[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return FLOAT_ARR_EMPTY;
+
+ default:
+ return new float[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<double[]> DOUBLE_ARR_CREATOR = new ArrayCreator<double[]>() {
+ @Override public double[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return DOUBLE_ARR_EMPTY;
+
+ default:
+ return new double[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<char[]> CHAR_ARR_CREATOR = new ArrayCreator<char[]>() {
+ @Override public char[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return CHAR_ARR_EMPTY;
+
+ default:
+ return new char[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<boolean[]> BOOLEAN_ARR_CREATOR = new ArrayCreator<boolean[]>() {
+ @Override public boolean[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return BOOLEAN_ARR_EMPTY;
+
+ default:
+ return new boolean[len];
+ }
+ }
+ };
+
+ /** */
+ private static final Object NULL = new Object();
+
+ /** */
+ private final MessageFactory msgFactory;
+
+ /** */
+ private ByteBuffer buf;
+
+ /** */
+ private byte[] heapArr;
+
+ /** */
+ private long baseOff;
+
+ /** */
+ private int arrOff = -1;
+
+ /** */
+ private Object tmpArr;
+
+ /** */
+ private int tmpArrOff;
+
+ /** */
+ private int tmpArrBytes;
+
+ /** */
+ private boolean msgTypeDone;
+
+ /** */
+ private Message msg;
+
+ /** */
+ private Iterator<?> mapIt;
+
+ /** */
+ private Iterator<?> it;
+
+ /** */
+ private Iterator<?> arrIt;
+
+ /** */
+ private Object arrCur = NULL;
+
+ /** */
+ private Object mapCur = NULL;
+
+ /** */
+ private Object cur = NULL;
+
+ /** */
+ private boolean keyDone;
+
+ /** */
+ private int readSize = -1;
+
+ /** */
+ private int readItems;
+
+ /** */
+ private Object[] objArr;
+
+ /** */
+ private Collection<Object> col;
+
+ /** */
+ private Map<Object, Object> map;
+
+ /** */
+ private boolean lastFinished;
+
+ /**
+ * @param msgFactory Message factory.
+ */
+ public DirectByteBufferStreamImplV1(MessageFactory msgFactory) {
+ this.msgFactory = msgFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setBuffer(ByteBuffer buf) {
+ assert buf != null;
+
+ if (this.buf != buf) {
+ this.buf = buf;
+
+ heapArr = buf.isDirect() ? null : buf.array();
+ baseOff = buf.isDirect() ? ((DirectBuffer)buf).address() : BYTE_ARR_OFF;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public int remaining() {
+ return buf.remaining();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean lastFinished() {
+ return lastFinished;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeByte(byte val) {
+ lastFinished = buf.remaining() >= 1;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putByte(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 1);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeShort(short val) {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putShort(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 2);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeInt(int val) {
+ lastFinished = buf.remaining() >= 4;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putInt(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 4);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeLong(long val) {
+ lastFinished = buf.remaining() >= 8;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putLong(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 8);
+ }
+ }
+
+ /**
+ * @param val Value.
+ */
+ @Override public void writeFloat(float val) {
+ lastFinished = buf.remaining() >= 4;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putFloat(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 4);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeDouble(double val) {
+ lastFinished = buf.remaining() >= 8;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putDouble(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 8);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeChar(char val) {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putChar(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 2);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeBoolean(boolean val) {
+ lastFinished = buf.remaining() >= 1;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putBoolean(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 1);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeByteArray(byte[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, BYTE_ARR_OFF, val.length, val.length);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeByteArray(byte[] val, long off, int len) {
+ if (val != null)
+ lastFinished = writeArray(val, BYTE_ARR_OFF + off, len, len);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeShortArray(short[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, SHORT_ARR_OFF, val.length, val.length << 1);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeIntArray(int[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, INT_ARR_OFF, val.length, val.length << 2);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeLongArray(long[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, LONG_ARR_OFF, val.length, val.length << 3);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeFloatArray(float[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, FLOAT_ARR_OFF, val.length, val.length << 2);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeDoubleArray(double[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, DOUBLE_ARR_OFF, val.length, val.length << 3);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeCharArray(char[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, CHAR_ARR_OFF, val.length, val.length << 1);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeBooleanArray(boolean[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, BOOLEAN_ARR_OFF, val.length, val.length);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeString(String val) {
+ writeByteArray(val != null ? val.getBytes() : null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeBitSet(BitSet val) {
+ writeLongArray(val != null ? val.toLongArray() : null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeUuid(UUID val) {
+ writeByteArray(val != null ? U.uuidToBytes(val) : null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeIgniteUuid(IgniteUuid val) {
+ writeByteArray(val != null ? U.igniteUuidToBytes(val) : null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeMessage(Message msg, MessageWriter writer) {
+ if (msg != null) {
+ if (buf.hasRemaining()) {
+ try {
+ writer.beforeInnerMessageWrite();
+
+ lastFinished = msg.writeTo(buf, writer);
+ }
+ finally {
+ writer.afterInnerMessageWrite(lastFinished);
+ }
+ }
+ else
+ lastFinished = false;
+ }
+ else
+ writeByte(Byte.MIN_VALUE);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> void writeObjectArray(T[] arr, MessageCollectionItemType itemType, MessageWriter writer) {
+ if (arr != null) {
+ if (arrIt == null) {
+ writeInt(arr.length);
+
+ if (!lastFinished)
+ return;
+
+ arrIt = arrayIterator(arr);
+ }
+
+ while (arrIt.hasNext() || arrCur != NULL) {
+ if (arrCur == NULL)
+ arrCur = arrIt.next();
+
+ write(itemType, arrCur, writer);
+
+ if (!lastFinished)
+ return;
+
+ arrCur = NULL;
+ }
+
+ arrIt = null;
+ }
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> void writeCollection(Collection<T> col, MessageCollectionItemType itemType,
+ MessageWriter writer) {
+ if (col != null) {
+ if (it == null) {
+ writeInt(col.size());
+
+ if (!lastFinished)
+ return;
+
+ it = col.iterator();
+ }
+
+ while (it.hasNext() || cur != NULL) {
+ if (cur == NULL)
+ cur = it.next();
+
+ write(itemType, cur, writer);
+
+ if (!lastFinished)
+ return;
+
+ cur = NULL;
+ }
+
+ it = null;
+ }
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <K, V> void writeMap(Map<K, V> map, MessageCollectionItemType keyType,
+ MessageCollectionItemType valType, MessageWriter writer) {
+ if (map != null) {
+ if (mapIt == null) {
+ writeInt(map.size());
+
+ if (!lastFinished)
+ return;
+
+ mapIt = map.entrySet().iterator();
+ }
+
+ while (mapIt.hasNext() || mapCur != NULL) {
+ Map.Entry<K, V> e;
+
+ if (mapCur == NULL)
+ mapCur = mapIt.next();
+
+ e = (Map.Entry<K, V>)mapCur;
+
+ if (!keyDone) {
+ write(keyType, e.getKey(), writer);
+
+ if (!lastFinished)
+ return;
+
+ keyDone = true;
+ }
+
+ write(valType, e.getValue(), writer);
+
+ if (!lastFinished)
+ return;
+
+ mapCur = NULL;
+ keyDone = false;
+ }
+
+ mapIt = null;
+ }
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte readByte() {
+ lastFinished = buf.remaining() >= 1;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 1);
+
+ return UNSAFE.getByte(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short readShort() {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 2);
+
+ return UNSAFE.getShort(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int readInt() {
+ lastFinished = buf.remaining() >= 4;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 4);
+
+ return UNSAFE.getInt(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long readLong() {
+ lastFinished = buf.remaining() >= 8;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 8);
+
+ return UNSAFE.getLong(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public float readFloat() {
+ lastFinished = buf.remaining() >= 4;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 4);
+
+ return UNSAFE.getFloat(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public double readDouble() {
+ lastFinished = buf.remaining() >= 8;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 8);
+
+ return UNSAFE.getDouble(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public char readChar() {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 2);
+
+ return UNSAFE.getChar(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readBoolean() {
+ lastFinished = buf.hasRemaining();
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 1);
+
+ return UNSAFE.getBoolean(heapArr, baseOff + pos);
+ }
+ else
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte[] readByteArray() {
+ return readArray(BYTE_ARR_CREATOR, 0, BYTE_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short[] readShortArray() {
+ return readArray(SHORT_ARR_CREATOR, 1, SHORT_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int[] readIntArray() {
+ return readArray(INT_ARR_CREATOR, 2, INT_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long[] readLongArray() {
+ return readArray(LONG_ARR_CREATOR, 3, LONG_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public float[] readFloatArray() {
+ return readArray(FLOAT_ARR_CREATOR, 2, FLOAT_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public double[] readDoubleArray() {
+ return readArray(DOUBLE_ARR_CREATOR, 3, DOUBLE_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public char[] readCharArray() {
+ return readArray(CHAR_ARR_CREATOR, 1, CHAR_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean[] readBooleanArray() {
+ return readArray(BOOLEAN_ARR_CREATOR, 0, BOOLEAN_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String readString() {
+ byte[] arr = readByteArray();
+
+ return arr != null ? new String(arr) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public BitSet readBitSet() {
+ long[] arr = readLongArray();
+
+ return arr != null ? BitSet.valueOf(arr) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public UUID readUuid() {
+ byte[] arr = readByteArray();
+
+ return arr != null ? U.bytesToUuid(arr, 0) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid readIgniteUuid() {
+ byte[] arr = readByteArray();
+
+ return arr != null ? U.bytesToIgniteUuid(arr, 0) : null;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <T extends Message> T readMessage(MessageReader reader) {
+ if (!msgTypeDone) {
+ if (!buf.hasRemaining()) {
+ lastFinished = false;
+
+ return null;
+ }
+
+ byte type = readByte();
+
+ msg = type == Byte.MIN_VALUE ? null : msgFactory.create(type);
+
+ msgTypeDone = true;
+ }
+
+ if (msg != null) {
+ try {
+ reader.beforeInnerMessageRead();
+
+ reader.setCurrentReadClass(msg.getClass());
+
+ lastFinished = msg.readFrom(buf, reader);
+ }
+ finally {
+ reader.afterInnerMessageRead(lastFinished);
+ }
+ }
+ else
+ lastFinished = true;
+
+ if (lastFinished) {
+ Message msg0 = msg;
+
+ msgTypeDone = false;
+ msg = null;
+
+ return (T)msg0;
+ }
+ else
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <T> T[] readObjectArray(MessageCollectionItemType itemType, Class<T> itemCls,
+ MessageReader reader) {
+ if (readSize == -1) {
+ int size = readInt();
+
+ if (!lastFinished)
+ return null;
+
+ readSize = size;
+ }
+
+ if (readSize >= 0) {
+ if (objArr == null)
+ objArr = itemCls != null ? (Object[])Array.newInstance(itemCls, readSize) : new Object[readSize];
+
+ for (int i = readItems; i < readSize; i++) {
+ Object item = read(itemType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ objArr[i] = item;
+
+ readItems++;
+ }
+ }
+
+ readSize = -1;
+ readItems = 0;
+ cur = null;
+
+ T[] objArr0 = (T[])objArr;
+
+ objArr = null;
+
+ return objArr0;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <C extends Collection<?>> C readCollection(MessageCollectionItemType itemType,
+ MessageReader reader) {
+ if (readSize == -1) {
+ int size = readInt();
+
+ if (!lastFinished)
+ return null;
+
+ readSize = size;
+ }
+
+ if (readSize >= 0) {
+ if (col == null)
+ col = new ArrayList<>(readSize);
+
+ for (int i = readItems; i < readSize; i++) {
+ Object item = read(itemType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ col.add(item);
+
+ readItems++;
+ }
+ }
+
+ readSize = -1;
+ readItems = 0;
+ cur = null;
+
+ C col0 = (C)col;
+
+ col = null;
+
+ return col0;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <M extends Map<?, ?>> M readMap(MessageCollectionItemType keyType,
+ MessageCollectionItemType valType, boolean linked, MessageReader reader) {
+ if (readSize == -1) {
+ int size = readInt();
+
+ if (!lastFinished)
+ return null;
+
+ readSize = size;
+ }
+
+ if (readSize >= 0) {
+ if (map == null)
+ map = linked ? U.newLinkedHashMap(readSize) : U.newHashMap(readSize);
+
+ for (int i = readItems; i < readSize; i++) {
+ if (!keyDone) {
+ Object key = read(keyType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ mapCur = key;
+ keyDone = true;
+ }
+
+ Object val = read(valType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ map.put(mapCur, val);
+
+ keyDone = false;
+
+ readItems++;
+ }
+ }
+
+ readSize = -1;
+ readItems = 0;
+ mapCur = null;
+
+ M map0 = (M)map;
+
+ map = null;
+
+ return map0;
+ }
+
+ /**
+ * @param arr Array.
+ * @param off Offset.
+ * @param len Length.
+ * @param bytes Length in bytes.
+ * @return Whether array was fully written
+ */
+ private boolean writeArray(Object arr, long off, int len, int bytes) {
+ assert arr != null;
+ assert arr.getClass().isArray() && arr.getClass().getComponentType().isPrimitive();
+ assert off > 0;
+ assert len >= 0;
+ assert bytes >= 0;
+ assert bytes >= arrOff;
+
+ if (arrOff == -1) {
+ if (buf.remaining() < 4)
+ return false;
+
+ writeInt(len);
+
+ arrOff = 0;
+ }
+
+ int toWrite = bytes - arrOff;
+ int pos = buf.position();
+ int remaining = buf.remaining();
+
+ if (toWrite <= remaining) {
+ UNSAFE.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, toWrite);
+
+ pos += toWrite;
+
+ buf.position(pos);
+
+ arrOff = -1;
+
+ return true;
+ }
+ else {
+ UNSAFE.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, remaining);
+
+ pos += remaining;
+
+ buf.position(pos);
+
+ arrOff += remaining;
+
+ return false;
+ }
+ }
+
+ /**
+ * @param creator Array creator.
+ * @param lenShift Array length shift size.
+ * @param off Base offset.
+ * @return Array or special value if it was not fully read.
+ */
+ @SuppressWarnings("unchecked")
+ private <T> T readArray(ArrayCreator<T> creator, int lenShift, long off) {
+ assert creator != null;
+
+ if (tmpArr == null) {
+ if (buf.remaining() < 4) {
+ lastFinished = false;
+
+ return null;
+ }
+
+ int len = readInt();
+
+ switch (len) {
+ case -1:
+ lastFinished = true;
+
+ return null;
+
+ case 0:
+ lastFinished = true;
+
+ return creator.create(0);
+
+ default:
+ tmpArr = creator.create(len);
+ tmpArrBytes = len << lenShift;
+ }
+ }
+
+ int toRead = tmpArrBytes - tmpArrOff;
+ int remaining = buf.remaining();
+ int pos = buf.position();
+
+ lastFinished = toRead <= remaining;
+
+ if (lastFinished) {
+ UNSAFE.copyMemory(heapArr, baseOff + pos, tmpArr, off + tmpArrOff, toRead);
+
+ buf.position(pos + toRead);
+
+ T arr = (T)tmpArr;
+
+ tmpArr = null;
+ tmpArrBytes = 0;
+ tmpArrOff = 0;
+
+ return arr;
+ }
+ else {
+ UNSAFE.copyMemory(heapArr, baseOff + pos, tmpArr, off + tmpArrOff, remaining);
+
+ buf.position(pos + remaining);
+
+ tmpArrOff += remaining;
+
+ return null;
+ }
+ }
+
+ /**
+ * @param type Type.
+ * @param val Value.
+ * @param writer Writer.
+ */
+ private void write(MessageCollectionItemType type, Object val, MessageWriter writer) {
+ switch (type) {
+ case BYTE:
+ writeByte((Byte)val);
+
+ break;
+
+ case SHORT:
+ writeShort((Short)val);
+
+ break;
+
+ case INT:
+ writeInt((Integer)val);
+
+ break;
+
+ case LONG:
+ writeLong((Long)val);
+
+ break;
+
+ case FLOAT:
+ writeFloat((Float)val);
+
+ break;
+
+ case DOUBLE:
+ writeDouble((Double)val);
+
+ break;
+
+ case CHAR:
+ writeChar((Character)val);
+
+ break;
+
+ case BOOLEAN:
+ writeBoolean((Boolean)val);
+
+ break;
+
+ case BYTE_ARR:
+ writeByteArray((byte[])val);
+
+ break;
+
+ case SHORT_ARR:
+ writeShortArray((short[])val);
+
+ break;
+
+ case INT_ARR:
+ writeIntArray((int[])val);
+
+ break;
+
+ case LONG_ARR:
+ writeLongArray((long[])val);
+
+ break;
+
+ case FLOAT_ARR:
+ writeFloatArray((float[])val);
+
+ break;
+
+ case DOUBLE_ARR:
+ writeDoubleArray((double[])val);
+
+ break;
+
+ case CHAR_ARR:
+ writeCharArray((char[])val);
+
+ break;
+
+ case BOOLEAN_ARR:
+ writeBooleanArray((boolean[])val);
+
+ break;
+
+ case STRING:
+ writeString((String)val);
+
+ break;
+
+ case BIT_SET:
+ writeBitSet((BitSet)val);
+
+ break;
+
+ case UUID:
+ writeUuid((UUID)val);
+
+ break;
+
+ case IGNITE_UUID:
+ writeIgniteUuid((IgniteUuid)val);
+
+ break;
+
+ case MSG:
+ try {
+ if (val != null)
+ writer.beforeInnerMessageWrite();
+
+ writeMessage((Message)val, writer);
+ }
+ finally {
+ if (val != null)
+ writer.afterInnerMessageWrite(lastFinished);
+ }
+
+ break;
+
+ default:
+ throw new IllegalArgumentException("Unknown type: " + type);
+ }
+ }
+
+ /**
+ * @param type Type.
+ * @param reader Reader.
+ * @return Value.
+ */
+ private Object read(MessageCollectionItemType type, MessageReader reader) {
+ switch (type) {
+ case BYTE:
+ return readByte();
+
+ case SHORT:
+ return readShort();
+
+ case INT:
+ return readInt();
+
+ case LONG:
+ return readLong();
+
+ case FLOAT:
+ return readFloat();
+
+ case DOUBLE:
+ return readDouble();
+
+ case CHAR:
+ return readChar();
+
+ case BOOLEAN:
+ return readBoolean();
+
+ case BYTE_ARR:
+ return readByteArray();
+
+ case SHORT_ARR:
+ return readShortArray();
+
+ case INT_ARR:
+ return readIntArray();
+
+ case LONG_ARR:
+ return readLongArray();
+
+ case FLOAT_ARR:
+ return readFloatArray();
+
+ case DOUBLE_ARR:
+ return readDoubleArray();
+
+ case CHAR_ARR:
+ return readCharArray();
+
+ case BOOLEAN_ARR:
+ return readBooleanArray();
+
+ case STRING:
+ return readString();
+
+ case BIT_SET:
+ return readBitSet();
+
+ case UUID:
+ return readUuid();
+
+ case IGNITE_UUID:
+ return readIgniteUuid();
+
+ case MSG:
+ return readMessage(reader);
+
+ default:
+ throw new IllegalArgumentException("Unknown type: " + type);
+ }
+ }
+
+ /**
+ * @param arr Array.
+ * @return Array iterator.
+ */
+ private Iterator<?> arrayIterator(final Object[] arr) {
+ return new Iterator<Object>() {
+ private int idx;
+
+ @Override public boolean hasNext() {
+ return idx < arr.length;
+ }
+
+ @Override public Object next() {
+ if (!hasNext())
+ throw new NoSuchElementException();
+
+ return arr[idx++];
+ }
+
+ @Override public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+
+ /**
+ * Array creator.
+ */
+ private static interface ArrayCreator<T> {
+ /**
+ * @param len Array length or {@code -1} if array was not fully read.
+ * @return New array.
+ */
+ public T create(int len);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java
new file mode 100644
index 0000000..89c9cc6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java
@@ -0,0 +1,1583 @@
+/*
+ * 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.direct.stream.v2;
+
+import java.lang.reflect.Array;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.RandomAccess;
+import java.util.UUID;
+import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
+import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.internal.util.typedef.internal.U;
+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.MessageFactory;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
+
+/**
+ * Direct marshalling I/O stream (version 2).
+ */
+public class DirectByteBufferStreamImplV2 implements DirectByteBufferStream {
+ /** */
+ private static final Unsafe UNSAFE = GridUnsafe.unsafe();
+
+ /** */
+ private static final long BYTE_ARR_OFF = UNSAFE.arrayBaseOffset(byte[].class);
+
+ /** */
+ private static final long SHORT_ARR_OFF = UNSAFE.arrayBaseOffset(short[].class);
+
+ /** */
+ private static final long INT_ARR_OFF = UNSAFE.arrayBaseOffset(int[].class);
+
+ /** */
+ private static final long LONG_ARR_OFF = UNSAFE.arrayBaseOffset(long[].class);
+
+ /** */
+ private static final long FLOAT_ARR_OFF = UNSAFE.arrayBaseOffset(float[].class);
+
+ /** */
+ private static final long DOUBLE_ARR_OFF = UNSAFE.arrayBaseOffset(double[].class);
+
+ /** */
+ private static final long CHAR_ARR_OFF = UNSAFE.arrayBaseOffset(char[].class);
+
+ /** */
+ private static final long BOOLEAN_ARR_OFF = UNSAFE.arrayBaseOffset(boolean[].class);
+
+ /** */
+ private static final byte[] BYTE_ARR_EMPTY = new byte[0];
+
+ /** */
+ private static final short[] SHORT_ARR_EMPTY = new short[0];
+
+ /** */
+ private static final int[] INT_ARR_EMPTY = U.EMPTY_INTS;
+
+ /** */
+ private static final long[] LONG_ARR_EMPTY = U.EMPTY_LONGS;
+
+ /** */
+ private static final float[] FLOAT_ARR_EMPTY = new float[0];
+
+ /** */
+ private static final double[] DOUBLE_ARR_EMPTY = new double[0];
+
+ /** */
+ private static final char[] CHAR_ARR_EMPTY = new char[0];
+
+ /** */
+ private static final boolean[] BOOLEAN_ARR_EMPTY = new boolean[0];
+
+ /** */
+ private static final ArrayCreator<byte[]> BYTE_ARR_CREATOR = new ArrayCreator<byte[]>() {
+ @Override public byte[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return BYTE_ARR_EMPTY;
+
+ default:
+ return new byte[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<short[]> SHORT_ARR_CREATOR = new ArrayCreator<short[]>() {
+ @Override public short[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return SHORT_ARR_EMPTY;
+
+ default:
+ return new short[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<int[]> INT_ARR_CREATOR = new ArrayCreator<int[]>() {
+ @Override public int[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return INT_ARR_EMPTY;
+
+ default:
+ return new int[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<long[]> LONG_ARR_CREATOR = new ArrayCreator<long[]>() {
+ @Override public long[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return LONG_ARR_EMPTY;
+
+ default:
+ return new long[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<float[]> FLOAT_ARR_CREATOR = new ArrayCreator<float[]>() {
+ @Override public float[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return FLOAT_ARR_EMPTY;
+
+ default:
+ return new float[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<double[]> DOUBLE_ARR_CREATOR = new ArrayCreator<double[]>() {
+ @Override public double[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return DOUBLE_ARR_EMPTY;
+
+ default:
+ return new double[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<char[]> CHAR_ARR_CREATOR = new ArrayCreator<char[]>() {
+ @Override public char[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return CHAR_ARR_EMPTY;
+
+ default:
+ return new char[len];
+ }
+ }
+ };
+
+ /** */
+ private static final ArrayCreator<boolean[]> BOOLEAN_ARR_CREATOR = new ArrayCreator<boolean[]>() {
+ @Override public boolean[] create(int len) {
+ assert len >= 0;
+
+ switch (len) {
+ case 0:
+ return BOOLEAN_ARR_EMPTY;
+
+ default:
+ return new boolean[len];
+ }
+ }
+ };
+
+ /** */
+ private static final Object NULL = new Object();
+
+ /** */
+ private final MessageFactory msgFactory;
+
+ /** */
+ private ByteBuffer buf;
+
+ /** */
+ private byte[] heapArr;
+
+ /** */
+ private long baseOff;
+
+ /** */
+ private int arrOff = -1;
+
+ /** */
+ private Object tmpArr;
+
+ /** */
+ private int tmpArrOff;
+
+ /** */
+ private int tmpArrBytes;
+
+ /** */
+ private boolean msgTypeDone;
+
+ /** */
+ private Message msg;
+
+ /** */
+ private Iterator<?> mapIt;
+
+ /** */
+ private Iterator<?> it;
+
+ /** */
+ private int arrPos = -1;
+
+ /** */
+ private Object arrCur = NULL;
+
+ /** */
+ private Object mapCur = NULL;
+
+ /** */
+ private Object cur = NULL;
+
+ /** */
+ private boolean keyDone;
+
+ /** */
+ private int readSize = -1;
+
+ /** */
+ private int readItems;
+
+ /** */
+ private Object[] objArr;
+
+ /** */
+ private Collection<Object> col;
+
+ /** */
+ private Map<Object, Object> map;
+
+ /** */
+ private long prim;
+
+ /** */
+ private int primShift;
+
+ /** */
+ private int uuidState;
+
+ /** */
+ private long uuidMost;
+
+ /** */
+ private long uuidLeast;
+
+ /** */
+ private long uuidLocId;
+
+ /** */
+ private boolean lastFinished;
+
+ /**
+ * @param msgFactory Message factory.
+ */
+ public DirectByteBufferStreamImplV2(MessageFactory msgFactory) {
+ this.msgFactory = msgFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setBuffer(ByteBuffer buf) {
+ assert buf != null;
+
+ if (this.buf != buf) {
+ this.buf = buf;
+
+ heapArr = buf.isDirect() ? null : buf.array();
+ baseOff = buf.isDirect() ? ((DirectBuffer)buf).address() : BYTE_ARR_OFF;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public int remaining() {
+ return buf.remaining();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean lastFinished() {
+ return lastFinished;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeByte(byte val) {
+ lastFinished = buf.remaining() >= 1;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putByte(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 1);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeShort(short val) {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putShort(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 2);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeInt(int val) {
+ lastFinished = buf.remaining() >= 5;
+
+ if (lastFinished) {
+ if (val == Integer.MAX_VALUE)
+ val = Integer.MIN_VALUE;
+ else
+ val++;
+
+ int pos = buf.position();
+
+ while ((val & 0xFFFF_FF80) != 0) {
+ byte b = (byte)(val | 0x80);
+
+ UNSAFE.putByte(heapArr, baseOff + pos++, b);
+
+ val >>>= 7;
+ }
+
+ UNSAFE.putByte(heapArr, baseOff + pos++, (byte)val);
+
+ buf.position(pos);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeLong(long val) {
+ lastFinished = buf.remaining() >= 10;
+
+ if (lastFinished) {
+ if (val == Long.MAX_VALUE)
+ val = Long.MIN_VALUE;
+ else
+ val++;
+
+ int pos = buf.position();
+
+ while ((val & 0xFFFF_FFFF_FFFF_FF80L) != 0) {
+ byte b = (byte)(val | 0x80);
+
+ UNSAFE.putByte(heapArr, baseOff + pos++, b);
+
+ val >>>= 7;
+ }
+
+ UNSAFE.putByte(heapArr, baseOff + pos++, (byte)val);
+
+ buf.position(pos);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeFloat(float val) {
+ lastFinished = buf.remaining() >= 4;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putFloat(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 4);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeDouble(double val) {
+ lastFinished = buf.remaining() >= 8;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putDouble(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 8);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeChar(char val) {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putChar(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 2);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeBoolean(boolean val) {
+ lastFinished = buf.remaining() >= 1;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ UNSAFE.putBoolean(heapArr, baseOff + pos, val);
+
+ buf.position(pos + 1);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeByteArray(byte[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, BYTE_ARR_OFF, val.length, val.length);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeByteArray(byte[] val, long off, int len) {
+ if (val != null)
+ lastFinished = writeArray(val, BYTE_ARR_OFF + off, len, len);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeShortArray(short[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, SHORT_ARR_OFF, val.length, val.length << 1);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeIntArray(int[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, INT_ARR_OFF, val.length, val.length << 2);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeLongArray(long[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, LONG_ARR_OFF, val.length, val.length << 3);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeFloatArray(float[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, FLOAT_ARR_OFF, val.length, val.length << 2);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeDoubleArray(double[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, DOUBLE_ARR_OFF, val.length, val.length << 3);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeCharArray(char[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, CHAR_ARR_OFF, val.length, val.length << 1);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeBooleanArray(boolean[] val) {
+ if (val != null)
+ lastFinished = writeArray(val, BOOLEAN_ARR_OFF, val.length, val.length);
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeString(String val) {
+ writeByteArray(val != null ? val.getBytes() : null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeBitSet(BitSet val) {
+ writeLongArray(val != null ? val.toLongArray() : null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeUuid(UUID val) {
+ switch (uuidState) {
+ case 0:
+ writeBoolean(val == null);
+
+ if (!lastFinished || val == null)
+ return;
+
+ uuidState++;
+
+ case 1:
+ writeLong(val.getMostSignificantBits());
+
+ if (!lastFinished)
+ return;
+
+ uuidState++;
+
+ case 2:
+ writeLong(val.getLeastSignificantBits());
+
+ if (!lastFinished)
+ return;
+
+ uuidState = 0;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeIgniteUuid(IgniteUuid val) {
+ switch (uuidState) {
+ case 0:
+ writeBoolean(val == null);
+
+ if (!lastFinished || val == null)
+ return;
+
+ uuidState++;
+
+ case 1:
+ writeLong(val.globalId().getMostSignificantBits());
+
+ if (!lastFinished)
+ return;
+
+ uuidState++;
+
+ case 2:
+ writeLong(val.globalId().getLeastSignificantBits());
+
+ if (!lastFinished)
+ return;
+
+ uuidState++;
+
+ case 3:
+ writeLong(val.localId());
+
+ if (!lastFinished)
+ return;
+
+ uuidState = 0;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeMessage(Message msg, MessageWriter writer) {
+ if (msg != null) {
+ if (buf.hasRemaining()) {
+ try {
+ writer.beforeInnerMessageWrite();
+
+ lastFinished = msg.writeTo(buf, writer);
+ }
+ finally {
+ writer.afterInnerMessageWrite(lastFinished);
+ }
+ }
+ else
+ lastFinished = false;
+ }
+ else
+ writeByte(Byte.MIN_VALUE);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> void writeObjectArray(T[] arr, MessageCollectionItemType itemType,
+ MessageWriter writer) {
+ if (arr != null) {
+ int len = arr.length;
+
+ if (arrPos == -1) {
+ writeInt(len);
+
+ if (!lastFinished)
+ return;
+
+ arrPos = 0;
+ }
+
+ while (arrPos < len || arrCur != NULL) {
+ if (arrCur == NULL)
+ arrCur = arr[arrPos++];
+
+ write(itemType, arrCur, writer);
+
+ if (!lastFinished)
+ return;
+
+ arrCur = NULL;
+ }
+
+ arrPos = -1;
+ }
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> void writeCollection(Collection<T> col, MessageCollectionItemType itemType,
+ MessageWriter writer) {
+ if (col != null) {
+ if (col instanceof List && col instanceof RandomAccess)
+ writeRandomAccessList((List<T>)col, itemType, writer);
+ else {
+ if (it == null) {
+ writeInt(col.size());
+
+ if (!lastFinished)
+ return;
+
+ it = col.iterator();
+ }
+
+ while (it.hasNext() || cur != NULL) {
+ if (cur == NULL)
+ cur = it.next();
+
+ write(itemType, cur, writer);
+
+ if (!lastFinished)
+ return;
+
+ cur = NULL;
+ }
+
+ it = null;
+ }
+ }
+ else
+ writeInt(-1);
+ }
+
+ /**
+ * @param list List.
+ * @param itemType Component type.
+ * @param writer Writer.
+ */
+ private <T> void writeRandomAccessList(List<T> list, MessageCollectionItemType itemType, MessageWriter writer) {
+ assert list instanceof RandomAccess;
+
+ int size = list.size();
+
+ if (arrPos == -1) {
+ writeInt(size);
+
+ if (!lastFinished)
+ return;
+
+ arrPos = 0;
+ }
+
+ while (arrPos < size || arrCur != NULL) {
+ if (arrCur == NULL)
+ arrCur = list.get(arrPos++);
+
+ write(itemType, arrCur, writer);
+
+ if (!lastFinished)
+ return;
+
+ arrCur = NULL;
+ }
+
+ arrPos = -1;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <K, V> void writeMap(Map<K, V> map, MessageCollectionItemType keyType,
+ MessageCollectionItemType valType, MessageWriter writer) {
+ if (map != null) {
+ if (mapIt == null) {
+ writeInt(map.size());
+
+ if (!lastFinished)
+ return;
+
+ mapIt = map.entrySet().iterator();
+ }
+
+ while (mapIt.hasNext() || mapCur != NULL) {
+ Map.Entry<K, V> e;
+
+ if (mapCur == NULL)
+ mapCur = mapIt.next();
+
+ e = (Map.Entry<K, V>)mapCur;
+
+ if (!keyDone) {
+ write(keyType, e.getKey(), writer);
+
+ if (!lastFinished)
+ return;
+
+ keyDone = true;
+ }
+
+ write(valType, e.getValue(), writer);
+
+ if (!lastFinished)
+ return;
+
+ mapCur = NULL;
+ keyDone = false;
+ }
+
+ mapIt = null;
+ }
+ else
+ writeInt(-1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte readByte() {
+ lastFinished = buf.remaining() >= 1;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 1);
+
+ return UNSAFE.getByte(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short readShort() {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 2);
+
+ return UNSAFE.getShort(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int readInt() {
+ lastFinished = false;
+
+ int val = 0;
+
+ while (buf.hasRemaining()) {
+ int pos = buf.position();
+
+ byte b = UNSAFE.getByte(heapArr, baseOff + pos);
+
+ buf.position(pos + 1);
+
+ prim |= ((long)b & 0x7F) << (7 * primShift);
+
+ if ((b & 0x80) == 0) {
+ lastFinished = true;
+
+ val = (int)prim;
+
+ if (val == Integer.MIN_VALUE)
+ val = Integer.MAX_VALUE;
+ else
+ val--;
+
+ prim = 0;
+ primShift = 0;
+
+ break;
+ }
+ else
+ primShift++;
+ }
+
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long readLong() {
+ lastFinished = false;
+
+ long val = 0;
+
+ while (buf.hasRemaining()) {
+ int pos = buf.position();
+
+ byte b = UNSAFE.getByte(heapArr, baseOff + pos);
+
+ buf.position(pos + 1);
+
+ prim |= ((long)b & 0x7F) << (7 * primShift);
+
+ if ((b & 0x80) == 0) {
+ lastFinished = true;
+
+ val = prim;
+
+ if (val == Long.MIN_VALUE)
+ val = Long.MAX_VALUE;
+ else
+ val--;
+
+ prim = 0;
+ primShift = 0;
+
+ break;
+ }
+ else
+ primShift++;
+ }
+
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public float readFloat() {
+ lastFinished = buf.remaining() >= 4;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 4);
+
+ return UNSAFE.getFloat(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public double readDouble() {
+ lastFinished = buf.remaining() >= 8;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 8);
+
+ return UNSAFE.getDouble(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public char readChar() {
+ lastFinished = buf.remaining() >= 2;
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 2);
+
+ return UNSAFE.getChar(heapArr, baseOff + pos);
+ }
+ else
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readBoolean() {
+ lastFinished = buf.hasRemaining();
+
+ if (lastFinished) {
+ int pos = buf.position();
+
+ buf.position(pos + 1);
+
+ return UNSAFE.getBoolean(heapArr, baseOff + pos);
+ }
+ else
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte[] readByteArray() {
+ return readArray(BYTE_ARR_CREATOR, 0, BYTE_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short[] readShortArray() {
+ return readArray(SHORT_ARR_CREATOR, 1, SHORT_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int[] readIntArray() {
+ return readArray(INT_ARR_CREATOR, 2, INT_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long[] readLongArray() {
+ return readArray(LONG_ARR_CREATOR, 3, LONG_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public float[] readFloatArray() {
+ return readArray(FLOAT_ARR_CREATOR, 2, FLOAT_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public double[] readDoubleArray() {
+ return readArray(DOUBLE_ARR_CREATOR, 3, DOUBLE_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public char[] readCharArray() {
+ return readArray(CHAR_ARR_CREATOR, 1, CHAR_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean[] readBooleanArray() {
+ return readArray(BOOLEAN_ARR_CREATOR, 0, BOOLEAN_ARR_OFF);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String readString() {
+ byte[] arr = readByteArray();
+
+ return arr != null ? new String(arr) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public BitSet readBitSet() {
+ long[] arr = readLongArray();
+
+ return arr != null ? BitSet.valueOf(arr) : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public UUID readUuid() {
+ switch (uuidState) {
+ case 0:
+ boolean isNull = readBoolean();
+
+ if (!lastFinished || isNull)
+ return null;
+
+ uuidState++;
+
+ case 1:
+ uuidMost = readLong();
+
+ if (!lastFinished)
+ return null;
+
+ uuidState++;
+
+ case 2:
+ uuidLeast = readLong();
+
+ if (!lastFinished)
+ return null;
+
+ uuidState = 0;
+ }
+
+ UUID val = new UUID(uuidMost, uuidLeast);
+
+ uuidMost = 0;
+ uuidLeast = 0;
+
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid readIgniteUuid() {
+ switch (uuidState) {
+ case 0:
+ boolean isNull = readBoolean();
+
+ if (!lastFinished || isNull)
+ return null;
+
+ uuidState++;
+
+ case 1:
+ uuidMost = readLong();
+
+ if (!lastFinished)
+ return null;
+
+ uuidState++;
+
+ case 2:
+ uuidLeast = readLong();
+
+ if (!lastFinished)
+ return null;
+
+ uuidState++;
+
+ case 3:
+ uuidLocId = readLong();
+
+ if (!lastFinished)
+ return null;
+
+ uuidState = 0;
+ }
+
+ IgniteUuid val = new IgniteUuid(new UUID(uuidMost, uuidLeast), uuidLocId);
+
+ uuidMost = 0;
+ uuidLeast = 0;
+ uuidLocId = 0;
+
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <T extends Message> T readMessage(MessageReader reader) {
+ if (!msgTypeDone) {
+ if (!buf.hasRemaining()) {
+ lastFinished = false;
+
+ return null;
+ }
+
+ byte type = readByte();
+
+ msg = type == Byte.MIN_VALUE ? null : msgFactory.create(type);
+
+ msgTypeDone = true;
+ }
+
+ if (msg != null) {
+ try {
+ reader.beforeInnerMessageRead();
+
+ reader.setCurrentReadClass(msg.getClass());
+
+ lastFinished = msg.readFrom(buf, reader);
+ }
+ finally {
+ reader.afterInnerMessageRead(lastFinished);
+ }
+ }
+ else
+ lastFinished = true;
+
+ if (lastFinished) {
+ Message msg0 = msg;
+
+ msgTypeDone = false;
+ msg = null;
+
+ return (T)msg0;
+ }
+ else
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <T> T[] readObjectArray(MessageCollectionItemType itemType, Class<T> itemCls,
+ MessageReader reader) {
+ if (readSize == -1) {
+ int size = readInt();
+
+ if (!lastFinished)
+ return null;
+
+ readSize = size;
+ }
+
+ if (readSize >= 0) {
+ if (objArr == null)
+ objArr = itemCls != null ? (Object[])Array.newInstance(itemCls, readSize) : new Object[readSize];
+
+ for (int i = readItems; i < readSize; i++) {
+ Object item = read(itemType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ objArr[i] = item;
+
+ readItems++;
+ }
+ }
+
+ readSize = -1;
+ readItems = 0;
+ cur = null;
+
+ T[] objArr0 = (T[])objArr;
+
+ objArr = null;
+
+ return objArr0;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <C extends Collection<?>> C readCollection(MessageCollectionItemType itemType,
+ MessageReader reader) {
+ if (readSize == -1) {
+ int size = readInt();
+
+ if (!lastFinished)
+ return null;
+
+ readSize = size;
+ }
+
+ if (readSize >= 0) {
+ if (col == null)
+ col = new ArrayList<>(readSize);
+
+ for (int i = readItems; i < readSize; i++) {
+ Object item = read(itemType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ col.add(item);
+
+ readItems++;
+ }
+ }
+
+ readSize = -1;
+ readItems = 0;
+ cur = null;
+
+ C col0 = (C)col;
+
+ col = null;
+
+ return col0;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public <M extends Map<?, ?>> M readMap(MessageCollectionItemType keyType,
+ MessageCollectionItemType valType, boolean linked, MessageReader reader) {
+ if (readSize == -1) {
+ int size = readInt();
+
+ if (!lastFinished)
+ return null;
+
+ readSize = size;
+ }
+
+ if (readSize >= 0) {
+ if (map == null)
+ map = linked ? U.newLinkedHashMap(readSize) : U.newHashMap(readSize);
+
+ for (int i = readItems; i < readSize; i++) {
+ if (!keyDone) {
+ Object key = read(keyType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ mapCur = key;
+ keyDone = true;
+ }
+
+ Object val = read(valType, reader);
+
+ if (!lastFinished)
+ return null;
+
+ map.put(mapCur, val);
+
+ keyDone = false;
+
+ readItems++;
+ }
+ }
+
+ readSize = -1;
+ readItems = 0;
+ mapCur = null;
+
+ M map0 = (M)map;
+
+ map = null;
+
+ return map0;
+ }
+
+ /**
+ * @param arr Array.
+ * @param off Offset.
+ * @param len Length.
+ * @param bytes Length in bytes.
+ * @return Whether array was fully written.
+ */
+ private boolean writeArray(Object arr, long off, int len, int bytes) {
+ assert arr != null;
+ assert arr.getClass().isArray() && arr.getClass().getComponentType().isPrimitive();
+ assert off > 0;
+ assert len >= 0;
+ assert bytes >= 0;
+ assert bytes >= arrOff;
+
+ if (arrOff == -1) {
+ writeInt(len);
+
+ if (!lastFinished)
+ return false;
+
+ arrOff = 0;
+ }
+
+ int toWrite = bytes - arrOff;
+ int pos = buf.position();
+ int remaining = buf.remaining();
+
+ if (toWrite <= remaining) {
+ if (toWrite > 0) {
+ UNSAFE.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, toWrite);
+
+ buf.position(pos + toWrite);
+ }
+
+ arrOff = -1;
+
+ return true;
+ }
+ else {
+ if (remaining > 0) {
+ UNSAFE.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, remaining);
+
+ buf.position(pos + remaining);
+
+ arrOff += remaining;
+ }
+
+ return false;
+ }
+ }
+
+ /**
+ * @param creator Array creator.
+ * @param lenShift Array length shift size.
+ * @param off Base offset.
+ * @return Array or special value if it was not fully read.
+ */
+ @SuppressWarnings("unchecked")
+ private <T> T readArray(ArrayCreator<T> creator, int lenShift, long off) {
+ assert creator != null;
+
+ if (tmpArr == null) {
+ int len = readInt();
+
+ if (!lastFinished)
+ return null;
+
+ switch (len) {
+ case -1:
+ lastFinished = true;
+
+ return null;
+
+ case 0:
+ lastFinished = true;
+
+ return creator.create(0);
+
+ default:
+ tmpArr = creator.create(len);
+ tmpArrBytes = len << lenShift;
+ }
+ }
+
+ int toRead = tmpArrBytes - tmpArrOff;
+ int remaining = buf.remaining();
+ int pos = buf.position();
+
+ lastFinished = toRead <= remaining;
+
+ if (lastFinished) {
+ UNSAFE.copyMemory(heapArr, baseOff + pos, tmpArr, off + tmpArrOff, toRead);
+
+ buf.position(pos + toRead);
+
+ T arr = (T)tmpArr;
+
+ tmpArr = null;
+ tmpArrBytes = 0;
+ tmpArrOff = 0;
+
+ return arr;
+ }
+ else {
+ UNSAFE.copyMemory(heapArr, baseOff + pos, tmpArr, off + tmpArrOff, remaining);
+
+ buf.position(pos + remaining);
+
+ tmpArrOff += remaining;
+
+ return null;
+ }
+ }
+
+ /**
+ * @param type Type.
+ * @param val Value.
+ * @param writer Writer.
+ */
+ private void write(MessageCollectionItemType type, Object val, MessageWriter writer) {
+ switch (type) {
+ case BYTE:
+ writeByte((Byte)val);
+
+ break;
+
+ case SHORT:
+ writeShort((Short)val);
+
+ break;
+
+ case INT:
+ writeInt((Integer)val);
+
+ break;
+
+ case LONG:
+ writeLong((Long)val);
+
+ break;
+
+ case FLOAT:
+ writeFloat((Float)val);
+
+ break;
+
+ case DOUBLE:
+ writeDouble((Double)val);
+
+ break;
+
+ case CHAR:
+ writeChar((Character)val);
+
+ break;
+
+ case BOOLEAN:
+ writeBoolean((Boolean)val);
+
+ break;
+
+ case BYTE_ARR:
+ writeByteArray((byte[])val);
+
+ break;
+
+ case SHORT_ARR:
+ writeShortArray((short[])val);
+
+ break;
+
+ case INT_ARR:
+ writeIntArray((int[])val);
+
+ break;
+
+ case LONG_ARR:
+ writeLongArray((long[])val);
+
+ break;
+
+ case FLOAT_ARR:
+ writeFloatArray((float[])val);
+
+ break;
+
+ case DOUBLE_ARR:
+ writeDoubleArray((double[])val);
+
+ break;
+
+ case CHAR_ARR:
+ writeCharArray((char[])val);
+
+ break;
+
+ case BOOLEAN_ARR:
+ writeBooleanArray((boolean[])val);
+
+ break;
+
+ case STRING:
+ writeString((String)val);
+
+ break;
+
+ case BIT_SET:
+ writeBitSet((BitSet)val);
+
+ break;
+
+ case UUID:
+ writeUuid((UUID)val);
+
+ break;
+
+ case IGNITE_UUID:
+ writeIgniteUuid((IgniteUuid)val);
+
+ break;
+
+ case MSG:
+ try {
+ if (val != null)
+ writer.beforeInnerMessageWrite();
+
+ writeMessage((Message)val, writer);
+ }
+ finally {
+ if (val != null)
+ writer.afterInnerMessageWrite(lastFinished);
+ }
+
+ break;
+
+ default:
+ throw new IllegalArgumentException("Unknown type: " + type);
+ }
+ }
+
+ /**
+ * @param type Type.
+ * @param reader Reader.
+ * @return Value.
+ */
+ private Object read(MessageCollectionItemType type, MessageReader reader) {
+ switch (type) {
+ case BYTE:
+ return readByte();
+
+ case SHORT:
+ return readShort();
+
+ case INT:
+ return readInt();
+
+ case LONG:
+ return readLong();
+
+ case FLOAT:
+ return readFloat();
+
+ case DOUBLE:
+ return readDouble();
+
+ case CHAR:
+ return readChar();
+
+ case BOOLEAN:
+ return readBoolean();
+
+ case BYTE_ARR:
+ return readByteArray();
+
+ case SHORT_ARR:
+ return readShortArray();
+
+ case INT_ARR:
+ return readIntArray();
+
+ case LONG_ARR:
+ return readLongArray();
+
+ case FLOAT_ARR:
+ return readFloatArray();
+
+ case DOUBLE_ARR:
+ return readDoubleArray();
+
+ case CHAR_ARR:
+ return readCharArray();
+
+ case BOOLEAN_ARR:
+ return readBooleanArray();
+
+ case STRING:
+ return readString();
+
+ case BIT_SET:
+ return readBitSet();
+
+ case UUID:
+ return readUuid();
+
+ case IGNITE_UUID:
+ return readIgniteUuid();
+
+ case MSG:
+ return readMessage(reader);
+
+ default:
+ throw new IllegalArgumentException("Unknown type: " + type);
+ }
+ }
+
+ /**
+ * Array creator.
+ */
+ private static interface ArrayCreator<T> {
+ /**
+ * @param len Array length or {@code -1} if array was not fully read.
+ * @return New array.
+ */
+ public T create(int len);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
index b8af8da..ea82d7f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
@@ -17,6 +17,26 @@
package org.apache.ignite.internal.managers.communication;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
@@ -64,27 +84,6 @@ import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
import org.jsr166.ConcurrentLinkedDeque8;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Queue;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentLinkedDeque;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
@@ -113,6 +112,12 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
/** Max closed topics to store. */
public static final int MAX_CLOSED_TOPICS = 10240;
+ /** Direct protocol version attribute name. */
+ public static final String DIRECT_PROTO_VER_ATTR = "comm.direct.proto.ver";
+
+ /** Direct protocol version. */
+ public static final byte DIRECT_PROTO_VER = 2;
+
/** Listeners by topic. */
private final ConcurrentMap<Object, GridMessageListener> lsnrMap = new ConcurrentHashMap8<>();
@@ -266,6 +271,8 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
}
});
+ ctx.addNodeAttribute(DIRECT_PROTO_VER_ATTR, DIRECT_PROTO_VER);
+
MessageFormatter[] formatterExt = ctx.plugins().extensions(MessageFormatter.class);
if (formatterExt != null && formatterExt.length > 0) {
@@ -277,12 +284,17 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
}
else {
formatter = new MessageFormatter() {
- @Override public MessageWriter writer() {
- return new DirectMessageWriter();
+ @Override public MessageWriter writer(UUID rmtNodeId) throws IgniteCheckedException {
+ assert rmtNodeId != null;
+
+ return new DirectMessageWriter(U.directProtocolVersion(ctx, rmtNodeId));
}
- @Override public MessageReader reader(MessageFactory factory, Class<? extends Message> msgCls) {
- return new DirectMessageReader(msgFactory, this);
+ @Override public MessageReader reader(UUID rmtNodeId, MessageFactory msgFactory)
+ throws IgniteCheckedException {
+ assert rmtNodeId != null;
+
+ return new DirectMessageReader(msgFactory, U.directProtocolVersion(ctx, rmtNodeId));
}
};
}
@@ -2432,4 +2444,4 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
return S.toString(DelayedMessage.class, this, super.toString());
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index 3c1913a..ced0c2a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -173,6 +173,7 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.compute.ComputeTaskCancelledCheckedException;
import org.apache.ignite.internal.compute.ComputeTaskTimeoutCheckedException;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
+import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
import org.apache.ignite.internal.mxbean.IgniteStandardMXBean;
import org.apache.ignite.internal.processors.cache.GridCacheAttributes;
@@ -9302,4 +9303,34 @@ public abstract class IgniteUtils {
throw new IgniteInterruptedCheckedException(e);
}
}
+
+ /**
+ * Defines which protocol version to use for
+ * communication with the provided node.
+ *
+ * @param ctx Context.
+ * @param nodeId Node ID.
+ * @return Protocol version.
+ * @throws IgniteCheckedException If node doesn't exist.
+ */
+ public static byte directProtocolVersion(GridKernalContext ctx, UUID nodeId) throws IgniteCheckedException {
+ assert nodeId != null;
+
+ ClusterNode node = ctx.discovery().node(nodeId);
+
+ if (node == null)
+ throw new IgniteCheckedException("Failed to define communication protocol version " +
+ "(has node left topology?): " + nodeId);
+
+ assert !node.isLocal();
+
+ Byte attr = node.attribute(GridIoManager.DIRECT_PROTO_VER_ATTR);
+
+ byte rmtProtoVer = attr != null ? attr : 1;
+
+ if (rmtProtoVer < GridIoManager.DIRECT_PROTO_VER)
+ return rmtProtoVer;
+ else
+ return GridIoManager.DIRECT_PROTO_VER;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcToNioAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcToNioAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcToNioAdapter.java
index e732a79..6820dc7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcToNioAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcToNioAdapter.java
@@ -30,13 +30,13 @@ import org.apache.ignite.internal.util.nio.GridNioFilterAdapter;
import org.apache.ignite.internal.util.nio.GridNioFilterChain;
import org.apache.ignite.internal.util.nio.GridNioFinishedFuture;
import org.apache.ignite.internal.util.nio.GridNioFuture;
+import org.apache.ignite.internal.util.nio.GridNioMessageWriterFactory;
import org.apache.ignite.internal.util.nio.GridNioMetricsListener;
import org.apache.ignite.internal.util.nio.GridNioServerListener;
import org.apache.ignite.internal.util.nio.GridNioSession;
import org.apache.ignite.internal.util.nio.GridNioSessionImpl;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
/**
* Allows to re-use existing {@link GridNioFilter}s on IPC (specifically shared memory IPC)
@@ -65,23 +65,23 @@ public class IpcToNioAdapter<T> {
private final GridNioMetricsListener metricsLsnr;
/** */
- private final MessageFormatter formatter;
+ private final GridNioMessageWriterFactory writerFactory;
/**
* @param metricsLsnr Metrics listener.
* @param log Log.
* @param endp Endpoint.
* @param lsnr Listener.
- * @param formatter Message formatter.
+ * @param writerFactory Writer factory.
* @param filters Filters.
*/
public IpcToNioAdapter(GridNioMetricsListener metricsLsnr, IgniteLogger log, IpcEndpoint endp,
- GridNioServerListener<T> lsnr, MessageFormatter formatter, GridNioFilter... filters) {
+ GridNioServerListener<T> lsnr, GridNioMessageWriterFactory writerFactory, GridNioFilter... filters) {
assert metricsLsnr != null;
this.metricsLsnr = metricsLsnr;
this.endp = endp;
- this.formatter = formatter;
+ this.writerFactory = writerFactory;
chain = new GridNioFilterChain<>(log, lsnr, new HeadFilter(), filters);
ses = new GridNioSessionImpl(chain, null, null, true);
@@ -163,7 +163,7 @@ public class IpcToNioAdapter<T> {
assert writeBuf.hasArray();
try {
- int cnt = U.writeMessageFully(msg, endp.outputStream(), writeBuf, formatter.writer());
+ int cnt = U.writeMessageFully(msg, endp.outputStream(), writeBuf, writerFactory.writer(ses));
metricsLsnr.onBytesSent(cnt);
}
@@ -255,4 +255,4 @@ public class IpcToNioAdapter<T> {
proceedSessionWriteTimeout(ses);
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridCommunicationClient.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridCommunicationClient.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridCommunicationClient.java
index a933916..0de54e9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridCommunicationClient.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridCommunicationClient.java
@@ -94,7 +94,7 @@ public interface GridCommunicationClient {
public void sendMessage(byte[] data, int len) throws IgniteCheckedException;
/**
- * @param nodeId Node ID (provided only if versions of local and remote nodes are different).
+ * @param nodeId Remote node ID. Provided only for sync clients.
* @param msg Message to send.
* @param closure Ack closure.
* @throws IgniteCheckedException If failed.
@@ -107,4 +107,4 @@ public interface GridCommunicationClient {
* @return {@code True} if send is asynchronous.
*/
public boolean async();
-}
\ No newline at end of file
+}
[3/3] ignite git commit: Direct marshalling optimizations
Posted by vk...@apache.org.
Direct marshalling optimizations
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2de3b19b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2de3b19b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2de3b19b
Branch: refs/heads/ignite-1.5
Commit: 2de3b19b32fea283c8cf41d59c2d85d665b0eb72
Parents: 281b277
Author: Valentin Kulichenko <va...@gmail.com>
Authored: Fri Nov 20 16:25:18 2015 -0800
Committer: Valentin Kulichenko <va...@gmail.com>
Committed: Fri Nov 20 16:25:18 2015 -0800
----------------------------------------------------------------------
.../internal/direct/DirectByteBufferStream.java | 1499 -----------------
.../internal/direct/DirectMessageReader.java | 144 +-
.../internal/direct/DirectMessageWriter.java | 65 +-
.../direct/DirectMessageWriterState.java | 123 --
.../direct/state/DirectMessageState.java | 98 ++
.../direct/state/DirectMessageStateItem.java | 28 +
.../direct/stream/DirectByteBufferStream.java | 316 ++++
.../stream/v1/DirectByteBufferStreamImplV1.java | 1360 +++++++++++++++
.../stream/v2/DirectByteBufferStreamImplV2.java | 1583 ++++++++++++++++++
.../managers/communication/GridIoManager.java | 64 +-
.../ignite/internal/util/IgniteUtils.java | 31 +
.../internal/util/ipc/IpcToNioAdapter.java | 14 +-
.../util/nio/GridCommunicationClient.java | 4 +-
.../internal/util/nio/GridDirectParser.java | 37 +-
.../util/nio/GridNioMessageReaderFactory.java | 37 +
.../util/nio/GridNioMessageWriterFactory.java | 35 +
.../ignite/internal/util/nio/GridNioServer.java | 47 +-
.../util/nio/GridShmemCommunicationClient.java | 12 +-
.../communication/MessageFormatter.java | 15 +-
.../extensions/communication/MessageReader.java | 26 +-
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 6 +-
.../communication/tcp/TcpCommunicationSpi.java | 80 +-
.../apache/ignite/stream/StreamTransformer.java | 9 +-
.../testframework/GridSpiTestContext.java | 12 +-
24 files changed, 3873 insertions(+), 1772 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectByteBufferStream.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectByteBufferStream.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectByteBufferStream.java
deleted file mode 100644
index cf56430..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectByteBufferStream.java
+++ /dev/null
@@ -1,1499 +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.internal.direct;
-
-import java.lang.reflect.Array;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.BitSet;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.UUID;
-import org.apache.ignite.internal.util.GridUnsafe;
-import org.apache.ignite.internal.util.typedef.internal.U;
-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.MessageFactory;
-import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
-import sun.misc.Unsafe;
-import sun.nio.ch.DirectBuffer;
-
-/**
- * Portable stream based on {@link ByteBuffer}.
- */
-public class DirectByteBufferStream {
- /** */
- private static final Unsafe UNSAFE = GridUnsafe.unsafe();
-
- /** */
- private static final long BYTE_ARR_OFF = UNSAFE.arrayBaseOffset(byte[].class);
-
- /** */
- private static final long SHORT_ARR_OFF = UNSAFE.arrayBaseOffset(short[].class);
-
- /** */
- private static final long INT_ARR_OFF = UNSAFE.arrayBaseOffset(int[].class);
-
- /** */
- private static final long LONG_ARR_OFF = UNSAFE.arrayBaseOffset(long[].class);
-
- /** */
- private static final long FLOAT_ARR_OFF = UNSAFE.arrayBaseOffset(float[].class);
-
- /** */
- private static final long DOUBLE_ARR_OFF = UNSAFE.arrayBaseOffset(double[].class);
-
- /** */
- private static final long CHAR_ARR_OFF = UNSAFE.arrayBaseOffset(char[].class);
-
- /** */
- private static final long BOOLEAN_ARR_OFF = UNSAFE.arrayBaseOffset(boolean[].class);
-
- /** */
- private static final byte[] BYTE_ARR_EMPTY = new byte[0];
-
- /** */
- private static final short[] SHORT_ARR_EMPTY = new short[0];
-
- /** */
- private static final int[] INT_ARR_EMPTY = U.EMPTY_INTS;
-
- /** */
- private static final long[] LONG_ARR_EMPTY = U.EMPTY_LONGS;
-
- /** */
- private static final float[] FLOAT_ARR_EMPTY = new float[0];
-
- /** */
- private static final double[] DOUBLE_ARR_EMPTY = new double[0];
-
- /** */
- private static final char[] CHAR_ARR_EMPTY = new char[0];
-
- /** */
- private static final boolean[] BOOLEAN_ARR_EMPTY = new boolean[0];
-
- /** */
- private static final ArrayCreator<byte[]> BYTE_ARR_CREATOR = new ArrayCreator<byte[]>() {
- @Override public byte[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return BYTE_ARR_EMPTY;
-
- default:
- return new byte[len];
- }
- }
- };
-
- /** */
- private static final ArrayCreator<short[]> SHORT_ARR_CREATOR = new ArrayCreator<short[]>() {
- @Override public short[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return SHORT_ARR_EMPTY;
-
- default:
- return new short[len];
- }
- }
- };
-
- /** */
- private static final ArrayCreator<int[]> INT_ARR_CREATOR = new ArrayCreator<int[]>() {
- @Override public int[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return INT_ARR_EMPTY;
-
- default:
- return new int[len];
- }
- }
- };
-
- /** */
- private static final ArrayCreator<long[]> LONG_ARR_CREATOR = new ArrayCreator<long[]>() {
- @Override public long[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return LONG_ARR_EMPTY;
-
- default:
- return new long[len];
- }
- }
- };
-
- /** */
- private static final ArrayCreator<float[]> FLOAT_ARR_CREATOR = new ArrayCreator<float[]>() {
- @Override public float[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return FLOAT_ARR_EMPTY;
-
- default:
- return new float[len];
- }
- }
- };
-
- /** */
- private static final ArrayCreator<double[]> DOUBLE_ARR_CREATOR = new ArrayCreator<double[]>() {
- @Override public double[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return DOUBLE_ARR_EMPTY;
-
- default:
- return new double[len];
- }
- }
- };
-
- /** */
- private static final ArrayCreator<char[]> CHAR_ARR_CREATOR = new ArrayCreator<char[]>() {
- @Override public char[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return CHAR_ARR_EMPTY;
-
- default:
- return new char[len];
- }
- }
- };
-
- /** */
- private static final ArrayCreator<boolean[]> BOOLEAN_ARR_CREATOR = new ArrayCreator<boolean[]>() {
- @Override public boolean[] create(int len) {
- assert len >= 0;
-
- switch (len) {
- case 0:
- return BOOLEAN_ARR_EMPTY;
-
- default:
- return new boolean[len];
- }
- }
- };
-
- /** */
- private static final Object NULL = new Object();
-
- /** */
- private final MessageFactory msgFactory;
-
- /** */
- private final MessageFormatter msgFormatter;
-
- /** */
- private ByteBuffer buf;
-
- /** */
- private byte[] heapArr;
-
- /** */
- private long baseOff;
-
- /** */
- private int arrOff = -1;
-
- /** */
- private Object tmpArr;
-
- /** */
- private int tmpArrOff;
-
- /** */
- private int tmpArrBytes;
-
- /** */
- private boolean msgTypeDone;
-
- /** */
- private Message msg;
-
- /** */
- private Iterator<?> mapIt;
-
- /** */
- private Iterator<?> it;
-
- /** */
- private Iterator<?> arrIt;
-
- /** */
- private Object arrCur = NULL;
-
- /** */
- private Object mapCur = NULL;
-
- /** */
- private Object cur = NULL;
-
- /** */
- private boolean keyDone;
-
- /** */
- private int readSize = -1;
-
- /** */
- private int readItems;
-
- /** */
- private Object[] objArr;
-
- /** */
- private Collection<Object> col;
-
- /** */
- private Map<Object, Object> map;
-
- /** */
- private boolean lastFinished;
-
- /** */
- private MessageReader reader;
-
- /**
- * @param msgFactory Message factory.
- * @param msgFormatter Message formatter.
- */
- public DirectByteBufferStream(MessageFactory msgFactory, MessageFormatter msgFormatter) {
- this.msgFactory = msgFactory;
- this.msgFormatter = msgFormatter;
- }
-
- /**
- * @param buf Buffer.
- */
- public void setBuffer(ByteBuffer buf) {
- assert buf != null;
-
- if (this.buf != buf) {
- this.buf = buf;
-
- heapArr = buf.isDirect() ? null : buf.array();
- baseOff = buf.isDirect() ? ((DirectBuffer)buf).address() : BYTE_ARR_OFF;
- }
- }
-
- /**
- * @return Number of remaining bytes.
- */
- public int remaining() {
- return buf.remaining();
- }
-
- /**
- * @return Whether last object was fully written or read.
- */
- public boolean lastFinished() {
- return lastFinished;
- }
-
- /**
- * @param val Value.
- */
- public void writeByte(byte val) {
- lastFinished = buf.remaining() >= 1;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putByte(heapArr, baseOff + pos, val);
-
- buf.position(pos + 1);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeShort(short val) {
- lastFinished = buf.remaining() >= 2;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putShort(heapArr, baseOff + pos, val);
-
- buf.position(pos + 2);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeInt(int val) {
- lastFinished = buf.remaining() >= 4;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putInt(heapArr, baseOff + pos, val);
-
- buf.position(pos + 4);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeLong(long val) {
- lastFinished = buf.remaining() >= 8;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putLong(heapArr, baseOff + pos, val);
-
- buf.position(pos + 8);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeFloat(float val) {
- lastFinished = buf.remaining() >= 4;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putFloat(heapArr, baseOff + pos, val);
-
- buf.position(pos + 4);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeDouble(double val) {
- lastFinished = buf.remaining() >= 8;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putDouble(heapArr, baseOff + pos, val);
-
- buf.position(pos + 8);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeChar(char val) {
- lastFinished = buf.remaining() >= 2;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putChar(heapArr, baseOff + pos, val);
-
- buf.position(pos + 2);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeBoolean(boolean val) {
- lastFinished = buf.remaining() >= 1;
-
- if (lastFinished) {
- int pos = buf.position();
-
- UNSAFE.putBoolean(heapArr, baseOff + pos, val);
-
- buf.position(pos + 1);
- }
- }
-
- /**
- * @param val Value.
- */
- public void writeByteArray(byte[] val) {
- if (val != null)
- lastFinished = writeArray(val, BYTE_ARR_OFF, val.length, val.length);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value.
- * @param off Offset.
- * @param len Length.
- */
- public void writeByteArray(byte[] val, long off, int len) {
- if (val != null)
- lastFinished = writeArray(val, BYTE_ARR_OFF + off, len, len);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeShortArray(short[] val) {
- if (val != null)
- lastFinished = writeArray(val, SHORT_ARR_OFF, val.length, val.length << 1);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeIntArray(int[] val) {
- if (val != null)
- lastFinished = writeArray(val, INT_ARR_OFF, val.length, val.length << 2);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeLongArray(long[] val) {
- if (val != null)
- lastFinished = writeArray(val, LONG_ARR_OFF, val.length, val.length << 3);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeFloatArray(float[] val) {
- if (val != null)
- lastFinished = writeArray(val, FLOAT_ARR_OFF, val.length, val.length << 2);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeDoubleArray(double[] val) {
- if (val != null)
- lastFinished = writeArray(val, DOUBLE_ARR_OFF, val.length, val.length << 3);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeCharArray(char[] val) {
- if (val != null)
- lastFinished = writeArray(val, CHAR_ARR_OFF, val.length, val.length << 1);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeBooleanArray(boolean[] val) {
- if (val != null)
- lastFinished = writeArray(val, BOOLEAN_ARR_OFF, val.length, val.length);
- else
- writeInt(-1);
- }
-
- /**
- * @param val Value
- */
- public void writeString(String val) {
- writeByteArray(val != null ? val.getBytes() : null);
- }
-
- /**
- * @param val Value
- */
- public void writeBitSet(BitSet val) {
- writeLongArray(val != null ? val.toLongArray() : null);
- }
-
- /**
- * @param val Value
- */
- public void writeUuid(UUID val) {
- writeByteArray(val != null ? U.uuidToBytes(val) : null);
- }
-
- /**
- * @param val Value
- */
- public void writeIgniteUuid(IgniteUuid val) {
- writeByteArray(val != null ? U.igniteUuidToBytes(val) : null);
- }
-
- /**
- * @param msg Message.
- */
- public void writeMessage(Message msg, MessageWriter writer) {
- if (msg != null) {
- if (buf.hasRemaining()) {
- try {
- writer.beforeInnerMessageWrite();
-
- lastFinished = msg.writeTo(buf, writer);
- }
- finally {
- writer.afterInnerMessageWrite(lastFinished);
- }
- }
- else
- lastFinished = false;
- }
- else
- writeByte(Byte.MIN_VALUE);
- }
-
- /**
- * @param arr Array.
- * @param itemType Component type.
- * @param writer Writer.
- */
- public <T> void writeObjectArray(T[] arr, MessageCollectionItemType itemType, MessageWriter writer) {
- if (arr != null) {
- if (arrIt == null) {
- writeInt(arr.length);
-
- if (!lastFinished)
- return;
-
- arrIt = arrayIterator(arr);
- }
-
- while (arrIt.hasNext() || arrCur != NULL) {
- if (arrCur == NULL)
- arrCur = arrIt.next();
-
- write(itemType, arrCur, writer);
-
- if (!lastFinished)
- return;
-
- arrCur = NULL;
- }
-
- arrIt = null;
- }
- else
- writeInt(-1);
- }
-
- /**
- * @param col Collection.
- * @param itemType Item type.
- * @param writer Writer.
- */
- public <T> void writeCollection(Collection<T> col, MessageCollectionItemType itemType, MessageWriter writer) {
- if (col != null) {
- if (it == null) {
- writeInt(col.size());
-
- if (!lastFinished)
- return;
-
- it = col.iterator();
- }
-
- while (it.hasNext() || cur != NULL) {
- if (cur == NULL)
- cur = it.next();
-
- write(itemType, cur, writer);
-
- if (!lastFinished)
- return;
-
- cur = NULL;
- }
-
- it = null;
- }
- else
- writeInt(-1);
- }
-
- /**
- * @param map Map.
- * @param keyType Key type.
- * @param valType Value type.
- * @param writer Writer.
- */
- @SuppressWarnings("unchecked")
- public <K, V> void writeMap(Map<K, V> map, MessageCollectionItemType keyType, MessageCollectionItemType valType,
- MessageWriter writer) {
- if (map != null) {
- if (mapIt == null) {
- writeInt(map.size());
-
- if (!lastFinished)
- return;
-
- mapIt = map.entrySet().iterator();
- }
-
- while (mapIt.hasNext() || mapCur != NULL) {
- Map.Entry<K, V> e;
-
- if (mapCur == NULL)
- mapCur = mapIt.next();
-
- e = (Map.Entry<K, V>)mapCur;
-
- if (!keyDone) {
- write(keyType, e.getKey(), writer);
-
- if (!lastFinished)
- return;
-
- keyDone = true;
- }
-
- write(valType, e.getValue(), writer);
-
- if (!lastFinished)
- return;
-
- mapCur = NULL;
- keyDone = false;
- }
-
- mapIt = null;
- }
- else
- writeInt(-1);
- }
-
- /**
- * @return Value.
- */
- public byte readByte() {
- lastFinished = buf.remaining() >= 1;
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 1);
-
- return UNSAFE.getByte(heapArr, baseOff + pos);
- }
- else
- return 0;
- }
-
- /**
- * @return Value.
- */
- public short readShort() {
- lastFinished = buf.remaining() >= 2;
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 2);
-
- return UNSAFE.getShort(heapArr, baseOff + pos);
- }
- else
- return 0;
- }
-
- /**
- * @return Value.
- */
- public int readInt() {
- lastFinished = buf.remaining() >= 4;
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 4);
-
- return UNSAFE.getInt(heapArr, baseOff + pos);
- }
- else
- return 0;
- }
-
- /**
- * @return Value.
- */
- public long readLong() {
- lastFinished = buf.remaining() >= 8;
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 8);
-
- return UNSAFE.getLong(heapArr, baseOff + pos);
- }
- else
- return 0;
- }
-
- /**
- * @return Value.
- */
- public float readFloat() {
- lastFinished = buf.remaining() >= 4;
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 4);
-
- return UNSAFE.getFloat(heapArr, baseOff + pos);
- }
- else
- return 0;
- }
-
- /**
- * @return Value.
- */
- public double readDouble() {
- lastFinished = buf.remaining() >= 8;
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 8);
-
- return UNSAFE.getDouble(heapArr, baseOff + pos);
- }
- else
- return 0;
- }
-
- /**
- * @return Value.
- */
- public char readChar() {
- lastFinished = buf.remaining() >= 2;
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 2);
-
- return UNSAFE.getChar(heapArr, baseOff + pos);
- }
- else
- return 0;
- }
-
- /**
- * @return Value.
- */
- public boolean readBoolean() {
- lastFinished = buf.hasRemaining();
-
- if (lastFinished) {
- int pos = buf.position();
-
- buf.position(pos + 1);
-
- return UNSAFE.getBoolean(heapArr, baseOff + pos);
- }
- else
- return false;
- }
-
- /**
- * @return Value.
- */
- public byte[] readByteArray() {
- return readArray(BYTE_ARR_CREATOR, 0, BYTE_ARR_OFF);
- }
-
- /**
- /**
- * @return Value.
- */
- public short[] readShortArray() {
- return readArray(SHORT_ARR_CREATOR, 1, SHORT_ARR_OFF);
- }
-
- /**
- * @return Value.
- */
- public int[] readIntArray() {
- return readArray(INT_ARR_CREATOR, 2, INT_ARR_OFF);
- }
-
- /**
- * @return Value.
- */
- public long[] readLongArray() {
- return readArray(LONG_ARR_CREATOR, 3, LONG_ARR_OFF);
- }
-
- /**
- * @return Value.
- */
- public float[] readFloatArray() {
- return readArray(FLOAT_ARR_CREATOR, 2, FLOAT_ARR_OFF);
- }
-
- /**
- * @return Value.
- */
- public double[] readDoubleArray() {
- return readArray(DOUBLE_ARR_CREATOR, 3, DOUBLE_ARR_OFF);
- }
-
- /**
- * @return Value.
- */
- public char[] readCharArray() {
- return readArray(CHAR_ARR_CREATOR, 1, CHAR_ARR_OFF);
- }
-
- /**
- * @return Value.
- */
- public boolean[] readBooleanArray() {
- return readArray(BOOLEAN_ARR_CREATOR, 0, BOOLEAN_ARR_OFF);
- }
-
- /**
- * @return Value.
- */
- public String readString() {
- byte[] arr = readByteArray();
-
- return arr != null ? new String(arr) : null;
- }
-
- /**
- * @return Value.
- */
- public BitSet readBitSet() {
- long[] arr = readLongArray();
-
- return arr != null ? BitSet.valueOf(arr) : null;
- }
-
- /**
- * @return Value.
- */
- public UUID readUuid() {
- byte[] arr = readByteArray();
-
- return arr != null ? U.bytesToUuid(arr, 0) : null;
- }
-
- /**
- * @return Value.
- */
- public IgniteUuid readIgniteUuid() {
- byte[] arr = readByteArray();
-
- return arr != null ? U.bytesToIgniteUuid(arr, 0) : null;
- }
-
- /**
- * @return Message.
- */
- @SuppressWarnings("unchecked")
- public <T extends Message> T readMessage() {
- if (!msgTypeDone) {
- if (!buf.hasRemaining()) {
- lastFinished = false;
-
- return null;
- }
-
- byte type = readByte();
-
- msg = type == Byte.MIN_VALUE ? null : msgFactory.create(type);
-
- if (msg != null)
- reader = msgFormatter.reader(msgFactory, msg.getClass());
-
- msgTypeDone = true;
- }
-
- lastFinished = msg == null || msg.readFrom(buf, reader);
-
- if (lastFinished) {
- Message msg0 = msg;
-
- msgTypeDone = false;
- msg = null;
-
- return (T)msg0;
- }
- else
- return null;
- }
-
- /**
- * @param itemType Component type.
- * @param itemCls Component class.
- * @return Array.
- */
- @SuppressWarnings("unchecked")
- public <T> T[] readObjectArray(MessageCollectionItemType itemType, Class<T> itemCls) {
- if (readSize == -1) {
- int size = readInt();
-
- if (!lastFinished)
- return null;
-
- readSize = size;
- }
-
- if (readSize >= 0) {
- if (objArr == null)
- objArr = itemCls != null ? (Object[])Array.newInstance(itemCls, readSize) : new Object[readSize];
-
- for (int i = readItems; i < readSize; i++) {
- Object item = read(itemType);
-
- if (!lastFinished)
- return null;
-
- objArr[i] = item;
-
- readItems++;
- }
- }
-
- readSize = -1;
- readItems = 0;
- cur = null;
-
- T[] objArr0 = (T[])objArr;
-
- objArr = null;
-
- return objArr0;
- }
-
- /**
- * @param itemType Item type.
- * @return Collection.
- */
- @SuppressWarnings("unchecked")
- public <C extends Collection<?>> C readCollection(MessageCollectionItemType itemType) {
- if (readSize == -1) {
- int size = readInt();
-
- if (!lastFinished)
- return null;
-
- readSize = size;
- }
-
- if (readSize >= 0) {
- if (col == null)
- col = new ArrayList<>(readSize);
-
- for (int i = readItems; i < readSize; i++) {
- Object item = read(itemType);
-
- if (!lastFinished)
- return null;
-
- col.add(item);
-
- readItems++;
- }
- }
-
- readSize = -1;
- readItems = 0;
- cur = null;
-
- C col0 = (C)col;
-
- col = null;
-
- return col0;
- }
-
- /**
- * @param keyType Key type.
- * @param valType Value type.
- * @param linked Whether linked map should be created.
- * @return Map.
- */
- @SuppressWarnings("unchecked")
- public <M extends Map<?, ?>> M readMap(MessageCollectionItemType keyType, MessageCollectionItemType valType,
- boolean linked) {
- if (readSize == -1) {
- int size = readInt();
-
- if (!lastFinished)
- return null;
-
- readSize = size;
- }
-
- if (readSize >= 0) {
- if (map == null)
- map = linked ? U.newLinkedHashMap(readSize) : U.newHashMap(readSize);
-
- for (int i = readItems; i < readSize; i++) {
- if (!keyDone) {
- Object key = read(keyType);
-
- if (!lastFinished)
- return null;
-
- mapCur = key;
- keyDone = true;
- }
-
- Object val = read(valType);
-
- if (!lastFinished)
- return null;
-
- map.put(mapCur, val);
-
- keyDone = false;
-
- readItems++;
- }
- }
-
- readSize = -1;
- readItems = 0;
- mapCur = null;
-
- M map0 = (M)map;
-
- map = null;
-
- return map0;
- }
-
- /**
- * @param arr Array.
- * @param off Offset.
- * @param len Length.
- * @param bytes Length in bytes.
- * @return Whether array was fully written
- */
- private boolean writeArray(Object arr, long off, int len, int bytes) {
- return writeArray(arr, off, len, bytes, false);
- }
-
- /**
- * @param arr Array.
- * @param off Offset.
- * @param len Length.
- * @param bytes Length in bytes.
- * @param skipLen {@code true} if length should not be written.
- * @return Whether array was fully written
- */
- private boolean writeArray(Object arr, long off, int len, int bytes, boolean skipLen) {
- assert arr != null;
- assert arr.getClass().isArray() && arr.getClass().getComponentType().isPrimitive();
- assert off > 0;
- assert len >= 0;
- assert bytes >= 0;
- assert bytes >= arrOff;
-
- if (arrOff == -1) {
- if (!skipLen) {
- if (buf.remaining() < 4)
- return false;
-
- writeInt(len);
- }
-
- arrOff = 0;
- }
-
- int toWrite = bytes - arrOff;
- int pos = buf.position();
- int remaining = buf.remaining();
-
- if (toWrite <= remaining) {
- UNSAFE.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, toWrite);
-
- pos += toWrite;
-
- buf.position(pos);
-
- arrOff = -1;
-
- return true;
- }
- else {
- UNSAFE.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, remaining);
-
- pos += remaining;
-
- buf.position(pos);
-
- arrOff += remaining;
-
- return false;
- }
- }
-
- /**
- * @param creator Array creator.
- * @param lenShift Array length shift size.
- * @param off Base offset.
- * @return Array or special value if it was not fully read.
- */
- private <T> T readArray(ArrayCreator<T> creator, int lenShift, long off) {
- return readArray(creator, lenShift, off, -1);
- }
-
- /**
- * @param creator Array creator.
- * @param lenShift Array length shift size.
- * @param off Base offset.
- * @param len Length.
- * @return Array or special value if it was not fully read.
- */
- @SuppressWarnings("unchecked")
- private <T> T readArray(ArrayCreator<T> creator, int lenShift, long off, int len) {
- assert creator != null;
-
- if (tmpArr == null) {
- if (len == -1) {
- if (buf.remaining() < 4) {
- lastFinished = false;
-
- return null;
- }
-
- len = readInt();
- }
-
- switch (len) {
- case -1:
- lastFinished = true;
-
- return null;
-
- case 0:
- lastFinished = true;
-
- return creator.create(0);
-
- default:
- tmpArr = creator.create(len);
- tmpArrBytes = len << lenShift;
- }
- }
-
- int toRead = tmpArrBytes - tmpArrOff;
- int remaining = buf.remaining();
- int pos = buf.position();
-
- lastFinished = toRead <= remaining;
-
- if (lastFinished) {
- UNSAFE.copyMemory(heapArr, baseOff + pos, tmpArr, off + tmpArrOff, toRead);
-
- buf.position(pos + toRead);
-
- T arr = (T)tmpArr;
-
- tmpArr = null;
- tmpArrBytes = 0;
- tmpArrOff = 0;
-
- return arr;
- }
- else {
- UNSAFE.copyMemory(heapArr, baseOff + pos, tmpArr, off + tmpArrOff, remaining);
-
- buf.position(pos + remaining);
-
- tmpArrOff += remaining;
-
- return null;
- }
- }
-
- /**
- * @param type Type.
- * @param val Value.
- * @param writer Writer.
- */
- private void write(MessageCollectionItemType type, Object val, MessageWriter writer) {
- switch (type) {
- case BYTE:
- writeByte((Byte)val);
-
- break;
-
- case SHORT:
- writeShort((Short)val);
-
- break;
-
- case INT:
- writeInt((Integer)val);
-
- break;
-
- case LONG:
- writeLong((Long)val);
-
- break;
-
- case FLOAT:
- writeFloat((Float)val);
-
- break;
-
- case DOUBLE:
- writeDouble((Double)val);
-
- break;
-
- case CHAR:
- writeChar((Character)val);
-
- break;
-
- case BOOLEAN:
- writeBoolean((Boolean)val);
-
- break;
-
- case BYTE_ARR:
- writeByteArray((byte[])val);
-
- break;
-
- case SHORT_ARR:
- writeShortArray((short[])val);
-
- break;
-
- case INT_ARR:
- writeIntArray((int[])val);
-
- break;
-
- case LONG_ARR:
- writeLongArray((long[])val);
-
- break;
-
- case FLOAT_ARR:
- writeFloatArray((float[])val);
-
- break;
-
- case DOUBLE_ARR:
- writeDoubleArray((double[])val);
-
- break;
-
- case CHAR_ARR:
- writeCharArray((char[])val);
-
- break;
-
- case BOOLEAN_ARR:
- writeBooleanArray((boolean[])val);
-
- break;
-
- case STRING:
- writeString((String)val);
-
- break;
-
- case BIT_SET:
- writeBitSet((BitSet)val);
-
- break;
-
- case UUID:
- writeUuid((UUID)val);
-
- break;
-
- case IGNITE_UUID:
- writeIgniteUuid((IgniteUuid)val);
-
- break;
-
- case MSG:
- try {
- if (val != null)
- writer.beforeInnerMessageWrite();
-
- writeMessage((Message)val, writer);
- }
- finally {
- if (val != null)
- writer.afterInnerMessageWrite(lastFinished);
- }
-
- break;
-
- default:
- throw new IllegalArgumentException("Unknown type: " + type);
- }
- }
-
- /**
- * @param type Type.
- * @return Value.
- */
- private Object read(MessageCollectionItemType type) {
- switch (type) {
- case BYTE:
- return readByte();
-
- case SHORT:
- return readShort();
-
- case INT:
- return readInt();
-
- case LONG:
- return readLong();
-
- case FLOAT:
- return readFloat();
-
- case DOUBLE:
- return readDouble();
-
- case CHAR:
- return readChar();
-
- case BOOLEAN:
- return readBoolean();
-
- case BYTE_ARR:
- return readByteArray();
-
- case SHORT_ARR:
- return readShortArray();
-
- case INT_ARR:
- return readIntArray();
-
- case LONG_ARR:
- return readLongArray();
-
- case FLOAT_ARR:
- return readFloatArray();
-
- case DOUBLE_ARR:
- return readDoubleArray();
-
- case CHAR_ARR:
- return readCharArray();
-
- case BOOLEAN_ARR:
- return readBooleanArray();
-
- case STRING:
- return readString();
-
- case BIT_SET:
- return readBitSet();
-
- case UUID:
- return readUuid();
-
- case IGNITE_UUID:
- return readIgniteUuid();
-
- case MSG:
- return readMessage();
-
- default:
- throw new IllegalArgumentException("Unknown type: " + type);
- }
- }
-
- /**
- * @param arr Array.
- * @return Array iterator.
- */
- private Iterator<?> arrayIterator(final Object[] arr) {
- return new Iterator<Object>() {
- private int idx;
-
- @Override public boolean hasNext() {
- return idx < arr.length;
- }
-
- @Override public Object next() {
- if (!hasNext())
- throw new NoSuchElementException();
-
- return arr[idx++];
- }
-
- @Override public void remove() {
- throw new UnsupportedOperationException();
- }
- };
- }
-
- /**
- * Array creator.
- */
- private static interface ArrayCreator<T> {
- /**
- * @param len Array length or {@code -1} if array was not fully read.
- * @return New array.
- */
- public T create(int len);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
index 7eaab76..e0b7b22 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
@@ -22,11 +22,16 @@ import java.util.BitSet;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
+import org.apache.ignite.internal.direct.state.DirectMessageState;
+import org.apache.ignite.internal.direct.state.DirectMessageStateItem;
+import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
+import org.apache.ignite.internal.direct.stream.v1.DirectByteBufferStreamImplV1;
+import org.apache.ignite.internal.direct.stream.v2.DirectByteBufferStreamImplV2;
+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.MessageFactory;
-import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.jetbrains.annotations.Nullable;
@@ -34,26 +39,32 @@ import org.jetbrains.annotations.Nullable;
* Message reader implementation.
*/
public class DirectMessageReader implements MessageReader {
- /** Stream. */
- private final DirectByteBufferStream stream;
+ /** State. */
+ private final DirectMessageState<StateItem> state;
/** Whether last field was fully read. */
private boolean lastRead;
- /** Current state. */
- private int state;
-
/**
* @param msgFactory Message factory.
- * @param msgFormatter Message formatter.
+ * @param protoVer Protocol version.
*/
- public DirectMessageReader(MessageFactory msgFactory, MessageFormatter msgFormatter) {
- this.stream = new DirectByteBufferStream(msgFactory, msgFormatter);
+ public DirectMessageReader(final MessageFactory msgFactory, final byte protoVer) {
+ state = new DirectMessageState<>(StateItem.class, new IgniteOutClosure<StateItem>() {
+ @Override public StateItem apply() {
+ return new StateItem(msgFactory, protoVer);
+ }
+ });
}
/** {@inheritDoc} */
@Override public void setBuffer(ByteBuffer buf) {
- stream.setBuffer(buf);
+ state.item().stream.setBuffer(buf);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setCurrentReadClass(Class<? extends Message> msgCls) {
+ // No-op.
}
/** {@inheritDoc} */
@@ -69,6 +80,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public byte readByte(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
byte val = stream.readByte();
lastRead = stream.lastFinished();
@@ -78,6 +91,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public short readShort(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
short val = stream.readShort();
lastRead = stream.lastFinished();
@@ -87,6 +102,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public int readInt(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
int val = stream.readInt();
lastRead = stream.lastFinished();
@@ -96,6 +113,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public long readLong(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
long val = stream.readLong();
lastRead = stream.lastFinished();
@@ -105,6 +124,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public float readFloat(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
float val = stream.readFloat();
lastRead = stream.lastFinished();
@@ -114,6 +135,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public double readDouble(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
double val = stream.readDouble();
lastRead = stream.lastFinished();
@@ -123,6 +146,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public char readChar(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
char val = stream.readChar();
lastRead = stream.lastFinished();
@@ -132,6 +157,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public boolean readBoolean(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
boolean val = stream.readBoolean();
lastRead = stream.lastFinished();
@@ -141,6 +168,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public byte[] readByteArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
byte[] arr = stream.readByteArray();
lastRead = stream.lastFinished();
@@ -150,6 +179,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public short[] readShortArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
short[] arr = stream.readShortArray();
lastRead = stream.lastFinished();
@@ -159,6 +190,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public int[] readIntArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
int[] arr = stream.readIntArray();
lastRead = stream.lastFinished();
@@ -168,6 +201,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public long[] readLongArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
long[] arr = stream.readLongArray();
lastRead = stream.lastFinished();
@@ -177,6 +212,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public float[] readFloatArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
float[] arr = stream.readFloatArray();
lastRead = stream.lastFinished();
@@ -186,6 +223,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public double[] readDoubleArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
double[] arr = stream.readDoubleArray();
lastRead = stream.lastFinished();
@@ -195,6 +234,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public char[] readCharArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
char[] arr = stream.readCharArray();
lastRead = stream.lastFinished();
@@ -204,6 +245,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public boolean[] readBooleanArray(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
boolean[] arr = stream.readBooleanArray();
lastRead = stream.lastFinished();
@@ -213,6 +256,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public String readString(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
String val = stream.readString();
lastRead = stream.lastFinished();
@@ -222,6 +267,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public BitSet readBitSet(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
BitSet val = stream.readBitSet();
lastRead = stream.lastFinished();
@@ -231,6 +278,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public UUID readUuid(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
UUID val = stream.readUuid();
lastRead = stream.lastFinished();
@@ -240,6 +289,8 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public IgniteUuid readIgniteUuid(String name) {
+ DirectByteBufferStream stream = state.item().stream;
+
IgniteUuid val = stream.readIgniteUuid();
lastRead = stream.lastFinished();
@@ -249,7 +300,9 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Nullable @Override public <T extends Message> T readMessage(String name) {
- T msg = stream.readMessage();
+ DirectByteBufferStream stream = state.item().stream;
+
+ T msg = stream.readMessage(this);
lastRead = stream.lastFinished();
@@ -258,7 +311,9 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public <T> T[] readObjectArray(String name, MessageCollectionItemType itemType, Class<T> itemCls) {
- T[] msg = stream.readObjectArray(itemType, itemCls);
+ DirectByteBufferStream stream = state.item().stream;
+
+ T[] msg = stream.readObjectArray(itemType, itemCls, this);
lastRead = stream.lastFinished();
@@ -267,7 +322,9 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public <C extends Collection<?>> C readCollection(String name, MessageCollectionItemType itemType) {
- C col = stream.readCollection(itemType);
+ DirectByteBufferStream stream = state.item().stream;
+
+ C col = stream.readCollection(itemType, this);
lastRead = stream.lastFinished();
@@ -277,7 +334,9 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public <M extends Map<?, ?>> M readMap(String name, MessageCollectionItemType keyType,
MessageCollectionItemType valType, boolean linked) {
- M map = stream.readMap(keyType, valType, linked);
+ DirectByteBufferStream stream = state.item().stream;
+
+ M map = stream.readMap(keyType, valType, linked, this);
lastRead = stream.lastFinished();
@@ -291,11 +350,62 @@ public class DirectMessageReader implements MessageReader {
/** {@inheritDoc} */
@Override public int state() {
- return state;
+ return state.item().state;
}
/** {@inheritDoc} */
@Override public void incrementState() {
- state++;
+ state.item().state++;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void beforeInnerMessageRead() {
+ state.forward();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void afterInnerMessageRead(boolean finished) {
+ state.backward(finished);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void reset() {
+ state.reset();
+ }
+
+ /**
+ */
+ private static class StateItem implements DirectMessageStateItem {
+ /** Stream. */
+ private final DirectByteBufferStream stream;
+
+ /** State. */
+ private int state;
+
+ /**
+ * @param msgFactory Message factory.
+ * @param protoVer Protocol version.
+ */
+ public StateItem(MessageFactory msgFactory, byte protoVer) {
+ switch (protoVer) {
+ case 1:
+ stream = new DirectByteBufferStreamImplV1(msgFactory);
+
+ break;
+
+ case 2:
+ stream = new DirectByteBufferStreamImplV2(msgFactory);
+
+ break;
+
+ default:
+ throw new IllegalStateException("Invalid protocol version: " + protoVer);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void reset() {
+ state = 0;
+ }
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
index ea0f37e..ad122ba 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
@@ -22,6 +22,12 @@ import java.util.BitSet;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
+import org.apache.ignite.internal.direct.state.DirectMessageState;
+import org.apache.ignite.internal.direct.state.DirectMessageStateItem;
+import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
+import org.apache.ignite.internal.direct.stream.v1.DirectByteBufferStreamImplV1;
+import org.apache.ignite.internal.direct.stream.v2.DirectByteBufferStreamImplV2;
+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;
@@ -33,10 +39,35 @@ import org.jetbrains.annotations.Nullable;
*/
public class DirectMessageWriter implements MessageWriter {
/** Stream. */
- private final DirectByteBufferStream stream = new DirectByteBufferStream(null, null);
+ private final DirectByteBufferStream stream;
/** State. */
- private final DirectMessageWriterState state = new DirectMessageWriterState();
+ private final DirectMessageState<StateItem> state = new DirectMessageState<>(StateItem.class,
+ new IgniteOutClosure<StateItem>() {
+ @Override public StateItem apply() {
+ return new StateItem();
+ }
+ });
+
+ /**
+ * @param protoVer Protocol version.
+ */
+ public DirectMessageWriter(byte protoVer) {
+ switch (protoVer) {
+ case 1:
+ stream = new DirectByteBufferStreamImplV1(null);
+
+ break;
+
+ case 2:
+ stream = new DirectByteBufferStreamImplV2(null);
+
+ break;
+
+ default:
+ throw new IllegalStateException("Invalid protocol version: " + protoVer);
+ }
+ }
/** {@inheritDoc} */
@Override public void setBuffer(ByteBuffer buf) {
@@ -228,36 +259,52 @@ public class DirectMessageWriter implements MessageWriter {
/** {@inheritDoc} */
@Override public boolean isHeaderWritten() {
- return state.isTypeWritten();
+ return state.item().hdrWritten;
}
/** {@inheritDoc} */
@Override public void onHeaderWritten() {
- state.onTypeWritten();
+ state.item().hdrWritten = true;
}
/** {@inheritDoc} */
@Override public int state() {
- return state.state();
+ return state.item().state;
}
/** {@inheritDoc} */
@Override public void incrementState() {
- state.incrementState();
+ state.item().state++;
}
/** {@inheritDoc} */
@Override public void beforeInnerMessageWrite() {
- state.beforeInnerMessageWrite();
+ state.forward();
}
/** {@inheritDoc} */
@Override public void afterInnerMessageWrite(boolean finished) {
- state.afterInnerMessageWrite(finished);
+ state.backward(finished);
}
/** {@inheritDoc} */
@Override public void reset() {
state.reset();
}
-}
\ No newline at end of file
+
+ /**
+ */
+ private static class StateItem implements DirectMessageStateItem {
+ /** */
+ private int state;
+
+ /** */
+ private boolean hdrWritten;
+
+ /** {@inheritDoc} */
+ @Override public void reset() {
+ state = 0;
+ hdrWritten = false;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriterState.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriterState.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriterState.java
deleted file mode 100644
index 7b4cd9e..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriterState.java
+++ /dev/null
@@ -1,123 +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.internal.direct;
-
-import java.util.Arrays;
-
-/**
- * Writer state.
- */
-public class DirectMessageWriterState {
- /** Initial array size. */
- private static final int INIT_SIZE = 10;
-
- /** Initial value. */
- private static final int INIT_VAL = -1;
-
- /** Stack array. */
- private int[] stack;
-
- /** Current position. */
- private int pos;
-
- /**
- *
- */
- public DirectMessageWriterState() {
- stack = new int[INIT_SIZE];
-
- Arrays.fill(stack, INIT_VAL);
- }
-
- /**
- * @return Whether type is written.
- */
- public boolean isTypeWritten() {
- return stack[pos] >= 0;
- }
-
- /**
- * Callback called after type is written.
- */
- public void onTypeWritten() {
- assert stack[pos] == -1;
-
- stack[pos] = 0;
- }
-
- /**
- * @return Current state.
- */
- public int state() {
- return stack[pos];
- }
-
- /**
- * Increments state.
- */
- public void incrementState() {
- stack[pos]++;
- }
-
- /**
- * @param val New state value.
- */
- protected void setState(int val) {
- stack[pos] = val;
- }
-
- /**
- * Callback called before inner message is written.
- */
- public void beforeInnerMessageWrite() {
- pos++;
-
- // Growing never happen for Ignite messages, but we need
- // to support it for custom messages from plugins.
- if (pos == stack.length) {
- int[] stack0 = stack;
-
- stack = new int[stack.length << 1];
-
- System.arraycopy(stack0, 0, stack, 0, stack0.length);
-
- Arrays.fill(stack, stack0.length, stack.length, INIT_VAL);
- }
- }
-
- /**
- * Callback called after inner message is written.
- *
- * @param finished Whether message was fully written.
- */
- public void afterInnerMessageWrite(boolean finished) {
- if (finished)
- stack[pos] = INIT_VAL;
-
- pos--;
- }
-
- /**
- * Resets state.
- */
- public void reset() {
- assert pos == 0;
-
- stack[0] = INIT_VAL;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java
new file mode 100644
index 0000000..a61bb30
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java
@@ -0,0 +1,98 @@
+/*
+ * 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.direct.state;
+
+import java.lang.reflect.Array;
+import org.apache.ignite.lang.IgniteOutClosure;
+
+/**
+ * Message state.
+ */
+public class DirectMessageState<T extends DirectMessageStateItem> {
+ /** Initial array size. */
+ private static final int INIT_SIZE = 10;
+
+ /** Item factory. */
+ private final IgniteOutClosure<T> factory;
+
+ /** Stack array. */
+ private T[] stack;
+
+ /** Current position. */
+ private int pos;
+
+ /**
+ * @param cls State item type.
+ * @param factory Item factory.
+ */
+ @SuppressWarnings("unchecked")
+ public DirectMessageState(Class<T> cls, IgniteOutClosure<T> factory) {
+ this.factory = factory;
+
+ stack = (T[])Array.newInstance(cls, INIT_SIZE);
+
+ stack[0] = factory.apply();
+ }
+
+ /**
+ * @return Current item.
+ */
+ public T item() {
+ return stack[pos];
+ }
+
+ /**
+ * Go forward.
+ */
+ @SuppressWarnings("unchecked")
+ public void forward() {
+ pos++;
+
+ if (pos == stack.length) {
+ T[] stack0 = stack;
+
+ stack = (T[])Array.newInstance(stack.getClass().getComponentType(), stack.length << 1);
+
+ System.arraycopy(stack0, 0, stack, 0, stack0.length);
+ }
+
+ if (stack[pos] == null)
+ stack[pos] = factory.apply();
+ }
+
+ /**
+ * Go backward.
+ *
+ * @param reset Whether to reset current item.
+ */
+ public void backward(boolean reset) {
+ if (reset)
+ stack[pos].reset();
+
+ pos--;
+ }
+
+ /**
+ * Resets state.
+ */
+ public void reset() {
+ assert pos == 0;
+
+ stack[0].reset();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageStateItem.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageStateItem.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageStateItem.java
new file mode 100644
index 0000000..e1aa8d3
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageStateItem.java
@@ -0,0 +1,28 @@
+/*
+ * 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.direct.state;
+
+/**
+ * Message state item.
+ */
+public interface DirectMessageStateItem {
+ /**
+ * Resets state.
+ */
+ public void reset();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2de3b19b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java
new file mode 100644
index 0000000..bc9de5a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/DirectByteBufferStream.java
@@ -0,0 +1,316 @@
+/*
+ * 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.direct.stream;
+
+import java.nio.ByteBuffer;
+import java.util.BitSet;
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+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.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * Direct marshalling I/O stream.
+ */
+public interface DirectByteBufferStream {
+ /**
+ * @param buf Buffer.
+ */
+ public void setBuffer(ByteBuffer buf);
+
+ /**
+ * @return Number of remaining bytes.
+ */
+ public int remaining();
+
+ /**
+ * @return Whether last object was fully written or read.
+ */
+ public boolean lastFinished();
+
+ /**
+ * @param val Value.
+ */
+ public void writeByte(byte val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeShort(short val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeInt(int val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeLong(long val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeFloat(float val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeDouble(double val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeChar(char val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeBoolean(boolean val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeByteArray(byte[] val);
+
+ /**
+ * @param val Value.
+ * @param off Offset.
+ * @param len Length.
+ */
+ public void writeByteArray(byte[] val, long off, int len);
+
+ /**
+ * @param val Value.
+ */
+ public void writeShortArray(short[] val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeIntArray(int[] val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeLongArray(long[] val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeFloatArray(float[] val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeDoubleArray(double[] val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeCharArray(char[] val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeBooleanArray(boolean[] val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeString(String val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeBitSet(BitSet val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeUuid(UUID val);
+
+ /**
+ * @param val Value.
+ */
+ public void writeIgniteUuid(IgniteUuid val);
+
+ /**
+ * @param msg Message.
+ * @param writer Writer.
+ */
+ public void writeMessage(Message msg, MessageWriter writer);
+
+ /**
+ * @param arr Array.
+ * @param itemType Component type.
+ * @param writer Writer.
+ */
+ public <T> void writeObjectArray(T[] arr, MessageCollectionItemType itemType, MessageWriter writer);
+
+ /**
+ * @param col Collection.
+ * @param itemType Component type.
+ * @param writer Writer.
+ */
+ public <T> void writeCollection(Collection<T> col, MessageCollectionItemType itemType, MessageWriter writer);
+
+ /**
+ * @param map Map.
+ * @param keyType Key type.
+ * @param valType Value type.
+ * @param writer Writer.
+ */
+ public <K, V> void writeMap(Map<K, V> map, MessageCollectionItemType keyType, MessageCollectionItemType valType,
+ MessageWriter writer);
+
+ /**
+ * @return Value.
+ */
+ public byte readByte();
+
+ /**
+ * @return Value.
+ */
+ public short readShort();
+
+ /**
+ * @return Value.
+ */
+ public int readInt();
+
+ /**
+ * @return Value.
+ */
+ public long readLong();
+
+ /**
+ * @return Value.
+ */
+ public float readFloat();
+
+ /**
+ * @return Value.
+ */
+ public double readDouble();
+
+ /**
+ * @return Value.
+ */
+ public char readChar();
+
+ /**
+ * @return Value.
+ */
+ public boolean readBoolean();
+
+ /**
+ * @return Value.
+ */
+ public byte[] readByteArray();
+
+ /**
+ * @return Value.
+ */
+ public short[] readShortArray();
+
+ /**
+ * @return Value.
+ */
+ public int[] readIntArray();
+
+ /**
+ * @return Value.
+ */
+ public long[] readLongArray();
+
+ /**
+ * @return Value.
+ */
+ public float[] readFloatArray();
+
+ /**
+ * @return Value.
+ */
+ public double[] readDoubleArray();
+
+ /**
+ * @return Value.
+ */
+ public char[] readCharArray();
+
+ /**
+ * @return Value.
+ */
+ public boolean[] readBooleanArray();
+
+ /**
+ * @return Value.
+ */
+ public String readString();
+
+ /**
+ * @return Value.
+ */
+ public BitSet readBitSet();
+
+ /**
+ * @return Value.
+ */
+ public UUID readUuid();
+
+ /**
+ * @return Value.
+ */
+ public IgniteUuid readIgniteUuid();
+
+ /**
+ * @param reader Reader.
+ * @return Message.
+ */
+ public <T extends Message> T readMessage(MessageReader reader);
+
+ /**
+ * @param itemType Item type.
+ * @param itemCls Item class.
+ * @param reader Reader.
+ * @return Array.
+ */
+ public <T> T[] readObjectArray(MessageCollectionItemType itemType, Class<T> itemCls, MessageReader reader);
+
+ /**
+ * @param itemType Item type.
+ * @param reader Reader.
+ * @return Collection.
+ */
+ public <C extends Collection<?>> C readCollection(MessageCollectionItemType itemType, MessageReader reader);
+
+ /**
+ * @param keyType Key type.
+ * @param valType Value type.
+ * @param linked Whether linked map should be created.
+ * @param reader Reader.
+ * @return Map.
+ */
+ public <M extends Map<?, ?>> M readMap(MessageCollectionItemType keyType, MessageCollectionItemType valType,
+ boolean linked, MessageReader reader);
+}