| /* |
| * 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.client.thin; |
| |
| import java.io.IOException; |
| import java.lang.reflect.Array; |
| import java.util.AbstractMap.SimpleEntry; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.LinkedHashMap; |
| import java.util.LinkedHashSet; |
| import java.util.LinkedList; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.function.BiConsumer; |
| import java.util.function.Consumer; |
| import java.util.function.Function; |
| import java.util.stream.Collectors; |
| import javax.cache.expiry.ExpiryPolicy; |
| import org.apache.ignite.binary.BinaryObject; |
| import org.apache.ignite.binary.BinaryRawWriter; |
| import org.apache.ignite.cache.CacheAtomicityMode; |
| import org.apache.ignite.cache.CacheKeyConfiguration; |
| import org.apache.ignite.cache.CacheMode; |
| import org.apache.ignite.cache.CacheRebalanceMode; |
| import org.apache.ignite.cache.CacheWriteSynchronizationMode; |
| import org.apache.ignite.cache.PartitionLossPolicy; |
| import org.apache.ignite.cache.QueryEntity; |
| import org.apache.ignite.cache.QueryIndex; |
| import org.apache.ignite.cache.QueryIndexType; |
| import org.apache.ignite.cache.query.SqlFieldsQuery; |
| import org.apache.ignite.client.ClientCacheConfiguration; |
| import org.apache.ignite.internal.binary.BinaryContext; |
| import org.apache.ignite.internal.binary.BinaryFieldMetadata; |
| import org.apache.ignite.internal.binary.BinaryMetadata; |
| import org.apache.ignite.internal.binary.BinaryObjectImpl; |
| import org.apache.ignite.internal.binary.BinaryRawWriterEx; |
| import org.apache.ignite.internal.binary.BinaryReaderExImpl; |
| import org.apache.ignite.internal.binary.BinaryReaderHandles; |
| import org.apache.ignite.internal.binary.BinarySchema; |
| import org.apache.ignite.internal.binary.BinaryThreadLocalContext; |
| import org.apache.ignite.internal.binary.BinaryUtils; |
| import org.apache.ignite.internal.binary.BinaryWriterExImpl; |
| import org.apache.ignite.internal.binary.streams.BinaryHeapInputStream; |
| import org.apache.ignite.internal.binary.streams.BinaryInputStream; |
| import org.apache.ignite.internal.binary.streams.BinaryOutputStream; |
| import org.apache.ignite.internal.processors.platform.cache.expiry.PlatformExpiryPolicy; |
| import org.apache.ignite.internal.util.MutableSingletonList; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.jetbrains.annotations.Nullable; |
| import static org.apache.ignite.internal.client.thin.ProtocolVersionFeature.EXPIRY_POLICY; |
| import static org.apache.ignite.internal.client.thin.ProtocolVersionFeature.QUERY_ENTITY_PRECISION_AND_SCALE; |
| import static org.apache.ignite.internal.processors.platform.cache.expiry.PlatformExpiryPolicy.convertDuration; |
| |
| /** |
| * Shared serialization/deserialization utils. |
| */ |
| public final class ClientUtils { |
| /** Marshaller. */ |
| private final ClientBinaryMarshaller marsh; |
| |
| /** |
| * Constructor. |
| */ |
| ClientUtils(ClientBinaryMarshaller marsh) { |
| this.marsh = marsh; |
| } |
| |
| /** |
| * Get cache ID by cache name. |
| */ |
| static int cacheId(String name) { |
| Objects.requireNonNull(name, "name"); |
| |
| return name.hashCode(); |
| } |
| |
| /** |
| * @param col Collection to serialize. |
| * @param out Output stream. |
| * @param elemWriter Collection element serializer |
| */ |
| public static <E> void collection( |
| Collection<E> col, BinaryOutputStream out, |
| BiConsumer<BinaryOutputStream, E> elemWriter |
| ) { |
| if (col == null || col.isEmpty()) |
| out.writeInt(0); |
| else { |
| out.writeInt(col.size()); |
| |
| for (E e : col) |
| elemWriter.accept(out, e); |
| } |
| } |
| |
| /** |
| * @param col Collection to serialize. |
| * @param out Output stream. |
| * @param elemWriter Collection element serializer |
| */ |
| static <E> void collection(E[] col, BinaryOutputStream out, BiConsumer<BinaryOutputStream, E> elemWriter) { |
| if (col == null || col.length == 0) |
| out.writeInt(0); |
| else { |
| out.writeInt(col.length); |
| |
| for (E e : col) |
| elemWriter.accept(out, e); |
| } |
| } |
| |
| /** |
| * @param in Input stream. |
| * @param elemReader Collection element deserializer. |
| * @return Deserialized collection. |
| */ |
| static <E> Collection<E> collection(BinaryInputStream in, Function<BinaryInputStream, E> elemReader) { |
| Collection<E> col = new LinkedList<>(); // needs to be ordered for some use cases |
| |
| int cnt = in.readInt(); |
| |
| for (int i = 0; i < cnt; i++) |
| col.add(elemReader.apply(in)); |
| |
| return col; |
| } |
| |
| /** |
| * @return Deserialized map |
| */ |
| private static <K, V> Map<K, V> map( |
| BinaryInputStream in, |
| Function<BinaryInputStream, K> keyReader, |
| Function<BinaryInputStream, V> valReader |
| ) { |
| int cnt = in.readInt(); |
| |
| Map<K, V> map = new HashMap<>(cnt); |
| |
| for (int i = 0; i < cnt; i++) |
| map.put(keyReader.apply(in), valReader.apply(in)); |
| |
| return map; |
| } |
| |
| /** Deserialize binary type metadata from stream. */ |
| BinaryMetadata binaryMetadata(BinaryInputStream in) throws IOException { |
| try (BinaryReaderExImpl reader = createBinaryReader(in)) { |
| int typeId = reader.readInt(); |
| String typeName = reader.readString(); |
| String affKeyFieldName = reader.readString(); |
| |
| Map<String, BinaryFieldMetadata> fields = ClientUtils.map( |
| in, |
| unused -> reader.readString(), |
| unused2 -> new BinaryFieldMetadata(reader.readInt(), reader.readInt()) |
| ); |
| |
| boolean isEnum = reader.readBoolean(); |
| |
| Map<String, Integer> enumValues = isEnum ? ClientUtils.map(in, unsed -> reader.readString(), unsed2 -> reader.readInt()) : null; |
| |
| Collection<BinarySchema> schemas = ClientUtils.collection( |
| in, |
| unused -> new BinarySchema( |
| reader.readInt(), |
| new ArrayList<>(ClientUtils.collection(in, unused2 -> reader.readInt())) |
| ) |
| ); |
| |
| return new BinaryMetadata( |
| typeId, |
| typeName, |
| fields, |
| affKeyFieldName, |
| schemas, |
| isEnum, |
| enumValues |
| ); |
| } |
| } |
| |
| /** Serialize binary type metadata to stream. */ |
| void binaryMetadata(BinaryMetadata meta, BinaryOutputStream out) { |
| try (BinaryRawWriterEx w = new BinaryWriterExImpl(marsh.context(), out, null, null)) { |
| w.writeInt(meta.typeId()); |
| w.writeString(meta.typeName()); |
| w.writeString(meta.affinityKeyFieldName()); |
| |
| collection( |
| meta.fieldsMap().entrySet(), |
| out, |
| (unused, e) -> { |
| w.writeString(e.getKey()); |
| w.writeInt(e.getValue().typeId()); |
| w.writeInt(e.getValue().fieldId()); |
| } |
| ); |
| |
| w.writeBoolean(meta.isEnum()); |
| |
| if (meta.isEnum()) |
| collection( |
| meta.enumMap().entrySet(), |
| out, |
| (unused, e) -> { |
| w.writeString(e.getKey()); |
| w.writeInt(e.getValue()); |
| } |
| ); |
| |
| collection( |
| meta.schemas(), |
| out, |
| (unused, s) -> { |
| w.writeInt(s.schemaId()); |
| |
| collection( |
| Arrays.stream(s.fieldIds()).boxed().collect(Collectors.toList()), |
| out, |
| (unused2, i) -> w.writeInt(i) |
| ); |
| } |
| ); |
| } |
| } |
| |
| /** Serialize configuration to stream. */ |
| void cacheConfiguration(ClientCacheConfiguration cfg, BinaryOutputStream out, ProtocolContext protocolCtx) { |
| try (BinaryRawWriterEx writer = new BinaryWriterExImpl(marsh.context(), out, null, null)) { |
| int origPos = out.position(); |
| |
| writer.writeInt(0); // configuration length is to be assigned in the end |
| writer.writeShort((short)0); // properties count is to be assigned in the end |
| |
| AtomicInteger propCnt = new AtomicInteger(0); |
| |
| BiConsumer<CfgItem, Consumer<BinaryRawWriter>> itemWriter = (cfgItem, cfgWriter) -> { |
| writer.writeShort(cfgItem.code()); |
| |
| cfgWriter.accept(writer); |
| |
| propCnt.incrementAndGet(); |
| }; |
| |
| itemWriter.accept(CfgItem.NAME, w -> w.writeString(cfg.getName())); |
| itemWriter.accept(CfgItem.CACHE_MODE, w -> w.writeInt(CacheMode.toCode(cfg.getCacheMode()))); |
| itemWriter.accept(CfgItem.ATOMICITY_MODE, w -> w.writeInt(cfg.getAtomicityMode().ordinal())); |
| itemWriter.accept(CfgItem.BACKUPS, w -> w.writeInt(cfg.getBackups())); |
| itemWriter.accept(CfgItem.WRITE_SYNC_MODE, w -> w.writeInt(cfg.getWriteSynchronizationMode().ordinal())); |
| itemWriter.accept(CfgItem.READ_FROM_BACKUP, w -> w.writeBoolean(cfg.isReadFromBackup())); |
| itemWriter.accept(CfgItem.EAGER_TTL, w -> w.writeBoolean(cfg.isEagerTtl())); |
| itemWriter.accept(CfgItem.GROUP_NAME, w -> w.writeString(cfg.getGroupName())); |
| itemWriter.accept(CfgItem.DEFAULT_LOCK_TIMEOUT, w -> w.writeLong(cfg.getDefaultLockTimeout())); |
| itemWriter.accept(CfgItem.PART_LOSS_POLICY, w -> w.writeInt(cfg.getPartitionLossPolicy().ordinal())); |
| itemWriter.accept(CfgItem.REBALANCE_BATCH_SIZE, w -> w.writeInt(cfg.getRebalanceBatchSize())); |
| itemWriter.accept(CfgItem.REBALANCE_BATCHES_PREFETCH_COUNT, w -> w.writeLong(cfg.getRebalanceBatchesPrefetchCount())); |
| itemWriter.accept(CfgItem.REBALANCE_DELAY, w -> w.writeLong(cfg.getRebalanceDelay())); |
| itemWriter.accept(CfgItem.REBALANCE_MODE, w -> w.writeInt(cfg.getRebalanceMode().ordinal())); |
| itemWriter.accept(CfgItem.REBALANCE_ORDER, w -> w.writeInt(cfg.getRebalanceOrder())); |
| itemWriter.accept(CfgItem.REBALANCE_THROTTLE, w -> w.writeLong(cfg.getRebalanceThrottle())); |
| itemWriter.accept(CfgItem.REBALANCE_TIMEOUT, w -> w.writeLong(cfg.getRebalanceTimeout())); |
| itemWriter.accept(CfgItem.COPY_ON_READ, w -> w.writeBoolean(cfg.isCopyOnRead())); |
| itemWriter.accept(CfgItem.DATA_REGION_NAME, w -> w.writeString(cfg.getDataRegionName())); |
| itemWriter.accept(CfgItem.STATS_ENABLED, w -> w.writeBoolean(cfg.isStatisticsEnabled())); |
| itemWriter.accept(CfgItem.MAX_ASYNC_OPS, w -> w.writeInt(cfg.getMaxConcurrentAsyncOperations())); |
| itemWriter.accept(CfgItem.MAX_QUERY_ITERATORS, w -> w.writeInt(cfg.getMaxQueryIteratorsCount())); |
| itemWriter.accept(CfgItem.ONHEAP_CACHE_ENABLED, w -> w.writeBoolean(cfg.isOnheapCacheEnabled())); |
| itemWriter.accept(CfgItem.QUERY_METRIC_SIZE, w -> w.writeInt(cfg.getQueryDetailMetricsSize())); |
| itemWriter.accept(CfgItem.QUERY_PARALLELISM, w -> w.writeInt(cfg.getQueryParallelism())); |
| itemWriter.accept(CfgItem.SQL_ESCAPE_ALL, w -> w.writeBoolean(cfg.isSqlEscapeAll())); |
| itemWriter.accept(CfgItem.SQL_IDX_MAX_INLINE_SIZE, w -> w.writeInt(cfg.getSqlIndexMaxInlineSize())); |
| itemWriter.accept(CfgItem.SQL_SCHEMA, w -> w.writeString(cfg.getSqlSchema())); |
| itemWriter.accept( |
| CfgItem.KEY_CONFIGS, |
| w -> ClientUtils.collection( |
| cfg.getKeyConfiguration(), |
| out, |
| (unused, i) -> { |
| w.writeString(i.getTypeName()); |
| w.writeString(i.getAffinityKeyFieldName()); |
| } |
| ) |
| ); |
| |
| itemWriter.accept( |
| CfgItem.QUERY_ENTITIES, |
| w -> ClientUtils.collection( |
| cfg.getQueryEntities(), |
| out, (unused, e) -> { |
| w.writeString(e.getKeyType()); |
| w.writeString(e.getValueType()); |
| w.writeString(e.getTableName()); |
| w.writeString(e.getKeyFieldName()); |
| w.writeString(e.getValueFieldName()); |
| ClientUtils.collection( |
| e.getFields().entrySet(), |
| out, |
| (unused2, f) -> { |
| QueryField qf = new QueryField(e, f); |
| |
| w.writeString(qf.getName()); |
| w.writeString(qf.getTypeName()); |
| w.writeBoolean(qf.isKey()); |
| w.writeBoolean(qf.isNotNull()); |
| w.writeObject(qf.getDefaultValue()); |
| |
| if (protocolCtx.isFeatureSupported(QUERY_ENTITY_PRECISION_AND_SCALE)) { |
| w.writeInt(qf.getPrecision()); |
| w.writeInt(qf.getScale()); |
| } |
| } |
| ); |
| ClientUtils.collection( |
| e.getAliases().entrySet(), |
| out, (unused3, a) -> { |
| w.writeString(a.getKey()); |
| w.writeString(a.getValue()); |
| } |
| ); |
| ClientUtils.collection( |
| e.getIndexes(), |
| out, |
| (unused4, i) -> { |
| w.writeString(i.getName()); |
| w.writeByte((byte)i.getIndexType().ordinal()); |
| w.writeInt(i.getInlineSize()); |
| ClientUtils.collection(i.getFields().entrySet(), out, (unused5, f) -> { |
| w.writeString(f.getKey()); |
| w.writeBoolean(f.getValue()); |
| } |
| ); |
| }); |
| } |
| ) |
| ); |
| |
| if (protocolCtx.isFeatureSupported(EXPIRY_POLICY)) { |
| itemWriter.accept(CfgItem.EXPIRE_POLICY, w -> { |
| ExpiryPolicy expiryPlc = cfg.getExpiryPolicy(); |
| if (expiryPlc == null) |
| w.writeBoolean(false); |
| else { |
| w.writeBoolean(true); |
| w.writeLong(convertDuration(expiryPlc.getExpiryForCreation())); |
| w.writeLong(convertDuration(expiryPlc.getExpiryForUpdate())); |
| w.writeLong(convertDuration(expiryPlc.getExpiryForAccess())); |
| } |
| }); |
| } |
| else if (cfg.getExpiryPolicy() != null) { |
| throw new ClientProtocolError(String.format("Expire policies are not supported by the server " + |
| "version %s, required version %s", protocolCtx.version(), EXPIRY_POLICY.verIntroduced())); |
| } |
| |
| writer.writeInt(origPos, out.position() - origPos - 4); // configuration length |
| writer.writeInt(origPos + 4, propCnt.get()); // properties count |
| } |
| } |
| |
| /** Deserialize configuration from stream. */ |
| ClientCacheConfiguration cacheConfiguration(BinaryInputStream in, ProtocolContext protocolCtx) |
| throws IOException { |
| try (BinaryReaderExImpl reader = createBinaryReader(in)) { |
| reader.readInt(); // Do not need length to read data. The protocol defines fixed configuration layout. |
| |
| return new ClientCacheConfiguration().setName("TBD") // cache name is to be assigned later |
| .setAtomicityMode(CacheAtomicityMode.fromOrdinal(reader.readInt())) |
| .setBackups(reader.readInt()) |
| .setCacheMode(CacheMode.fromCode(reader.readInt())) |
| .setCopyOnRead(reader.readBoolean()) |
| .setDataRegionName(reader.readString()) |
| .setEagerTtl(reader.readBoolean()) |
| .setStatisticsEnabled(reader.readBoolean()) |
| .setGroupName(reader.readString()) |
| .setDefaultLockTimeout(reader.readLong()) |
| .setMaxConcurrentAsyncOperations(reader.readInt()) |
| .setMaxQueryIteratorsCount(reader.readInt()) |
| .setName(reader.readString()) |
| .setOnheapCacheEnabled(reader.readBoolean()) |
| .setPartitionLossPolicy(PartitionLossPolicy.fromOrdinal((byte)reader.readInt())) |
| .setQueryDetailMetricsSize(reader.readInt()) |
| .setQueryParallelism(reader.readInt()) |
| .setReadFromBackup(reader.readBoolean()) |
| .setRebalanceBatchSize(reader.readInt()) |
| .setRebalanceBatchesPrefetchCount(reader.readLong()) |
| .setRebalanceDelay(reader.readLong()) |
| .setRebalanceMode(CacheRebalanceMode.fromOrdinal(reader.readInt())) |
| .setRebalanceOrder(reader.readInt()) |
| .setRebalanceThrottle(reader.readLong()) |
| .setRebalanceTimeout(reader.readLong()) |
| .setSqlEscapeAll(reader.readBoolean()) |
| .setSqlIndexMaxInlineSize(reader.readInt()) |
| .setSqlSchema(reader.readString()) |
| .setWriteSynchronizationMode(CacheWriteSynchronizationMode.fromOrdinal(reader.readInt())) |
| .setKeyConfiguration( |
| ClientUtils.collection(in, unused -> new CacheKeyConfiguration(reader.readString(), reader.readString())) |
| .toArray(new CacheKeyConfiguration[0]) |
| ).setQueryEntities(ClientUtils.collection( |
| in, |
| unused -> { |
| QueryEntity qryEntity = new QueryEntity(reader.readString(), reader.readString()) |
| .setTableName(reader.readString()) |
| .setKeyFieldName(reader.readString()) |
| .setValueFieldName(reader.readString()); |
| |
| boolean isPrecisionAndScaleSupported = |
| protocolCtx.isFeatureSupported(QUERY_ENTITY_PRECISION_AND_SCALE); |
| |
| Collection<QueryField> qryFields = ClientUtils.collection( |
| in, |
| unused2 -> { |
| String name = reader.readString(); |
| String typeName = reader.readString(); |
| boolean isKey = reader.readBoolean(); |
| boolean isNotNull = reader.readBoolean(); |
| Object dfltVal = reader.readObject(); |
| int precision = isPrecisionAndScaleSupported ? reader.readInt() : -1; |
| int scale = isPrecisionAndScaleSupported ? reader.readInt() : -1; |
| |
| return new QueryField(name, |
| typeName, |
| isKey, |
| isNotNull, |
| dfltVal, |
| precision, |
| scale); |
| } |
| ); |
| |
| return qryEntity |
| .setFields(qryFields.stream().collect(Collectors.toMap( |
| QueryField::getName, QueryField::getTypeName, (a, b) -> a, LinkedHashMap::new |
| ))) |
| .setKeyFields(qryFields.stream() |
| .filter(QueryField::isKey) |
| .map(QueryField::getName) |
| .collect(Collectors.toCollection(LinkedHashSet::new)) |
| ) |
| .setNotNullFields(qryFields.stream() |
| .filter(QueryField::isNotNull) |
| .map(QueryField::getName) |
| .collect(Collectors.toSet()) |
| ) |
| .setDefaultFieldValues(qryFields.stream() |
| .filter(f -> f.getDefaultValue() != null) |
| .collect(Collectors.toMap(QueryField::getName, QueryField::getDefaultValue)) |
| ) |
| .setFieldsPrecision(qryFields.stream() |
| .filter(f -> f.getPrecision() != -1) |
| .collect(Collectors.toMap(QueryField::getName, QueryField::getPrecision)) |
| ) |
| .setFieldsScale(qryFields.stream() |
| .filter(f -> f.getScale() != -1) |
| .collect(Collectors.toMap(QueryField::getName, QueryField::getScale)) |
| ) |
| .setAliases(ClientUtils.collection( |
| in, |
| unused3 -> new SimpleEntry<>(reader.readString(), reader.readString()) |
| ).stream().collect(Collectors.toMap(SimpleEntry::getKey, SimpleEntry::getValue))) |
| .setIndexes(ClientUtils.collection( |
| in, |
| unused4 -> { |
| String name = reader.readString(); |
| QueryIndexType type = QueryIndexType.fromOrdinal(reader.readByte()); |
| int inlineSize = reader.readInt(); |
| |
| LinkedHashMap<String, Boolean> fields = ClientUtils.collection( |
| in, |
| unused5 -> new SimpleEntry<>(reader.readString(), reader.readBoolean()) |
| ).stream().collect(Collectors.toMap( |
| SimpleEntry::getKey, |
| SimpleEntry::getValue, |
| (a, b) -> a, |
| LinkedHashMap::new |
| )); |
| |
| return new QueryIndex(fields, type).setName(name).setInlineSize(inlineSize); |
| } |
| )); |
| } |
| ).toArray(new QueryEntity[0])) |
| .setExpiryPolicy(!protocolCtx.isFeatureSupported(EXPIRY_POLICY) ? |
| null : reader.readBoolean() ? |
| new PlatformExpiryPolicy(reader.readLong(), reader.readLong(), reader.readLong()) : null |
| ); |
| } |
| } |
| |
| /** Serialize SQL field query to stream. */ |
| void write(SqlFieldsQuery qry, BinaryOutputStream out) { |
| writeObject(out, qry.getSchema()); |
| out.writeInt(qry.getPageSize()); |
| out.writeInt(-1); // do not limit |
| writeObject(out, qry.getSql()); |
| ClientUtils.collection(qry.getArgs() == null ? null : Arrays.asList(qry.getArgs()), out, this::writeObject); |
| out.writeByte((byte)0); // statement type ANY |
| out.writeBoolean(qry.isDistributedJoins()); |
| out.writeBoolean(qry.isLocal()); |
| out.writeBoolean(qry.isReplicatedOnly()); |
| out.writeBoolean(qry.isEnforceJoinOrder()); |
| out.writeBoolean(qry.isCollocated()); |
| out.writeBoolean(qry.isLazy()); |
| out.writeLong(qry.getTimeout()); |
| out.writeBoolean(true); // include column names |
| |
| if (qry.getPartitions() != null) { |
| out.writeInt(qry.getPartitions().length); |
| |
| for (int part : qry.getPartitions()) |
| out.writeInt(part); |
| } |
| else |
| out.writeInt(-1); |
| |
| out.writeInt(qry.getUpdateBatchSize()); |
| } |
| |
| /** Write Ignite binary object to output stream. */ |
| void writeObject(BinaryOutputStream out, Object obj) { |
| out.writeByteArray(marsh.marshal(obj)); |
| } |
| |
| /** |
| * @param out Output stream. |
| */ |
| BinaryRawWriterEx createBinaryWriter(BinaryOutputStream out) { |
| return new BinaryWriterExImpl(marsh.context(), out, BinaryThreadLocalContext.get().schemaHolder(), null); |
| } |
| |
| /** |
| * @param in Input stream. |
| */ |
| BinaryReaderExImpl createBinaryReader(BinaryInputStream in) { |
| return createBinaryReader(marsh.context(), in); |
| } |
| |
| /** |
| * @param binaryCtx Binary context. |
| * @param in Input stream. |
| */ |
| static BinaryReaderExImpl createBinaryReader(@Nullable BinaryContext binaryCtx, BinaryInputStream in) { |
| return new BinaryReaderExImpl(binaryCtx, in, null, null, true, true); |
| } |
| |
| /** Read Ignite binary object from input stream. */ |
| <T> T readObject(BinaryInputStream in, boolean keepBinary) { |
| return readObject(in, keepBinary, null); |
| } |
| |
| /** Read Ignite binary object from input stream. */ |
| <T> T readObject(BinaryInputStream in, boolean keepBinary, Class<T> clazz) { |
| if (keepBinary) |
| return (T)marsh.unmarshal(in); |
| else { |
| BinaryReaderHandles hnds = new BinaryReaderHandles(); |
| |
| return (T)unwrapBinary(marsh.deserialize(in, hnds), hnds, clazz); |
| } |
| } |
| |
| /** |
| * Unwrap binary object. |
| */ |
| private Object unwrapBinary(Object obj, BinaryReaderHandles hnds, Class<?> clazz) { |
| if (obj instanceof BinaryObjectImpl) { |
| BinaryObjectImpl obj0 = (BinaryObjectImpl)obj; |
| |
| return marsh.deserialize(BinaryHeapInputStream.create(obj0.array(), obj0.start()), hnds); |
| } |
| else if (obj instanceof BinaryObject) |
| return ((BinaryObject)obj).deserialize(); |
| else if (BinaryUtils.knownCollection(obj)) |
| return unwrapCollection((Collection<Object>)obj, hnds); |
| else if (BinaryUtils.knownMap(obj)) |
| return unwrapMap((Map<Object, Object>)obj, hnds); |
| else if (obj instanceof Object[]) |
| return unwrapArray((Object[])obj, hnds, clazz); |
| else |
| return obj; |
| } |
| |
| /** |
| * Unwrap collection with binary objects. |
| */ |
| private Collection<Object> unwrapCollection(Collection<Object> col, BinaryReaderHandles hnds) { |
| Collection<Object> col0 = BinaryUtils.newKnownCollection(col); |
| |
| for (Object obj0 : col) |
| col0.add(unwrapBinary(obj0, hnds, null)); |
| |
| return (col0 instanceof MutableSingletonList) ? U.convertToSingletonList(col0) : col0; |
| } |
| |
| /** |
| * Unwrap map with binary objects. |
| */ |
| private Map<Object, Object> unwrapMap(Map<Object, Object> map, BinaryReaderHandles hnds) { |
| Map<Object, Object> map0 = BinaryUtils.newMap(map); |
| |
| for (Map.Entry<Object, Object> e : map.entrySet()) |
| map0.put(unwrapBinary(e.getKey(), hnds, null), unwrapBinary(e.getValue(), hnds, null)); |
| |
| return map0; |
| } |
| |
| /** |
| * Unwrap array with binary objects. |
| */ |
| private Object[] unwrapArray(Object[] arr, BinaryReaderHandles hnds, Class<?> arrayClass) { |
| if (BinaryUtils.knownArray(arr)) |
| return arr; |
| |
| Class<?> componentType = arrayClass != null && arrayClass.isArray() |
| ? arrayClass.getComponentType() |
| : arr.getClass().getComponentType(); |
| |
| Object[] res = (Object[])Array.newInstance(componentType, arr.length); |
| |
| for (int i = 0; i < arr.length; i++) |
| res[i] = unwrapBinary(arr[i], hnds, null); |
| |
| return res; |
| } |
| |
| /** A helper class to translate query fields. */ |
| private static final class QueryField { |
| /** Name. */ |
| private final String name; |
| |
| /** Type name. */ |
| private final String typeName; |
| |
| /** Is key. */ |
| private final boolean isKey; |
| |
| /** Is not null. */ |
| private final boolean isNotNull; |
| |
| /** Default value. */ |
| private final Object dfltVal; |
| |
| /** Precision. */ |
| private final int precision; |
| |
| /** Scale. */ |
| private final int scale; |
| |
| /** Serialization constructor. */ |
| QueryField(QueryEntity e, Map.Entry<String, String> nameAndTypeName) { |
| name = nameAndTypeName.getKey(); |
| typeName = nameAndTypeName.getValue(); |
| |
| Set<String> keys = e.getKeyFields(); |
| Set<String> notNulls = e.getNotNullFields(); |
| Map<String, Object> dflts = e.getDefaultFieldValues(); |
| Map<String, Integer> fldsPrecision = e.getFieldsPrecision(); |
| Map<String, Integer> fldsScale = e.getFieldsScale(); |
| |
| isKey = keys != null && keys.contains(name); |
| isNotNull = notNulls != null && notNulls.contains(name); |
| dfltVal = dflts == null ? null : dflts.get(name); |
| precision = fldsPrecision == null ? -1 : fldsPrecision.getOrDefault(name, -1); |
| scale = fldsScale == null ? -1 : fldsScale.getOrDefault(name, -1); |
| } |
| |
| /** Deserialization constructor. */ |
| public QueryField(String name, String typeName, boolean isKey, boolean isNotNull, Object dfltVal, |
| int precision, int scale) { |
| this.name = name; |
| this.typeName = typeName; |
| this.isKey = isKey; |
| this.isNotNull = isNotNull; |
| this.dfltVal = dfltVal; |
| this.precision = precision; |
| this.scale = scale; |
| } |
| |
| /** |
| * @return Name. |
| */ |
| String getName() { |
| return name; |
| } |
| |
| /** |
| * @return Type name. |
| */ |
| String getTypeName() { |
| return typeName; |
| } |
| |
| /** |
| * @return Is Key. |
| */ |
| boolean isKey() { |
| return isKey; |
| } |
| |
| /** |
| * @return Is Not Null. |
| */ |
| boolean isNotNull() { |
| return isNotNull; |
| } |
| |
| /** |
| * @return Default value. |
| */ |
| Object getDefaultValue() { |
| return dfltVal; |
| } |
| |
| /** |
| * @return Precision. |
| */ |
| public int getPrecision() { |
| return precision; |
| } |
| |
| /** |
| * @return Scale. |
| */ |
| public int getScale() { |
| return scale; |
| } |
| } |
| |
| /** Thin client protocol cache configuration item codes. */ |
| private enum CfgItem { |
| /** Name. */ |
| NAME(0), |
| |
| /** Cache mode. */ |
| CACHE_MODE(1), |
| |
| /** Atomicity mode. */ |
| ATOMICITY_MODE(2), |
| |
| /** Backups. */ |
| BACKUPS(3), |
| |
| /** Write synchronization mode. */ |
| WRITE_SYNC_MODE(4), |
| |
| /** Read from backup. */ |
| READ_FROM_BACKUP(6), |
| |
| /** Eager ttl. */ |
| EAGER_TTL(405), |
| |
| /** Group name. */ |
| GROUP_NAME(400), |
| |
| /** Default lock timeout. */ |
| DEFAULT_LOCK_TIMEOUT(402), |
| |
| /** Partition loss policy. */ |
| PART_LOSS_POLICY(404), |
| |
| /** Rebalance batch size. */ |
| REBALANCE_BATCH_SIZE(303), |
| |
| /** Rebalance batches prefetch count. */ |
| REBALANCE_BATCHES_PREFETCH_COUNT(304), |
| |
| /** Rebalance delay. */ |
| REBALANCE_DELAY(301), |
| |
| /** Rebalance mode. */ |
| REBALANCE_MODE(300), |
| |
| /** Rebalance order. */ |
| REBALANCE_ORDER(305), |
| |
| /** Rebalance throttle. */ |
| REBALANCE_THROTTLE(306), |
| |
| /** Rebalance timeout. */ |
| REBALANCE_TIMEOUT(302), |
| |
| /** Copy on read. */ |
| COPY_ON_READ(5), |
| |
| /** Data region name. */ |
| DATA_REGION_NAME(100), |
| |
| /** Stats enabled. */ |
| STATS_ENABLED(406), |
| |
| /** Max async ops. */ |
| MAX_ASYNC_OPS(403), |
| |
| /** Max query iterators. */ |
| MAX_QUERY_ITERATORS(206), |
| |
| /** Onheap cache enabled. */ |
| ONHEAP_CACHE_ENABLED(101), |
| |
| /** Query metric size. */ |
| QUERY_METRIC_SIZE(202), |
| |
| /** Query parallelism. */ |
| QUERY_PARALLELISM(201), |
| |
| /** Sql escape all. */ |
| SQL_ESCAPE_ALL(205), |
| |
| /** Sql index max inline size. */ |
| SQL_IDX_MAX_INLINE_SIZE(204), |
| |
| /** Sql schema. */ |
| SQL_SCHEMA(203), |
| |
| /** Key configs. */ |
| KEY_CONFIGS(401), |
| |
| /** Key entities. */ |
| QUERY_ENTITIES(200), |
| |
| /** Expire policy. */ |
| EXPIRE_POLICY(407); |
| |
| /** Code. */ |
| private final short code; |
| |
| /** */ |
| CfgItem(int code) { |
| this.code = (short)code; |
| } |
| |
| /** @return Code. */ |
| short code() { |
| return code; |
| } |
| } |
| } |
