| /* |
| * 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.network.internal.direct.stream; |
| |
| 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.util.ArrayFactory; |
| import org.apache.ignite.internal.util.GridUnsafe; |
| import org.apache.ignite.internal.util.IgniteUtils; |
| import org.apache.ignite.lang.IgniteUuid; |
| import org.apache.ignite.network.NetworkMessage; |
| import org.apache.ignite.network.serialization.MessageDeserializer; |
| import org.apache.ignite.network.serialization.MessageReader; |
| import org.apache.ignite.network.serialization.MessageSerializationRegistry; |
| import org.apache.ignite.network.serialization.MessageSerializer; |
| import org.apache.ignite.network.serialization.MessageWriter; |
| import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType; |
| |
| import static org.apache.ignite.internal.util.ArrayUtils.BOOLEAN_ARRAY; |
| import static org.apache.ignite.internal.util.ArrayUtils.BYTE_ARRAY; |
| import static org.apache.ignite.internal.util.ArrayUtils.CHAR_ARRAY; |
| import static org.apache.ignite.internal.util.ArrayUtils.DOUBLE_ARRAY; |
| import static org.apache.ignite.internal.util.ArrayUtils.FLOAT_ARRAY; |
| import static org.apache.ignite.internal.util.ArrayUtils.INT_ARRAY; |
| import static org.apache.ignite.internal.util.ArrayUtils.LONG_ARRAY; |
| import static org.apache.ignite.internal.util.ArrayUtils.SHORT_ARRAY; |
| import static org.apache.ignite.internal.util.GridUnsafe.BIG_ENDIAN; |
| import static org.apache.ignite.internal.util.GridUnsafe.BYTE_ARR_OFF; |
| import static org.apache.ignite.internal.util.GridUnsafe.CHAR_ARR_OFF; |
| import static org.apache.ignite.internal.util.GridUnsafe.DOUBLE_ARR_OFF; |
| import static org.apache.ignite.internal.util.GridUnsafe.FLOAT_ARR_OFF; |
| import static org.apache.ignite.internal.util.GridUnsafe.INT_ARR_OFF; |
| import static org.apache.ignite.internal.util.GridUnsafe.LONG_ARR_OFF; |
| import static org.apache.ignite.internal.util.GridUnsafe.SHORT_ARR_OFF; |
| |
| public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream { |
| /** Poison object. */ |
| private static final Object NULL = new Object(); |
| |
| /** */ |
| private final MessageSerializationRegistry serializationRegistry; |
| |
| /** */ |
| private ByteBuffer buf; |
| |
| /** */ |
| private byte[] heapArr; |
| |
| /** */ |
| private long baseOff; |
| |
| /** */ |
| private int arrOff = -1; |
| |
| /** */ |
| private Object tmpArr; |
| |
| /** */ |
| private int tmpArrOff; |
| |
| /** Number of bytes of the boundary value, read from previous message. */ |
| private int valReadBytes; |
| |
| /** */ |
| private int tmpArrBytes; |
| |
| /** */ |
| private boolean msgTypeDone; |
| |
| /** */ |
| private MessageDeserializer<NetworkMessage> msgDeserializer; |
| |
| /** */ |
| 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; |
| |
| /** */ |
| protected boolean lastFinished; |
| |
| /** byte-array representation of string */ |
| private byte[] curStrBackingArr; |
| |
| /** |
| * @param serializationRegistry Message mappers. |
| */ |
| public DirectByteBufferStreamImplV1(MessageSerializationRegistry serializationRegistry) { |
| this.serializationRegistry = serializationRegistry; |
| } |
| |
| /** {@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() ? GridUnsafe.bufferAddress(buf) : 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(); |
| |
| GridUnsafe.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(); |
| |
| long off = baseOff + pos; |
| |
| if (BIG_ENDIAN) |
| GridUnsafe.putShortLE(heapArr, off, val); |
| else |
| GridUnsafe.putShort(heapArr, off, 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); |
| |
| GridUnsafe.putByte(heapArr, baseOff + pos++, b); |
| |
| val >>>= 7; |
| } |
| |
| GridUnsafe.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); |
| |
| GridUnsafe.putByte(heapArr, baseOff + pos++, b); |
| |
| val >>>= 7; |
| } |
| |
| GridUnsafe.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(); |
| |
| long off = baseOff + pos; |
| |
| if (BIG_ENDIAN) |
| GridUnsafe.putFloatLE(heapArr, off, val); |
| else |
| GridUnsafe.putFloat(heapArr, off, val); |
| |
| buf.position(pos + 4); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeDouble(double val) { |
| lastFinished = buf.remaining() >= 8; |
| |
| if (lastFinished) { |
| int pos = buf.position(); |
| |
| long off = baseOff + pos; |
| |
| if (BIG_ENDIAN) |
| GridUnsafe.putDoubleLE(heapArr, off, val); |
| else |
| GridUnsafe.putDouble(heapArr, off, val); |
| |
| buf.position(pos + 8); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeChar(char val) { |
| lastFinished = buf.remaining() >= 2; |
| |
| if (lastFinished) { |
| int pos = buf.position(); |
| |
| long off = baseOff + pos; |
| |
| if (BIG_ENDIAN) |
| GridUnsafe.putCharLE(heapArr, off, val); |
| else |
| GridUnsafe.putChar(heapArr, off, val); |
| |
| buf.position(pos + 2); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeBoolean(boolean val) { |
| lastFinished = buf.remaining() >= 1; |
| |
| if (lastFinished) { |
| int pos = buf.position(); |
| |
| GridUnsafe.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) |
| if (BIG_ENDIAN) |
| lastFinished = writeArrayLE(val, SHORT_ARR_OFF, val.length, 2, 1); |
| else |
| lastFinished = writeArray(val, SHORT_ARR_OFF, val.length, val.length << 1); |
| else |
| writeInt(-1); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeIntArray(int[] val) { |
| if (val != null) |
| if (BIG_ENDIAN) |
| lastFinished = writeArrayLE(val, INT_ARR_OFF, val.length, 4, 2); |
| else |
| lastFinished = writeArray(val, INT_ARR_OFF, val.length, val.length << 2); |
| else |
| writeInt(-1); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeLongArray(long[] val) { |
| if (val != null) |
| if (BIG_ENDIAN) |
| lastFinished = writeArrayLE(val, LONG_ARR_OFF, val.length, 8, 3); |
| else |
| lastFinished = writeArray(val, LONG_ARR_OFF, val.length, val.length << 3); |
| else |
| writeInt(-1); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeLongArray(long[] val, int len) { |
| if (val != null) |
| if (BIG_ENDIAN) |
| lastFinished = writeArrayLE(val, LONG_ARR_OFF, len, 8, 3); |
| else |
| lastFinished = writeArray(val, LONG_ARR_OFF, len, len << 3); |
| else |
| writeInt(-1); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeFloatArray(float[] val) { |
| if (val != null) |
| if (BIG_ENDIAN) |
| lastFinished = writeArrayLE(val, FLOAT_ARR_OFF, val.length, 4, 2); |
| else |
| lastFinished = writeArray(val, FLOAT_ARR_OFF, val.length, val.length << 2); |
| else |
| writeInt(-1); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeDoubleArray(double[] val) { |
| if (val != null) |
| if (BIG_ENDIAN) |
| lastFinished = writeArrayLE(val, DOUBLE_ARR_OFF, val.length, 8, 3); |
| else |
| lastFinished = writeArray(val, DOUBLE_ARR_OFF, val.length, val.length << 3); |
| else |
| writeInt(-1); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeCharArray(char[] val) { |
| if (val != null) { |
| if (BIG_ENDIAN) |
| lastFinished = writeArrayLE(val, CHAR_ARR_OFF, val.length, 2, 1); |
| else |
| 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, GridUnsafe.BOOLEAN_ARR_OFF, val.length, val.length); |
| else |
| writeInt(-1); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeString(String val) { |
| if (val != null) { |
| if (curStrBackingArr == null) |
| curStrBackingArr = val.getBytes(); |
| |
| writeByteArray(curStrBackingArr); |
| |
| if (lastFinished) |
| curStrBackingArr = null; |
| } |
| else |
| writeByteArray(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++; |
| |
| //noinspection fallthrough |
| case 1: |
| writeLong(val.getMostSignificantBits()); |
| |
| if (!lastFinished) |
| return; |
| |
| uuidState++; |
| |
| //noinspection fallthrough |
| 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++; |
| |
| //noinspection fallthrough |
| case 1: |
| writeLong(val.globalId().getMostSignificantBits()); |
| |
| if (!lastFinished) |
| return; |
| |
| uuidState++; |
| |
| //noinspection fallthrough |
| case 2: |
| writeLong(val.globalId().getLeastSignificantBits()); |
| |
| if (!lastFinished) |
| return; |
| |
| uuidState++; |
| |
| //noinspection fallthrough |
| case 3: |
| writeLong(val.localId()); |
| |
| if (!lastFinished) |
| return; |
| |
| uuidState = 0; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeMessage(NetworkMessage msg, MessageWriter writer) { |
| if (msg != null) { |
| if (buf.hasRemaining()) { |
| try { |
| writer.beforeInnerMessageWrite(); |
| |
| writer.setCurrentWriteClass(msg.getClass()); |
| |
| MessageSerializer<NetworkMessage> serializer = serializationRegistry.createSerializer(msg.directType()); |
| |
| writer.setBuffer(buf); |
| |
| lastFinished = serializer.writeMessage(msg, writer); |
| } |
| finally { |
| writer.afterInnerMessageWrite(lastFinished); |
| } |
| } |
| else |
| lastFinished = false; |
| } |
| else |
| writeShort(Short.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} */ |
| @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) { |
| if (mapCur == NULL) |
| mapCur = mapIt.next(); |
| |
| Map.Entry<K, V> 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 GridUnsafe.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); |
| |
| long off = baseOff + pos; |
| |
| return BIG_ENDIAN ? GridUnsafe.getShortLE(heapArr, off) : GridUnsafe.getShort(heapArr, off); |
| } |
| else |
| return 0; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int readInt() { |
| lastFinished = false; |
| |
| int val = 0; |
| |
| while (buf.hasRemaining()) { |
| int pos = buf.position(); |
| |
| byte b = GridUnsafe.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 = GridUnsafe.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); |
| |
| long off = baseOff + pos; |
| |
| return BIG_ENDIAN ? GridUnsafe.getFloatLE(heapArr, off) : GridUnsafe.getFloat(heapArr, off); |
| } |
| else |
| return 0; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public double readDouble() { |
| lastFinished = buf.remaining() >= 8; |
| |
| if (lastFinished) { |
| int pos = buf.position(); |
| |
| buf.position(pos + 8); |
| |
| long off = baseOff + pos; |
| |
| return BIG_ENDIAN ? GridUnsafe.getDoubleLE(heapArr, off) : GridUnsafe.getDouble(heapArr, off); |
| } |
| else |
| return 0; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public char readChar() { |
| lastFinished = buf.remaining() >= 2; |
| |
| if (lastFinished) { |
| int pos = buf.position(); |
| |
| buf.position(pos + 2); |
| |
| long off = baseOff + pos; |
| |
| return BIG_ENDIAN ? GridUnsafe.getCharLE(heapArr, off) : GridUnsafe.getChar(heapArr, off); |
| } |
| else |
| return 0; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean readBoolean() { |
| lastFinished = buf.hasRemaining(); |
| |
| if (lastFinished) { |
| int pos = buf.position(); |
| |
| buf.position(pos + 1); |
| |
| return GridUnsafe.getBoolean(heapArr, baseOff + pos); |
| } |
| else |
| return false; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public byte[] readByteArray() { |
| return readArray(BYTE_ARRAY, 0, BYTE_ARR_OFF); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public short[] readShortArray() { |
| if (BIG_ENDIAN) |
| return readArrayLE(SHORT_ARRAY, 2, 1, SHORT_ARR_OFF); |
| else |
| return readArray(SHORT_ARRAY, 1, SHORT_ARR_OFF); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int[] readIntArray() { |
| if (BIG_ENDIAN) |
| return readArrayLE(INT_ARRAY, 4, 2, INT_ARR_OFF); |
| else |
| return readArray(INT_ARRAY, 2, INT_ARR_OFF); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public long[] readLongArray() { |
| if (BIG_ENDIAN) |
| return readArrayLE(LONG_ARRAY, 8, 3, LONG_ARR_OFF); |
| else |
| return readArray(LONG_ARRAY, 3, LONG_ARR_OFF); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public float[] readFloatArray() { |
| if (BIG_ENDIAN) |
| return readArrayLE(FLOAT_ARRAY, 4, 2, FLOAT_ARR_OFF); |
| else |
| return readArray(FLOAT_ARRAY, 2, FLOAT_ARR_OFF); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public double[] readDoubleArray() { |
| if (BIG_ENDIAN) |
| return readArrayLE(DOUBLE_ARRAY, 8, 3, DOUBLE_ARR_OFF); |
| else |
| return readArray(DOUBLE_ARRAY, 3, DOUBLE_ARR_OFF); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public char[] readCharArray() { |
| if (BIG_ENDIAN) |
| return readArrayLE(CHAR_ARRAY, 2, 1, CHAR_ARR_OFF); |
| else |
| return readArray(CHAR_ARRAY, 1, CHAR_ARR_OFF); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean[] readBooleanArray() { |
| return readArray(BOOLEAN_ARRAY, 0, GridUnsafe.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++; |
| |
| //noinspection fallthrough |
| case 1: |
| uuidMost = readLong(); |
| |
| if (!lastFinished) |
| return null; |
| |
| uuidState++; |
| |
| //noinspection fallthrough |
| 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++; |
| |
| //noinspection fallthrough |
| case 1: |
| uuidMost = readLong(); |
| |
| if (!lastFinished) |
| return null; |
| |
| uuidState++; |
| |
| //noinspection fallthrough |
| case 2: |
| uuidLeast = readLong(); |
| |
| if (!lastFinished) |
| return null; |
| |
| uuidState++; |
| |
| //noinspection fallthrough |
| 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} */ |
| @Override public <T extends NetworkMessage> T readMessage(MessageReader reader) { |
| if (!msgTypeDone) { |
| if (buf.remaining() < NetworkMessage.DIRECT_TYPE_SIZE) { |
| lastFinished = false; |
| |
| return null; |
| } |
| |
| short type = readShort(); |
| |
| msgDeserializer = type == Short.MIN_VALUE ? null : serializationRegistry.createDeserializer(type); |
| |
| msgTypeDone = true; |
| } |
| |
| if (msgDeserializer != null) { |
| try { |
| reader.beforeInnerMessageRead(); |
| |
| reader.setCurrentReadClass(msgDeserializer.klass()); |
| |
| reader.setBuffer(buf); |
| lastFinished = msgDeserializer.readMessage(reader); |
| } |
| finally { |
| reader.afterInnerMessageRead(lastFinished); |
| } |
| } |
| else |
| lastFinished = true; |
| |
| if (lastFinished) { |
| NetworkMessage msg0 = msgDeserializer != null ? msgDeserializer.getMessage() : null; |
| |
| msgTypeDone = false; |
| msgDeserializer = null; |
| |
| return (T)msg0; |
| } |
| else |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @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} */ |
| @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} */ |
| @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 ? IgniteUtils.newLinkedHashMap(readSize) : IgniteUtils.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. |
| */ |
| 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 (writeArrayLength(len)) |
| return false; |
| |
| int toWrite = bytes - arrOff; |
| int pos = buf.position(); |
| int remaining = buf.remaining(); |
| |
| if (toWrite <= remaining) { |
| if (toWrite > 0) { |
| GridUnsafe.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, toWrite); |
| |
| buf.position(pos + toWrite); |
| } |
| |
| arrOff = -1; |
| |
| return true; |
| } |
| else { |
| if (remaining > 0) { |
| GridUnsafe.copyMemory(arr, off + arrOff, heapArr, baseOff + pos, remaining); |
| |
| buf.position(pos + remaining); |
| |
| arrOff += remaining; |
| } |
| |
| return false; |
| } |
| } |
| |
| /** |
| * @param arr Array. |
| * @param off Offset. |
| * @param len Length. |
| * @param typeSize Primitive type size in bytes. Needs for byte reverse. |
| * @param shiftCnt Shift for length. |
| * @return Whether array was fully written. |
| */ |
| boolean writeArrayLE(Object arr, long off, int len, int typeSize, int shiftCnt) { |
| assert arr != null; |
| assert arr.getClass().isArray() && arr.getClass().getComponentType().isPrimitive(); |
| assert off > 0; |
| assert len >= 0; |
| |
| int bytes = len << shiftCnt; |
| |
| assert bytes >= arrOff; |
| |
| if (writeArrayLength(len)) |
| return false; |
| |
| int toWrite = (bytes - arrOff) >> shiftCnt; |
| int remaining = buf.remaining() >> shiftCnt; |
| |
| if (toWrite <= remaining) { |
| writeArrayLE(arr, off, toWrite, typeSize); |
| |
| arrOff = -1; |
| |
| return true; |
| } |
| else { |
| if (remaining > 0) |
| writeArrayLE(arr, off, remaining, typeSize); |
| |
| return false; |
| } |
| } |
| |
| /** |
| * @param arr Array. |
| * @param off Offset. |
| * @param len Length. |
| * @param typeSize Primitive type size in bytes. |
| */ |
| private void writeArrayLE(Object arr, long off, int len, int typeSize) { |
| int pos = buf.position(); |
| |
| for (int i = 0; i < len; i++) { |
| for (int j = 0; j < typeSize; j++) { |
| byte b = GridUnsafe.getByteField(arr, off + arrOff + (typeSize - j - 1)); |
| |
| GridUnsafe.putByte(heapArr, baseOff + pos++, b); |
| } |
| |
| buf.position(pos); |
| arrOff += typeSize; |
| } |
| } |
| |
| /** |
| * @param len Length. |
| */ |
| private boolean writeArrayLength(int len) { |
| if (arrOff == -1) { |
| writeInt(len); |
| |
| if (!lastFinished) |
| return true; |
| |
| arrOff = 0; |
| } |
| return false; |
| } |
| |
| /** |
| * @param <T> Type of an array. |
| * @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. |
| */ |
| <T> T readArray(ArrayFactory<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.of(0); |
| |
| default: |
| tmpArr = creator.of(len); |
| tmpArrBytes = len << lenShift; |
| } |
| } |
| |
| int toRead = tmpArrBytes - tmpArrOff; |
| int remaining = buf.remaining(); |
| int pos = buf.position(); |
| |
| lastFinished = toRead <= remaining; |
| |
| if (lastFinished) { |
| GridUnsafe.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 { |
| GridUnsafe.copyMemory(heapArr, baseOff + pos, tmpArr, off + tmpArrOff, remaining); |
| |
| buf.position(pos + remaining); |
| |
| tmpArrOff += remaining; |
| |
| return null; |
| } |
| } |
| |
| /** |
| * @param <T> Type of an array. |
| * @param creator Array creator. |
| * @param typeSize Primitive type size in bytes. |
| * @param lenShift Array length shift size. |
| * @param off Base offset. |
| * @return Array or special value if it was not fully read. |
| */ |
| <T> T readArrayLE(ArrayFactory<T> creator, int typeSize, 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.of(0); |
| |
| default: |
| tmpArr = creator.of(len); |
| tmpArrBytes = len << lenShift; |
| } |
| } |
| |
| int toRead = tmpArrBytes - tmpArrOff - valReadBytes; |
| int remaining = buf.remaining(); |
| |
| lastFinished = toRead <= remaining; |
| |
| if (!lastFinished) |
| toRead = remaining; |
| |
| int pos = buf.position(); |
| |
| for (int i = 0; i < toRead; i++) { |
| byte b = GridUnsafe.getByte(heapArr, baseOff + pos + i); |
| |
| GridUnsafe.putByteField(tmpArr, off + tmpArrOff + (typeSize - valReadBytes - 1), b); |
| |
| if (++valReadBytes == typeSize) { |
| valReadBytes = 0; |
| tmpArrOff += typeSize; |
| } |
| } |
| |
| buf.position(pos + toRead); |
| |
| if (lastFinished) { |
| T arr = (T)tmpArr; |
| |
| tmpArr = null; |
| tmpArrBytes = 0; |
| tmpArrOff = 0; |
| |
| return arr; |
| } |
| else |
| return null; |
| } |
| |
| protected 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((NetworkMessage) 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. |
| */ |
| protected 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); |
| } |
| } |
| } |
