You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2021/12/27 06:20:14 UTC
[ignite-3] branch main updated: IGNITE-15240 Thin client: Implement transactions (#524)
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 9177a98 IGNITE-15240 Thin client: Implement transactions (#524)
9177a98 is described below
commit 9177a98b9c93253b9e449c0f2c33ad5257cba95f
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Mon Dec 27 09:20:03 2021 +0300
IGNITE-15240 Thin client: Implement transactions (#524)
* Add TX_BEGIN, TX_COMMIT, TX_ROLLBACK operations to the protocol.
* Add `ClientResourceRegistry` to store active transactions.
* Pass transaction id from all table API variants.
---
.../org/apache/ignite/tx/IgniteTransactions.java | 27 +-
.../ignite/internal/client/proto/ClientOp.java | 9 +
.../ignite/client/handler/ItClientHandlerTest.java | 4 +-
.../ignite/client/handler/ClientHandlerModule.java | 25 +-
.../handler/ClientInboundMessageHandler.java | 97 ++++---
.../ignite/client/handler/ClientResource.java | 61 ++++
.../client/handler/ClientResourceRegistry.java | 104 +++++++
.../handler/requests/table/ClientTableCommon.java | 18 ++
.../table/ClientTupleContainsKeyRequest.java | 15 +-
.../table/ClientTupleDeleteAllExactRequest.java | 15 +-
.../table/ClientTupleDeleteAllRequest.java | 15 +-
.../table/ClientTupleDeleteExactRequest.java | 15 +-
.../requests/table/ClientTupleDeleteRequest.java | 15 +-
.../requests/table/ClientTupleGetAllRequest.java | 15 +-
.../table/ClientTupleGetAndDeleteRequest.java | 15 +-
.../table/ClientTupleGetAndReplaceRequest.java | 15 +-
.../table/ClientTupleGetAndUpsertRequest.java | 15 +-
.../requests/table/ClientTupleGetRequest.java | 15 +-
.../table/ClientTupleInsertAllRequest.java | 15 +-
.../requests/table/ClientTupleInsertRequest.java | 15 +-
.../table/ClientTupleReplaceExactRequest.java | 15 +-
.../requests/table/ClientTupleReplaceRequest.java | 15 +-
.../table/ClientTupleUpsertAllRequest.java | 16 +-
.../requests/table/ClientTupleUpsertRequest.java | 16 +-
.../ClientTransactionBeginRequest.java} | 30 +-
.../ClientTransactionCommitRequest.java} | 25 +-
.../ClientTransactionRollbackRequest.java} | 25 +-
.../ignite/internal/client/ClientChannel.java | 2 +-
.../ignite/internal/client/ReliableChannel.java | 3 +-
.../ignite/internal/client/TcpIgniteClient.java | 7 +-
.../client/table/ClientKeyValueBinaryView.java | 77 ++---
.../internal/client/table/ClientKeyValueView.java | 81 ++----
.../client/table/ClientRecordBinaryView.java | 79 ++---
.../client/table/ClientRecordSerializer.java | 46 ++-
.../internal/client/table/ClientRecordView.java | 194 +++----------
.../ignite/internal/client/table/ClientTable.java | 286 ++----------------
.../client/table/ClientTupleSerializer.java | 320 +++++++++++++++++++++
.../internal/client/tx/ClientTransaction.java | 117 ++++++++
.../internal/client/tx/ClientTransactions.java} | 55 ++--
.../org/apache/ignite/client/ClientTableTest.java | 2 +-
.../java/org/apache/ignite/client/TestServer.java | 8 +-
.../org/apache/ignite/client/fakes/FakeIgnite.java | 19 +-
.../dotnet/Apache.Ignite/Internal/Table/Table.cs | 2 +
.../ItAbstractThinClientTest.java} | 120 ++++----
.../app/client/ItThinClientConnectionTest.java | 82 ++++++
.../app/client/ItThinClientTransactionsTest.java | 312 ++++++++++++++++++++
.../org/apache/ignite/internal/app/IgniteImpl.java | 4 +-
.../internal/tx/impl/IgniteTransactionsImpl.java | 38 +--
48 files changed, 1659 insertions(+), 862 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/tx/IgniteTransactions.java b/modules/api/src/main/java/org/apache/ignite/tx/IgniteTransactions.java
index 3fcbc5b..1d4f92b 100644
--- a/modules/api/src/main/java/org/apache/ignite/tx/IgniteTransactions.java
+++ b/modules/api/src/main/java/org/apache/ignite/tx/IgniteTransactions.java
@@ -57,7 +57,12 @@ public interface IgniteTransactions {
*
* @throws TransactionException If a transaction can't be finished successfully.
*/
- void runInTransaction(Consumer<Transaction> clo) throws TransactionException;
+ default void runInTransaction(Consumer<Transaction> clo) throws TransactionException {
+ runInTransaction(tx -> {
+ clo.accept(tx);
+ return null;
+ });
+ }
/**
* Executes a closure within a transaction and returns a result.
@@ -73,5 +78,23 @@ public interface IgniteTransactions {
*
* @throws TransactionException If a transaction can't be finished successfully.
*/
- <T> T runInTransaction(Function<Transaction, T> clo) throws TransactionException;
+ default <T> T runInTransaction(Function<Transaction, T> clo) throws TransactionException {
+ Transaction tx = begin();
+
+ try {
+ T ret = clo.apply(tx);
+
+ tx.commit();
+
+ return ret;
+ } catch (Throwable t) {
+ try {
+ tx.rollback(); // Try rolling back on user exception.
+ } catch (Exception e) {
+ t.addSuppressed(e);
+ }
+
+ throw t;
+ }
+ }
}
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
index d12ba5c..0aaef34 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
@@ -110,4 +110,13 @@ public class ClientOp {
/** Get query metadata. */
public static final int SQL_QUERY_META = 42;
+
+ /** Begin transaction. */
+ public static final int TX_BEGIN = 43;
+
+ /** Commit transaction. */
+ public static final int TX_COMMIT = 44;
+
+ /** Rollback transaction. */
+ public static final int TX_ROLLBACK = 45;
}
diff --git a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java
index c14286a..401fa40 100644
--- a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java
+++ b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.configuration.storage.TestConfigurationStorage
import org.apache.ignite.internal.processors.query.calcite.QueryProcessor;
import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.IgniteTransactions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -203,7 +204,8 @@ public class ItClientHandlerTest {
bootstrapFactory.start();
- var module = new ClientHandlerModule(mock(QueryProcessor.class), mock(IgniteTables.class), registry, bootstrapFactory);
+ var module = new ClientHandlerModule(mock(QueryProcessor.class), mock(IgniteTables.class), mock(IgniteTransactions.class), registry,
+ bootstrapFactory);
module.start();
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
index 461256d..fb36c91 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
@@ -33,6 +33,7 @@ import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.network.NettyBootstrapFactory;
import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.IgniteTransactions;
import org.jetbrains.annotations.Nullable;
/**
@@ -48,11 +49,14 @@ public class ClientHandlerModule implements IgniteComponent {
/** Ignite tables API. */
private final IgniteTables igniteTables;
+ /** Ignite transactions API. */
+ private final IgniteTransactions igniteTransactions;
+
/** Netty channel. */
private volatile Channel channel;
/** Processor. */
- private final QueryProcessor processor;
+ private final QueryProcessor queryProcessor;
/** Netty bootstrap factory. */
private final NettyBootstrapFactory bootstrapFactory;
@@ -60,23 +64,26 @@ public class ClientHandlerModule implements IgniteComponent {
/**
* Constructor.
*
- * @param processor Sql query processor.
- * @param igniteTables Ignite.
- * @param registry Configuration registry.
- * @param bootstrapFactory Bootstrap factory.
+ * @param queryProcessor Sql query processor.
+ * @param igniteTables Ignite.
+ * @param igniteTransactions Transactions.
+ * @param registry Configuration registry.
+ * @param bootstrapFactory Bootstrap factory.
*/
public ClientHandlerModule(
- QueryProcessor processor,
+ QueryProcessor queryProcessor,
IgniteTables igniteTables,
+ IgniteTransactions igniteTransactions,
ConfigurationRegistry registry,
NettyBootstrapFactory bootstrapFactory) {
assert igniteTables != null;
assert registry != null;
- assert processor != null;
+ assert queryProcessor != null;
assert bootstrapFactory != null;
- this.processor = processor;
+ this.queryProcessor = queryProcessor;
this.igniteTables = igniteTables;
+ this.igniteTransactions = igniteTransactions;
this.registry = registry;
this.bootstrapFactory = bootstrapFactory;
}
@@ -138,7 +145,7 @@ public class ClientHandlerModule implements IgniteComponent {
protected void initChannel(Channel ch) {
ch.pipeline().addLast(
new ClientMessageDecoder(),
- new ClientInboundMessageHandler(igniteTables, processor));
+ new ClientInboundMessageHandler(igniteTables, igniteTransactions, queryProcessor));
}
})
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, configuration.connectTimeout());
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
index d474ed9..4b98153 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
@@ -54,6 +54,9 @@ import org.apache.ignite.client.handler.requests.table.ClientTupleReplaceExactRe
import org.apache.ignite.client.handler.requests.table.ClientTupleReplaceRequest;
import org.apache.ignite.client.handler.requests.table.ClientTupleUpsertAllRequest;
import org.apache.ignite.client.handler.requests.table.ClientTupleUpsertRequest;
+import org.apache.ignite.client.handler.requests.tx.ClientTransactionBeginRequest;
+import org.apache.ignite.client.handler.requests.tx.ClientTransactionCommitRequest;
+import org.apache.ignite.client.handler.requests.tx.ClientTransactionRollbackRequest;
import org.apache.ignite.client.proto.query.JdbcQueryEventHandler;
import org.apache.ignite.internal.client.proto.ClientErrorCode;
import org.apache.ignite.internal.client.proto.ClientMessageCommon;
@@ -66,6 +69,7 @@ import org.apache.ignite.internal.processors.query.calcite.QueryProcessor;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.IgniteTransactions;
/**
* Handles messages from thin clients.
@@ -78,25 +82,37 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter {
/** Ignite tables API. */
private final IgniteTables igniteTables;
+ /** Ignite tables API. */
+ private final IgniteTransactions igniteTransactions;
+
+ /** JDBC Handler. */
+ private final JdbcQueryEventHandler jdbcQueryEventHandler;
+
+ /** Connection resources. */
+ private final ClientResourceRegistry resources = new ClientResourceRegistry();
+
/** Context. */
private ClientContext clientContext;
- /** Handler. */
- private final JdbcQueryEventHandler handler;
-
/**
* Constructor.
*
- * @param igniteTables Ignite tables API entry point.
- * @param processor Sql query processor.
+ * @param igniteTables Ignite tables API entry point.
+ * @param igniteTransactions Transactions API.
+ * @param processor Sql query processor.
*/
- public ClientInboundMessageHandler(IgniteTables igniteTables,
+ public ClientInboundMessageHandler(
+ IgniteTables igniteTables,
+ IgniteTransactions igniteTransactions,
QueryProcessor processor) {
assert igniteTables != null;
+ assert igniteTransactions != null;
+ assert processor != null;
this.igniteTables = igniteTables;
+ this.igniteTransactions = igniteTransactions;
- this.handler = new JdbcQueryEventHandlerImpl(processor, new JdbcMetadataCatalog(igniteTables));
+ this.jdbcQueryEventHandler = new JdbcQueryEventHandlerImpl(processor, new JdbcMetadataCatalog(igniteTables));
}
/** {@inheritDoc} */
@@ -115,6 +131,14 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter {
}
}
+ /** {@inheritDoc} */
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ resources.clean();
+
+ super.channelInactive(ctx);
+ }
+
private void handshake(ChannelHandlerContext ctx, ClientMessageUnpacker unpacker, ClientMessagePacker packer) {
try {
writeMagic(ctx);
@@ -268,79 +292,88 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter {
return ClientTableGetRequest.process(in, out, igniteTables);
case ClientOp.TUPLE_UPSERT:
- return ClientTupleUpsertRequest.process(in, igniteTables);
+ return ClientTupleUpsertRequest.process(in, igniteTables, resources);
case ClientOp.TUPLE_GET:
- return ClientTupleGetRequest.process(in, out, igniteTables);
+ return ClientTupleGetRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_UPSERT_ALL:
- return ClientTupleUpsertAllRequest.process(in, igniteTables);
+ return ClientTupleUpsertAllRequest.process(in, igniteTables, resources);
case ClientOp.TUPLE_GET_ALL:
- return ClientTupleGetAllRequest.process(in, out, igniteTables);
+ return ClientTupleGetAllRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_GET_AND_UPSERT:
- return ClientTupleGetAndUpsertRequest.process(in, out, igniteTables);
+ return ClientTupleGetAndUpsertRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_INSERT:
- return ClientTupleInsertRequest.process(in, out, igniteTables);
+ return ClientTupleInsertRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_INSERT_ALL:
- return ClientTupleInsertAllRequest.process(in, out, igniteTables);
+ return ClientTupleInsertAllRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_REPLACE:
- return ClientTupleReplaceRequest.process(in, out, igniteTables);
+ return ClientTupleReplaceRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_REPLACE_EXACT:
- return ClientTupleReplaceExactRequest.process(in, out, igniteTables);
+ return ClientTupleReplaceExactRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_GET_AND_REPLACE:
- return ClientTupleGetAndReplaceRequest.process(in, out, igniteTables);
+ return ClientTupleGetAndReplaceRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_DELETE:
- return ClientTupleDeleteRequest.process(in, out, igniteTables);
+ return ClientTupleDeleteRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_DELETE_ALL:
- return ClientTupleDeleteAllRequest.process(in, out, igniteTables);
+ return ClientTupleDeleteAllRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_DELETE_EXACT:
- return ClientTupleDeleteExactRequest.process(in, out, igniteTables);
+ return ClientTupleDeleteExactRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_DELETE_ALL_EXACT:
- return ClientTupleDeleteAllExactRequest.process(in, out, igniteTables);
+ return ClientTupleDeleteAllExactRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_GET_AND_DELETE:
- return ClientTupleGetAndDeleteRequest.process(in, out, igniteTables);
+ return ClientTupleGetAndDeleteRequest.process(in, out, igniteTables, resources);
case ClientOp.TUPLE_CONTAINS_KEY:
- return ClientTupleContainsKeyRequest.process(in, out, igniteTables);
+ return ClientTupleContainsKeyRequest.process(in, out, igniteTables, resources);
case ClientOp.SQL_EXEC:
- return ClientSqlExecuteRequest.execute(in, out, handler);
+ return ClientSqlExecuteRequest.execute(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_EXEC_BATCH:
- return ClientSqlExecuteBatchRequest.process(in, out, handler);
+ return ClientSqlExecuteBatchRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_NEXT:
- return ClientSqlFetchRequest.process(in, out, handler);
+ return ClientSqlFetchRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_CURSOR_CLOSE:
- return ClientSqlCloseRequest.process(in, out, handler);
+ return ClientSqlCloseRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_TABLE_META:
- return ClientSqlTableMetadataRequest.process(in, out, handler);
+ return ClientSqlTableMetadataRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_COLUMN_META:
- return ClientSqlColumnMetadataRequest.process(in, out, handler);
+ return ClientSqlColumnMetadataRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_SCHEMAS_META:
- return ClientSqlSchemasMetadataRequest.process(in, out, handler);
+ return ClientSqlSchemasMetadataRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_PK_META:
- return ClientSqlPrimaryKeyMetadataRequest.process(in, out, handler);
+ return ClientSqlPrimaryKeyMetadataRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.SQL_QUERY_META:
- return ClientSqlQueryMetadataRequest.process(in, out, handler);
+ return ClientSqlQueryMetadataRequest.process(in, out, jdbcQueryEventHandler);
+
+ case ClientOp.TX_BEGIN:
+ return ClientTransactionBeginRequest.process(out, igniteTransactions, resources);
+
+ case ClientOp.TX_COMMIT:
+ return ClientTransactionCommitRequest.process(in, resources);
+
+ case ClientOp.TX_ROLLBACK:
+ return ClientTransactionRollbackRequest.process(in, resources);
default:
throw new IgniteException("Unexpected operation code: " + opCode);
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResource.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResource.java
new file mode 100644
index 0000000..6eaa6a8
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResource.java
@@ -0,0 +1,61 @@
+/*
+ * 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.client.handler;
+
+import org.apache.ignite.lang.IgniteException;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Client resource holder.
+ */
+public class ClientResource {
+ /** Object. */
+ private final Object obj;
+
+ /** Release delegate. */
+ private final @Nullable Runnable releaseRunnable;
+
+ public ClientResource(@NotNull Object obj, @Nullable Runnable releaseRunnable) {
+ this.obj = obj;
+ this.releaseRunnable = releaseRunnable;
+ }
+
+ /**
+ * Gets the underlying object.
+ *
+ * @param <T> Object type.
+ * @return Object.
+ */
+ public <T> T get(Class<T> type) {
+ if (!type.isInstance(obj)) {
+ throw new IgniteException("Incorrect resource type. Expected " + type.getName() + ", but got " + obj.getClass().getName());
+ }
+
+ return (T) obj;
+ }
+
+ /**
+ * Releases resources.
+ */
+ public void release() {
+ if (releaseRunnable != null) {
+ releaseRunnable.run();
+ }
+ }
+}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResourceRegistry.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResourceRegistry.java
new file mode 100644
index 0000000..9b4d16e
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResourceRegistry.java
@@ -0,0 +1,104 @@
+/*
+ * 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.client.handler;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.lang.IgniteException;
+
+/**
+ * Per-connection resource registry.
+ */
+public class ClientResourceRegistry {
+ /** Resources. */
+ private final Map<Long, ClientResource> res = new ConcurrentHashMap<>();
+
+ /** ID generator. */
+ private final AtomicLong idGen = new AtomicLong();
+
+ /**
+ * Stores the resource and returns the generated id.
+ *
+ * @param obj Object.
+ * @return Id.
+ */
+ public long put(ClientResource obj) {
+ long id = idGen.incrementAndGet();
+
+ res.put(id, obj);
+
+ return id;
+ }
+
+ /**
+ * Gets the resource by id.
+ *
+ * @param id Id.
+ * @return Object.
+ */
+ public ClientResource get(long id) {
+ ClientResource res = this.res.get(id);
+
+ if (res == null) {
+ throw new IgniteException("Failed to find resource with id: " + id);
+ }
+
+ return res;
+ }
+
+ /**
+ * Removes the resource.
+ *
+ * @param id Id.
+ */
+ public ClientResource remove(long id) {
+ ClientResource res = this.res.remove(id);
+
+ if (res == null) {
+ throw new IgniteException("Failed to find resource with id: " + id);
+ }
+
+ return res;
+ }
+
+ /**
+ * Releases all resources.
+ */
+ public void clean() {
+ IgniteException ex = null;
+
+ for (ClientResource r : res.values()) {
+ try {
+ r.release();
+ } catch (Exception e) {
+ if (ex == null) {
+ ex = new IgniteException(e);
+ } else {
+ ex.addSuppressed(e);
+ }
+ }
+ }
+
+ res.clear();
+
+ if (ex != null) {
+ throw ex;
+ }
+ }
+}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
index bee1462..91128b1 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
@@ -29,6 +29,7 @@ import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.UUID;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientDataType;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
@@ -45,7 +46,9 @@ import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
import org.msgpack.core.MessageTypeException;
/**
@@ -380,6 +383,21 @@ class ClientTableCommon {
}
}
+ /**
+ * Reads transaction.
+ *
+ * @param in Unpacker.
+ * @param resources Resource registry.
+ * @return Transaction, if present, or null.
+ */
+ public static @Nullable Transaction readTx(ClientMessageUnpacker in, ClientResourceRegistry resources) {
+ if (in.tryUnpackNil()) {
+ return null;
+ }
+
+ return resources.get(in.unpackLong()).get(Transaction.class);
+ }
+
private static void readAndSetColumnValue(ClientMessageUnpacker unpacker, Tuple tuple, Column col) {
try {
int type = getClientDataType(col.type().spec());
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java
index 84dbf44..6d66cd3 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -32,19 +34,22 @@ public class ClientTupleContainsKeyRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var keyTuple = readTuple(in, table, true);
- return table.recordView().getAsync(null, keyTuple).thenAccept(t -> out.packBoolean(t != null));
+ return table.recordView().getAsync(tx, keyTuple).thenAccept(t -> out.packBoolean(t != null));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java
index 76c6277..4cf5ccd 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java
@@ -19,9 +19,11 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuples;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -33,20 +35,23 @@ public class ClientTupleDeleteAllExactRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuples = readTuples(in, table, false);
- return table.recordView().deleteAllExactAsync(null, tuples)
+ return table.recordView().deleteAllExactAsync(tx, tuples)
.thenAccept(skippedTuples -> writeTuples(out, skippedTuples, table.schemaView(), true));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java
index b553b0e..186b370 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java
@@ -19,9 +19,11 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuples;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -34,20 +36,23 @@ public class ClientTupleDeleteAllRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuples = readTuples(in, table, true);
- return table.recordView().deleteAllAsync(null, tuples).thenAccept(skippedTuples ->
+ return table.recordView().deleteAllAsync(tx, tuples).thenAccept(skippedTuples ->
writeTuples(out, skippedTuples, TuplePart.KEY, table.schemaView(), true));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java
index 95f5772..a35a604 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -32,19 +34,22 @@ public class ClientTupleDeleteExactRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, false);
- return table.recordView().deleteExactAsync(null, tuple).thenAccept(out::packBoolean);
+ return table.recordView().deleteExactAsync(tx, tuple).thenAccept(out::packBoolean);
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
index 0953c69..4294df7 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -32,19 +34,22 @@ public class ClientTupleDeleteRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, true);
- return table.recordView().deleteAsync(null, tuple).thenAccept(out::packBoolean);
+ return table.recordView().deleteAsync(tx, tuple).thenAccept(out::packBoolean);
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java
index 4562942..b6d3d18 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java
@@ -19,9 +19,11 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuplesNullable;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -34,20 +36,23 @@ public class ClientTupleGetAllRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var keyTuples = readTuples(in, table, true);
- return table.recordView().getAllAsync(null, keyTuples).thenAccept(tuples ->
+ return table.recordView().getAllAsync(tx, keyTuples).thenAccept(tuples ->
writeTuplesNullable(out, tuples, TuplePart.KEY_AND_VAL, table.schemaView(), true));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java
index 9f29694..da35ff7 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -33,20 +35,23 @@ public class ClientTupleGetAndDeleteRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, true);
- return table.recordView().getAndDeleteAsync(null, tuple).thenAccept(
+ return table.recordView().getAndDeleteAsync(tx, tuple).thenAccept(
resTuple -> ClientTableCommon.writeTupleOrNil(out, resTuple, TuplePart.VAL));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java
index fa91936..0431934 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -33,20 +35,23 @@ public class ClientTupleGetAndReplaceRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, false);
- return table.recordView().getAndReplaceAsync(null, tuple).thenAccept(
+ return table.recordView().getAndReplaceAsync(tx, tuple).thenAccept(
resTuple -> ClientTableCommon.writeTupleOrNil(out, resTuple, TuplePart.VAL));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java
index 82388bd..f8d23ff 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -33,20 +35,23 @@ public class ClientTupleGetAndUpsertRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, false);
- return table.recordView().getAndUpsertAsync(null, tuple).thenAccept(
+ return table.recordView().getAndUpsertAsync(tx, tuple).thenAccept(
resTuple -> ClientTableCommon.writeTupleOrNil(out, resTuple, TuplePart.VAL));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
index 11a8b58..c6c1824 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -33,19 +35,22 @@ public class ClientTupleGetRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var keyTuple = readTuple(in, table, true);
- return table.recordView().getAsync(null, keyTuple).thenAccept(t -> ClientTableCommon.writeTupleOrNil(out, t, TuplePart.VAL));
+ return table.recordView().getAsync(tx, keyTuple).thenAccept(t -> ClientTableCommon.writeTupleOrNil(out, t, TuplePart.VAL));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java
index 6419dbc..3997174 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java
@@ -19,9 +19,11 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuples;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -33,20 +35,23 @@ public class ClientTupleInsertAllRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuples = readTuples(in, table, false);
- return table.recordView().insertAllAsync(null, tuples).thenAccept(skippedTuples ->
+ return table.recordView().insertAllAsync(tx, tuples).thenAccept(skippedTuples ->
writeTuples(out, skippedTuples, table.schemaView(), true));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java
index 1e0ebcf..73e211c 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -32,19 +34,22 @@ public class ClientTupleInsertRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, false);
- return table.recordView().insertAsync(null, tuple).thenAccept(out::packBoolean);
+ return table.recordView().insertAsync(tx, tuple).thenAccept(out::packBoolean);
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java
index d0d95f4..8c567b8 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java
@@ -20,8 +20,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readSchema;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -33,22 +35,25 @@ public class ClientTupleReplaceExactRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var schema = readSchema(in, table);
var oldTuple = readTuple(in, false, schema);
var newTuple = readTuple(in, false, schema);
- return table.recordView().replaceAsync(null, oldTuple, newTuple).thenAccept(out::packBoolean);
+ return table.recordView().replaceAsync(tx, oldTuple, newTuple).thenAccept(out::packBoolean);
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java
index 104e2e6..408ebc3 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -32,19 +34,22 @@ public class ClientTupleReplaceRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
+ IgniteTables tables,
+ ClientResourceRegistry resources
) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, false);
- return table.recordView().replaceAsync(null, tuple).thenAccept(out::packBoolean);
+ return table.recordView().replaceAsync(tx, tuple).thenAccept(out::packBoolean);
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java
index 9a9d66d..cebdd43 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -31,14 +33,20 @@ public class ClientTupleUpsertAllRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
- public static CompletableFuture<Void> process(ClientMessageUnpacker in, IgniteTables tables) {
+ public static CompletableFuture<Void> process(
+ ClientMessageUnpacker in,
+ IgniteTables tables,
+ ClientResourceRegistry resources
+ ) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuples = readTuples(in, table, false);
- return table.recordView().upsertAllAsync(null, tuples);
+ return table.recordView().upsertAllAsync(tx, tuples);
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
index 878da48..178cb9d 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
@@ -19,8 +19,10 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.table.manager.IgniteTables;
@@ -31,14 +33,20 @@ public class ClientTupleUpsertRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param tables Ignite tables.
+ * @param resources Resource registry.
* @return Future.
*/
- public static CompletableFuture<Void> process(ClientMessageUnpacker in, IgniteTables tables) {
+ public static CompletableFuture<Void> process(
+ ClientMessageUnpacker in,
+ IgniteTables tables,
+ ClientResourceRegistry resources
+ ) {
var table = readTable(in, tables);
+ var tx = readTx(in, resources);
var tuple = readTuple(in, table, false);
- return table.recordView().upsertAsync(null, tuple);
+ return table.recordView().upsertAsync(tx, tuple);
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionBeginRequest.java
similarity index 58%
copy from modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
copy to modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionBeginRequest.java
index 0953c69..dfb8bd2 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionBeginRequest.java
@@ -15,36 +15,30 @@
* limitations under the License.
*/
-package org.apache.ignite.client.handler.requests.table;
-
-import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
-import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+package org.apache.ignite.client.handler.requests.tx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResource;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
-import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
-import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.IgniteTransactions;
/**
- * Client tuple delete request.
+ * Client transaction begin request.
*/
-public class ClientTupleDeleteRequest {
+public class ClientTransactionBeginRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param tables Ignite tables.
+ * @param out Packer.
+ * @param transactions Transactions.
+ * @param resources Resources.
* @return Future.
*/
public static CompletableFuture<Void> process(
- ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteTables tables
- ) {
- var table = readTable(in, tables);
- var tuple = readTuple(in, table, true);
-
- return table.recordView().deleteAsync(null, tuple).thenAccept(out::packBoolean);
+ IgniteTransactions transactions,
+ ClientResourceRegistry resources) {
+ return transactions.beginAsync().thenAccept(t -> out.packLong(resources.put(new ClientResource(t, t::rollback))));
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
similarity index 63%
copy from modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
copy to modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
index 878da48..d64693d 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
@@ -15,30 +15,29 @@
* limitations under the License.
*/
-package org.apache.ignite.client.handler.requests.table;
-
-import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
-import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+package org.apache.ignite.client.handler.requests.tx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
-import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.Transaction;
/**
- * Client tuple upsert request.
+ * Client transaction commit request.
*/
-public class ClientTupleUpsertRequest {
+public class ClientTransactionCommitRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param resources Resources.
* @return Future.
*/
- public static CompletableFuture<Void> process(ClientMessageUnpacker in, IgniteTables tables) {
- var table = readTable(in, tables);
- var tuple = readTuple(in, table, false);
+ public static CompletableFuture<Void> process(ClientMessageUnpacker in, ClientResourceRegistry resources) {
+ long resourceId = in.unpackLong();
+
+ Transaction t = resources.remove(resourceId).get(Transaction.class);
- return table.recordView().upsertAsync(null, tuple);
+ return t.commitAsync();
}
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionRollbackRequest.java
similarity index 63%
copy from modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
copy to modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionRollbackRequest.java
index 878da48..5440c94 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionRollbackRequest.java
@@ -15,30 +15,29 @@
* limitations under the License.
*/
-package org.apache.ignite.client.handler.requests.table;
-
-import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTable;
-import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+package org.apache.ignite.client.handler.requests.tx;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
-import org.apache.ignite.table.manager.IgniteTables;
+import org.apache.ignite.tx.Transaction;
/**
- * Client tuple upsert request.
+ * Client transaction rollback request.
*/
-public class ClientTupleUpsertRequest {
+public class ClientTransactionRollbackRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param tables Ignite tables.
+ * @param in Unpacker.
+ * @param resources Resources.
* @return Future.
*/
- public static CompletableFuture<Void> process(ClientMessageUnpacker in, IgniteTables tables) {
- var table = readTable(in, tables);
- var tuple = readTuple(in, table, false);
+ public static CompletableFuture<Void> process(ClientMessageUnpacker in, ClientResourceRegistry resources) {
+ long resourceId = in.unpackLong();
+
+ Transaction t = resources.remove(resourceId).get(Transaction.class);
- return table.recordView().upsertAsync(null, tuple);
+ return t.rollbackAsync();
}
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/ClientChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/ClientChannel.java
index 6c06b5c..9c2e28e 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/ClientChannel.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/ClientChannel.java
@@ -25,7 +25,7 @@ import org.apache.ignite.client.IgniteClientException;
/**
* Processing thin client requests and responses.
*/
-interface ClientChannel extends AutoCloseable {
+public interface ClientChannel extends AutoCloseable {
/**
* Send request and handle response asynchronously for client operation.
*
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
index 2cc5afb..149d479 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
@@ -25,7 +25,6 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -444,7 +443,7 @@ public final class ReliableChannel implements AutoCloseable {
}
if (dfltChannelIdx == -1) {
- dfltChannelIdx = new Random().nextInt(reinitHolders.size());
+ dfltChannelIdx = 0;
}
curChannelsGuard.writeLock().lock();
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
index 8e77037..86f4d74 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
@@ -26,6 +26,7 @@ import org.apache.ignite.client.IgniteClientException;
import org.apache.ignite.client.proto.query.ClientMessage;
import org.apache.ignite.internal.client.io.ClientConnectionMultiplexer;
import org.apache.ignite.internal.client.table.ClientTables;
+import org.apache.ignite.internal.client.tx.ClientTransactions;
import org.apache.ignite.table.manager.IgniteTables;
import org.apache.ignite.tx.IgniteTransactions;
@@ -42,6 +43,9 @@ public class TcpIgniteClient implements IgniteClient {
/** Tables. */
private final ClientTables tables;
+ /** Transactions. */
+ private final ClientTransactions transactions;
+
/**
* Constructor.
*
@@ -68,6 +72,7 @@ public class TcpIgniteClient implements IgniteClient {
ch = new ReliableChannel(chFactory, cfg);
tables = new ClientTables(ch);
+ transactions = new ClientTransactions(ch);
}
/**
@@ -100,7 +105,7 @@ public class TcpIgniteClient implements IgniteClient {
/** {@inheritDoc} */
@Override
public IgniteTransactions transactions() {
- return null;
+ return transactions;
}
/**
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
index 366810b..4b19ec5 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
@@ -39,6 +39,9 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
/** Underlying table. */
private final ClientTable tbl;
+ /** Tuple serializer. */
+ private final ClientTupleSerializer ser;
+
/**
* Constructor.
*
@@ -48,6 +51,7 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
assert tbl != null;
this.tbl = tbl;
+ ser = new ClientTupleSerializer(tbl.tableId());
}
/** {@inheritDoc} */
@@ -60,11 +64,11 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Tuple> getAsync(@Nullable Transaction tx, @NotNull Tuple key) {
Objects.requireNonNull(key);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (schema, out) -> tbl.writeTuple(key, schema, out, true),
- ClientTable::readValueTuple);
+ (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ ClientTupleSerializer::readValueTuple);
}
/** {@inheritDoc} */
@@ -77,11 +81,11 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Map<Tuple, Tuple>> getAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> keys) {
Objects.requireNonNull(keys);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (s, w) -> tbl.writeTuples(keys, s, w, true),
- tbl::readKvTuplesNullable,
+ (s, w) -> ser.writeTuples(tx, keys, s, w, true),
+ ClientTupleSerializer::readKvTuplesNullable,
Collections.emptyMap());
}
@@ -95,10 +99,10 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public CompletableFuture<Boolean> containsAsync(@Nullable Transaction tx, @NotNull Tuple key) {
Objects.requireNonNull(key);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_KEY,
- (schema, out) -> tbl.writeTuple(key, schema, out, true),
+ (s, w) -> ser.writeTuple(tx, key, s, w, true),
ClientMessageUnpacker::unpackBoolean);
}
@@ -112,12 +116,12 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Void> putAsync(@Nullable Transaction tx, @NotNull Tuple key, Tuple val) {
Objects.requireNonNull(key);
- // TODO: Transactions IGNITE-15240
+
// TODO IGNITE-15194: Convert Tuple to a schema-order Array as a first step.
// If it does not match the latest schema, then request latest and convert again.
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> tbl.writeKvTuple(key, val, s, w, false),
+ (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
r -> null);
}
@@ -131,10 +135,10 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Void> putAllAsync(@Nullable Transaction tx, @NotNull Map<Tuple, Tuple> pairs) {
Objects.requireNonNull(pairs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
- (s, w) -> tbl.writeKvTuples(pairs, s, w),
+ (s, w) -> ser.writeKvTuples(tx, pairs, s, w),
r -> null);
}
@@ -147,11 +151,12 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Tuple> getAndPutAsync(@Nullable Transaction tx, @NotNull Tuple key, Tuple val) {
- // TODO: Transactions IGNITE-15240
+ Objects.requireNonNull(key);
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> tbl.writeKvTuple(key, val, s, w, false),
- ClientTable::readValueTuple);
+ (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ ClientTupleSerializer::readValueTuple);
}
/** {@inheritDoc} */
@@ -163,10 +168,11 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Boolean> putIfAbsentAsync(@Nullable Transaction tx, @NotNull Tuple key, Tuple val) {
- // TODO: Transactions IGNITE-15240
+ Objects.requireNonNull(key);
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> tbl.writeKvTuple(key, val, s, w, false),
+ (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
ClientMessageUnpacker::unpackBoolean);
}
@@ -186,10 +192,10 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Boolean> removeAsync(@Nullable Transaction tx, @NotNull Tuple key) {
Objects.requireNonNull(key);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> tbl.writeTuple(key, s, w, true),
+ (s, w) -> ser.writeTuple(tx, key, s, w, true),
ClientMessageUnpacker::unpackBoolean);
}
@@ -198,10 +204,10 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
public @NotNull CompletableFuture<Boolean> removeAsync(@Nullable Transaction tx, @NotNull Tuple key, @NotNull Tuple val) {
Objects.requireNonNull(key);
Objects.requireNonNull(val);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> tbl.writeKvTuple(key, val, s, w, false),
+ (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
ClientMessageUnpacker::unpackBoolean);
}
@@ -215,11 +221,11 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Collection<Tuple>> removeAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> keys) {
Objects.requireNonNull(keys);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (s, w) -> tbl.writeTuples(keys, s, w, true),
- (schema, in) -> tbl.readTuples(schema, in, true),
+ (s, w) -> ser.writeTuples(tx, keys, s, w, true),
+ (s, r) -> ClientTupleSerializer.readTuples(s, r, true),
Collections.emptyList());
}
@@ -232,11 +238,12 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
/** {@inheritDoc} */
@Override
public @NotNull CompletableFuture<Tuple> getAndRemoveAsync(@Nullable Transaction tx, @NotNull Tuple key) {
- // TODO: Transactions IGNITE-15240
+ Objects.requireNonNull(key);
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> tbl.writeTuple(key, s, w, true),
- ClientTable::readValueTuple);
+ (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ ClientTupleSerializer::readValueTuple);
}
/** {@inheritDoc} */
@@ -255,10 +262,10 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull Tuple key, Tuple val) {
Objects.requireNonNull(key);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> tbl.writeKvTuple(key, val, s, w, false),
+ (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
ClientMessageUnpacker::unpackBoolean);
}
@@ -266,12 +273,12 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull Tuple key, Tuple oldVal, Tuple newVal) {
Objects.requireNonNull(key);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
(s, w) -> {
- tbl.writeKvTuple(key, oldVal, s, w, false);
- tbl.writeKvTuple(key, newVal, s, w, true);
+ ser.writeKvTuple(tx, key, oldVal, s, w, false);
+ ser.writeKvTuple(tx, key, newVal, s, w, true);
},
ClientMessageUnpacker::unpackBoolean);
}
@@ -286,11 +293,11 @@ public class ClientKeyValueBinaryView implements KeyValueView<Tuple, Tuple> {
@Override
public @NotNull CompletableFuture<Tuple> getAndReplaceAsync(@Nullable Transaction tx, @NotNull Tuple key, Tuple val) {
Objects.requireNonNull(key);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> tbl.writeKvTuple(key, val, s, w, false),
- ClientTable::readValueTuple);
+ (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ ClientTupleSerializer::readValueTuple);
}
/** {@inheritDoc} */
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
index 7f0ddc0..8e4fc3b 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.client.table;
+import static org.apache.ignite.internal.client.table.ClientTable.writeTx;
+
import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
@@ -27,7 +29,7 @@ import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.IgniteClientException;
-import org.apache.ignite.internal.client.proto.ClientMessagePacker;
+import org.apache.ignite.internal.client.PayloadOutputChannel;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.ClientOp;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -82,12 +84,11 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<V> getAsync(@Nullable Transaction tx, @NotNull K key) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (schema, out) -> keySer.writeRec(key, schema, out, TuplePart.KEY),
- (inSchema, in) -> valSer.readRec(inSchema, in, TuplePart.VAL));
+ (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, r) -> valSer.readRec(s, r, TuplePart.VAL));
}
/** {@inheritDoc} */
@@ -100,7 +101,6 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Map<K, V>> getAllAsync(@Nullable Transaction tx, @NotNull Collection<K> keys) {
Objects.requireNonNull(keys);
- throwUnsupportedIfTxNotNull(tx);
if (keys.isEmpty()) {
return CompletableFuture.completedFuture(Collections.emptyMap());
@@ -108,7 +108,7 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (schema, out) -> keySer.writeRecs(keys, schema, out, TuplePart.KEY),
+ (s, w) -> keySer.writeRecs(tx, keys, s, w, TuplePart.KEY),
this::readGetAllResponse,
Collections.emptyMap());
}
@@ -123,11 +123,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public CompletableFuture<Boolean> containsAsync(@Nullable Transaction tx, @NotNull K key) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_KEY,
- (schema, out) -> keySer.writeRec(key, schema, out, TuplePart.KEY),
+ (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
ClientMessageUnpacker::unpackBoolean);
}
@@ -141,11 +140,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Void> putAsync(@Nullable Transaction tx, @NotNull K key, V val) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> writeKeyValue(s, w, key, val),
+ (s, w) -> writeKeyValue(s, w, tx, key, val),
r -> null);
}
@@ -159,7 +157,6 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Void> putAllAsync(@Nullable Transaction tx, @NotNull Map<K, V> pairs) {
Objects.requireNonNull(pairs);
- throwUnsupportedIfTxNotNull(tx);
if (pairs.isEmpty()) {
return CompletableFuture.completedFuture(null);
@@ -168,13 +165,14 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
(s, w) -> {
- w.packIgniteUuid(tbl.tableId());
- w.packInt(s.version());
- w.packInt(pairs.size());
+ w.out().packIgniteUuid(tbl.tableId());
+ writeTx(tx, w);
+ w.out().packInt(s.version());
+ w.out().packInt(pairs.size());
for (Entry<K, V> e : pairs.entrySet()) {
- keySer.writeRecRaw(e.getKey(), s, w, TuplePart.KEY);
- valSer.writeRecRaw(e.getValue(), s, w, TuplePart.VAL);
+ keySer.writeRecRaw(e.getKey(), s, w.out(), TuplePart.KEY);
+ valSer.writeRecRaw(e.getValue(), s, w.out(), TuplePart.VAL);
}
},
r -> null);
@@ -190,11 +188,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<V> getAndPutAsync(@Nullable Transaction tx, @NotNull K key, V val) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> writeKeyValue(s, w, key, val),
+ (s, w) -> writeKeyValue(s, w, tx, key, val),
(s, r) -> valSer.readRec(s, r, TuplePart.VAL));
}
@@ -208,11 +205,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Boolean> putIfAbsentAsync(@Nullable Transaction tx, @NotNull K key, V val) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> writeKeyValue(s, w, key, val),
+ (s, w) -> writeKeyValue(s, w, tx, key, val),
ClientMessageUnpacker::unpackBoolean);
}
@@ -232,11 +228,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Boolean> removeAsync(@Nullable Transaction tx, @NotNull K key) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> keySer.writeRec(key, s, w, TuplePart.KEY),
+ (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
ClientMessageUnpacker::unpackBoolean);
}
@@ -244,11 +239,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Boolean> removeAsync(@Nullable Transaction tx, @NotNull K key, V val) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> writeKeyValue(s, w, key, val),
+ (s, w) -> writeKeyValue(s, w, tx, key, val),
ClientMessageUnpacker::unpackBoolean);
}
@@ -262,7 +256,6 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Collection<K>> removeAllAsync(@Nullable Transaction tx, @NotNull Collection<K> keys) {
Objects.requireNonNull(keys);
- throwUnsupportedIfTxNotNull(tx);
if (keys.isEmpty()) {
return CompletableFuture.completedFuture(Collections.emptyList());
@@ -270,8 +263,8 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (schema, out) -> keySer.writeRecs(keys, schema, out, TuplePart.KEY),
- (inSchema, in) -> keySer.readRecs(inSchema, in, false, TuplePart.KEY),
+ (s, w) -> keySer.writeRecs(tx, keys, s, w, TuplePart.KEY),
+ (s, r) -> keySer.readRecs(s, r, false, TuplePart.KEY),
Collections.emptyList());
}
@@ -285,11 +278,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<V> getAndRemoveAsync(@Nullable Transaction tx, @NotNull K key) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> keySer.writeRec(key, s, w, TuplePart.KEY),
+ (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
(s, r) -> valSer.readRec(s, r, TuplePart.VAL));
}
@@ -303,7 +295,6 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public boolean replace(@Nullable Transaction tx, @NotNull K key, V oldVal, V newVal) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return replaceAsync(tx, key, oldVal, newVal).join();
}
@@ -312,11 +303,10 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull K key, V val) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> writeKeyValue(s, w, key, val),
+ (s, w) -> writeKeyValue(s, w, tx, key, val),
ClientMessageUnpacker::unpackBoolean);
}
@@ -324,16 +314,15 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull K key, V oldVal, V newVal) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
(s, w) -> {
- keySer.writeRec(key, s, w, TuplePart.KEY);
- valSer.writeRecRaw(oldVal, s, w, TuplePart.VAL);
+ keySer.writeRec(tx, key, s, w, TuplePart.KEY);
+ valSer.writeRecRaw(oldVal, s, w.out(), TuplePart.VAL);
- keySer.writeRecRaw(key, s, w, TuplePart.KEY);
- valSer.writeRecRaw(newVal, s, w, TuplePart.VAL);
+ keySer.writeRecRaw(key, s, w.out(), TuplePart.KEY);
+ valSer.writeRecRaw(newVal, s, w.out(), TuplePart.VAL);
},
ClientMessageUnpacker::unpackBoolean);
}
@@ -348,12 +337,11 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
@Override
public @NotNull CompletableFuture<V> getAndReplaceAsync(@Nullable Transaction tx, @NotNull K key, V val) {
Objects.requireNonNull(key);
- throwUnsupportedIfTxNotNull(tx);
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> writeKeyValue(s, w, key, val),
- (inSchema, in) -> valSer.readRec(inSchema, in, TuplePart.VAL));
+ (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, r) -> valSer.readRec(s, r, TuplePart.VAL));
}
/** {@inheritDoc} */
@@ -400,9 +388,9 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
throw new UnsupportedOperationException("Not implemented yet.");
}
- private void writeKeyValue(ClientSchema s, ClientMessagePacker w, @NotNull K key, V val) {
- keySer.writeRec(key, s, w, TuplePart.KEY);
- valSer.writeRecRaw(val, s, w, TuplePart.VAL);
+ private void writeKeyValue(ClientSchema s, PayloadOutputChannel w, @Nullable Transaction tx, @NotNull K key, V val) {
+ keySer.writeRec(tx, key, s, w, TuplePart.KEY);
+ valSer.writeRecRaw(val, s, w.out(), TuplePart.VAL);
}
private HashMap<K, V> readGetAllResponse(ClientSchema schema, ClientMessageUnpacker in) {
@@ -426,11 +414,4 @@ public class ClientKeyValueView<K, V> implements KeyValueView<K, V> {
throw new IgniteClientException(e.getMessage(), e);
}
}
-
- private static void throwUnsupportedIfTxNotNull(Transaction tx) {
- if (tx != null) {
- // TODO: IGNITE-15240.
- throw new UnsupportedOperationException("Not implemented yet.");
- }
- }
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
index e77b01c..a1b1c55 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
@@ -39,6 +39,9 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
/** Underlying table. */
private final ClientTable tbl;
+ /** Tuple serializer. */
+ private final ClientTupleSerializer ser;
+
/**
* Constructor.
*
@@ -48,6 +51,7 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
assert tbl != null;
this.tbl = tbl;
+ ser = new ClientTupleSerializer(tbl.tableId());
}
/** {@inheritDoc} */
@@ -63,8 +67,8 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (schema, out) -> tbl.writeTuple(keyRec, schema, out, true),
- (inSchema, in) -> ClientTable.readValueTuple(inSchema, in, keyRec));
+ (s, w) -> ser.writeTuple(tx, keyRec, s, w, true),
+ (s, r) -> ClientTupleSerializer.readValueTuple(s, r, keyRec));
}
/** {@inheritDoc} */
@@ -77,11 +81,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Collection<Tuple>> getAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> keyRecs) {
Objects.requireNonNull(keyRecs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (s, w) -> tbl.writeTuples(keyRecs, s, w, true),
- tbl::readTuplesNullable,
+ (s, w) -> ser.writeTuples(tx, keyRecs, s, w, true),
+ ClientTupleSerializer::readTuplesNullable,
Collections.emptyList());
}
@@ -95,12 +99,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Void> upsertAsync(@Nullable Transaction tx, @NotNull Tuple rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
// TODO IGNITE-15194: Convert Tuple to a schema-order Array as a first step.
// If it does not match the latest schema, then request latest and convert again.
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> tbl.writeTuple(rec, s, w),
+ (s, w) -> ser.writeTuple(tx, rec, s, w),
r -> null);
}
@@ -114,10 +117,10 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Void> upsertAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> recs) {
Objects.requireNonNull(recs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
- (s, w) -> tbl.writeTuples(recs, s, w, false),
+ (s, w) -> ser.writeTuples(tx, recs, s, w, false),
r -> null);
}
@@ -131,11 +134,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Tuple> getAndUpsertAsync(@Nullable Transaction tx, @NotNull Tuple rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> tbl.writeTuple(rec, s, w, false),
- (schema, in) -> ClientTable.readValueTuple(schema, in, rec));
+ (s, w) -> ser.writeTuple(tx, rec, s, w, false),
+ (s, r) -> ClientTupleSerializer.readValueTuple(s, r, rec));
}
/** {@inheritDoc} */
@@ -148,10 +151,10 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Boolean> insertAsync(@Nullable Transaction tx, @NotNull Tuple rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> tbl.writeTuple(rec, s, w, false),
+ (s, w) -> ser.writeTuple(tx, rec, s, w, false),
ClientMessageUnpacker::unpackBoolean);
}
@@ -165,11 +168,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Collection<Tuple>> insertAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> recs) {
Objects.requireNonNull(recs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_INSERT_ALL,
- (s, w) -> tbl.writeTuples(recs, s, w, false),
- tbl::readTuples,
+ (s, w) -> ser.writeTuples(tx, recs, s, w, false),
+ ClientTupleSerializer::readTuples,
Collections.emptyList());
}
@@ -189,10 +192,10 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull Tuple rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> tbl.writeTuple(rec, s, w, false),
+ (s, w) -> ser.writeTuple(tx, rec, s, w, false),
ClientMessageUnpacker::unpackBoolean);
}
@@ -201,12 +204,12 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull Tuple oldRec, @NotNull Tuple newRec) {
Objects.requireNonNull(oldRec);
Objects.requireNonNull(newRec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
(s, w) -> {
- tbl.writeTuple(oldRec, s, w, false, false);
- tbl.writeTuple(newRec, s, w, false, true);
+ ser.writeTuple(tx, oldRec, s, w, false, false);
+ ser.writeTuple(tx, newRec, s, w, false, true);
},
ClientMessageUnpacker::unpackBoolean);
}
@@ -221,11 +224,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Tuple> getAndReplaceAsync(@Nullable Transaction tx, @NotNull Tuple rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> tbl.writeTuple(rec, s, w, false),
- (schema, in) -> ClientTable.readValueTuple(schema, in, rec));
+ (s, w) -> ser.writeTuple(tx, rec, s, w, false),
+ (s, r) -> ClientTupleSerializer.readValueTuple(s, r, rec));
}
/** {@inheritDoc} */
@@ -238,10 +241,10 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Boolean> deleteAsync(@Nullable Transaction tx, @NotNull Tuple keyRec) {
Objects.requireNonNull(keyRec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> tbl.writeTuple(keyRec, s, w, true),
+ (s, w) -> ser.writeTuple(tx, keyRec, s, w, true),
ClientMessageUnpacker::unpackBoolean);
}
@@ -255,10 +258,10 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Boolean> deleteExactAsync(@Nullable Transaction tx, @NotNull Tuple rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> tbl.writeTuple(rec, s, w, false),
+ (s, w) -> ser.writeTuple(tx, rec, s, w, false),
ClientMessageUnpacker::unpackBoolean);
}
@@ -272,11 +275,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Tuple> getAndDeleteAsync(@Nullable Transaction tx, @NotNull Tuple keyRec) {
Objects.requireNonNull(keyRec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> tbl.writeTuple(keyRec, s, w, true),
- (schema, in) -> ClientTable.readValueTuple(schema, in, keyRec));
+ (s, w) -> ser.writeTuple(tx, keyRec, s, w, true),
+ (s, r) -> ClientTupleSerializer.readValueTuple(s, r, keyRec));
}
/** {@inheritDoc} */
@@ -289,11 +292,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Collection<Tuple>> deleteAllAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> keyRecs) {
Objects.requireNonNull(keyRecs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (s, w) -> tbl.writeTuples(keyRecs, s, w, true),
- (schema, in) -> tbl.readTuples(schema, in, true),
+ (s, w) -> ser.writeTuples(tx, keyRecs, s, w, true),
+ (s, r) -> ClientTupleSerializer.readTuples(s, r, true),
Collections.emptyList());
}
@@ -307,11 +310,11 @@ public class ClientRecordBinaryView implements RecordView<Tuple> {
@Override
public @NotNull CompletableFuture<Collection<Tuple>> deleteAllExactAsync(@Nullable Transaction tx, @NotNull Collection<Tuple> recs) {
Objects.requireNonNull(recs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL_EXACT,
- (s, w) -> tbl.writeTuples(recs, s, w, false),
- tbl::readTuples,
+ (s, w) -> ser.writeTuples(tx, recs, s, w, false),
+ ClientTupleSerializer::readTuples,
Collections.emptyList());
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
index ca4645e..72e8fcc 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
@@ -17,10 +17,13 @@
package org.apache.ignite.internal.client.table;
+import static org.apache.ignite.internal.client.table.ClientTable.writeTx;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import org.apache.ignite.client.IgniteClientException;
+import org.apache.ignite.internal.client.PayloadOutputChannel;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -31,6 +34,7 @@ import org.apache.ignite.internal.marshaller.MarshallerException;
import org.apache.ignite.internal.marshaller.MarshallerUtil;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.table.mapper.Mapper;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -67,11 +71,12 @@ class ClientRecordSerializer<R> {
return mapper;
}
- public void writeRec(@Nullable R rec, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
- out.packIgniteUuid(tableId);
- out.packInt(schema.version());
+ public void writeRec(@Nullable Transaction tx, @Nullable R rec, ClientSchema schema, PayloadOutputChannel out, TuplePart part) {
+ out.out().packIgniteUuid(tableId);
+ writeTx(tx, out);
+ out.out().packInt(schema.version());
- writeRecRaw(rec, schema, out, part);
+ writeRecRaw(rec, schema, out.out(), part);
}
public void writeRecRaw(@Nullable R rec, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
@@ -85,12 +90,20 @@ class ClientRecordSerializer<R> {
}
}
- public void writeRecs(@Nullable R rec, @Nullable R rec2, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
- out.packIgniteUuid(tableId);
- out.packInt(schema.version());
+ public void writeRecs(
+ @Nullable Transaction tx,
+ @Nullable R rec,
+ @Nullable R rec2,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ TuplePart part
+ ) {
+ out.out().packIgniteUuid(tableId);
+ writeTx(tx, out);
+ out.out().packInt(schema.version());
Marshaller marshaller = schema.getMarshaller(mapper, part);
- ClientMarshallerWriter writer = new ClientMarshallerWriter(out);
+ ClientMarshallerWriter writer = new ClientMarshallerWriter(out.out());
try {
marshaller.writeObject(rec, writer);
@@ -100,13 +113,20 @@ class ClientRecordSerializer<R> {
}
}
- public void writeRecs(@NotNull Collection<R> recs, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
- out.packIgniteUuid(tableId);
- out.packInt(schema.version());
- out.packInt(recs.size());
+ public void writeRecs(
+ @Nullable Transaction tx,
+ @NotNull Collection<R> recs,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ TuplePart part
+ ) {
+ out.out().packIgniteUuid(tableId);
+ writeTx(tx, out);
+ out.out().packInt(schema.version());
+ out.out().packInt(recs.size());
Marshaller marshaller = schema.getMarshaller(mapper, part);
- ClientMarshallerWriter writer = new ClientMarshallerWriter(out);
+ ClientMarshallerWriter writer = new ClientMarshallerWriter(out.out());
try {
for (R rec : recs) {
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
index c228f5c..ff590fa 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
@@ -18,22 +18,14 @@
package org.apache.ignite.internal.client.table;
import java.io.Serializable;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.client.IgniteClientException;
-import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.ClientOp;
import org.apache.ignite.internal.client.proto.TuplePart;
-import org.apache.ignite.internal.marshaller.ClientMarshallerReader;
-import org.apache.ignite.internal.marshaller.ClientMarshallerWriter;
-import org.apache.ignite.internal.marshaller.Marshaller;
-import org.apache.ignite.internal.marshaller.MarshallerException;
-import org.apache.ignite.internal.marshaller.MarshallerUtil;
import org.apache.ignite.table.InvokeProcessor;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.mapper.Mapper;
@@ -45,14 +37,11 @@ import org.jetbrains.annotations.Nullable;
* Client record view implementation.
*/
public class ClientRecordView<R> implements RecordView<R> {
- /** Mapper. */
- private final Mapper<R> recMapper;
-
/** Underlying table. */
private final ClientTable tbl;
- /** Simple mapping mode: single column maps to a basic type. For example, {@code RecordView<String>}. */
- private final boolean oneColumnMode;
+ /** Serializer. */
+ private final ClientRecordSerializer<R> ser;
/**
* Constructor.
@@ -62,9 +51,7 @@ public class ClientRecordView<R> implements RecordView<R> {
*/
public ClientRecordView(ClientTable tbl, Mapper<R> recMapper) {
this.tbl = tbl;
- this.recMapper = recMapper;
-
- oneColumnMode = MarshallerUtil.mode(recMapper.targetType()) != null;
+ ser = new ClientRecordSerializer<>(tbl.tableId(), recMapper);
}
/** {@inheritDoc} */
@@ -77,11 +64,11 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<R> getAsync(@Nullable Transaction tx, @NotNull R keyRec) {
Objects.requireNonNull(keyRec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (schema, out) -> writeRec(keyRec, schema, out, TuplePart.KEY),
- (inSchema, in) -> readValRec(keyRec, inSchema, in));
+ (s, w) -> ser.writeRec(tx, keyRec, s, w, TuplePart.KEY),
+ (s, r) -> ser.readValRec(keyRec, s, r));
}
/** {@inheritDoc} */
@@ -94,11 +81,11 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Collection<R>> getAllAsync(@Nullable Transaction tx, @NotNull Collection<R> keyRecs) {
Objects.requireNonNull(keyRecs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (schema, out) -> writeRecs(keyRecs, schema, out, TuplePart.KEY),
- this::readRecsNullable,
+ (s, w) -> ser.writeRecs(tx, keyRecs, s, w, TuplePart.KEY),
+ (s, r) -> ser.readRecs(s, r, true, TuplePart.KEY_AND_VAL),
Collections.emptyList());
}
@@ -112,10 +99,10 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Void> upsertAsync(@Nullable Transaction tx, @NotNull R rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> writeRec(rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
r -> null);
}
@@ -129,10 +116,10 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Void> upsertAllAsync(@Nullable Transaction tx, @NotNull Collection<R> recs) {
Objects.requireNonNull(recs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
- (s, w) -> writeRecs(recs, s, w, TuplePart.KEY_AND_VAL),
+ (s, w) -> ser.writeRecs(tx, recs, s, w, TuplePart.KEY_AND_VAL),
r -> null);
}
@@ -146,11 +133,11 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<R> getAndUpsertAsync(@Nullable Transaction tx, @NotNull R rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> writeRec(rec, s, w, TuplePart.KEY_AND_VAL),
- (s, r) -> readValRec(rec, s, r));
+ (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, r) -> ser.readValRec(rec, s, r));
}
/** {@inheritDoc} */
@@ -163,10 +150,10 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Boolean> insertAsync(@Nullable Transaction tx, @NotNull R rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> writeRec(rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
ClientMessageUnpacker::unpackBoolean);
}
@@ -180,11 +167,11 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Collection<R>> insertAllAsync(@Nullable Transaction tx, @NotNull Collection<R> recs) {
Objects.requireNonNull(recs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_INSERT_ALL,
- (s, w) -> writeRecs(recs, s, w, TuplePart.KEY_AND_VAL),
- this::readRecs,
+ (s, w) -> ser.writeRecs(tx, recs, s, w, TuplePart.KEY_AND_VAL),
+ (s, r) -> ser.readRecs(s, r, false, TuplePart.KEY_AND_VAL),
Collections.emptyList());
}
@@ -204,10 +191,10 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull R rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> writeRec(rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
ClientMessageUnpacker::unpackBoolean);
}
@@ -216,10 +203,10 @@ public class ClientRecordView<R> implements RecordView<R> {
public @NotNull CompletableFuture<Boolean> replaceAsync(@Nullable Transaction tx, @NotNull R oldRec, @NotNull R newRec) {
Objects.requireNonNull(oldRec);
Objects.requireNonNull(newRec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
- (s, w) -> writeRecs(oldRec, newRec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w) -> ser.writeRecs(tx, oldRec, newRec, s, w, TuplePart.KEY_AND_VAL),
ClientMessageUnpacker::unpackBoolean);
}
@@ -233,11 +220,11 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<R> getAndReplaceAsync(@Nullable Transaction tx, @NotNull R rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> writeRec(rec, s, w, TuplePart.KEY_AND_VAL),
- (s, r) -> readValRec(rec, s, r));
+ (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, r) -> ser.readValRec(rec, s, r));
}
/** {@inheritDoc} */
@@ -250,10 +237,10 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Boolean> deleteAsync(@Nullable Transaction tx, @NotNull R keyRec) {
Objects.requireNonNull(keyRec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> writeRec(keyRec, s, w, TuplePart.KEY),
+ (s, w) -> ser.writeRec(tx, keyRec, s, w, TuplePart.KEY),
ClientMessageUnpacker::unpackBoolean);
}
@@ -267,10 +254,10 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Boolean> deleteExactAsync(@Nullable Transaction tx, @NotNull R rec) {
Objects.requireNonNull(rec);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> writeRec(rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
ClientMessageUnpacker::unpackBoolean);
}
@@ -287,8 +274,8 @@ public class ClientRecordView<R> implements RecordView<R> {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> writeRec(keyRec, s, w, TuplePart.KEY),
- (s, r) -> readValRec(keyRec, s, r));
+ (s, w) -> ser.writeRec(tx, keyRec, s, w, TuplePart.KEY),
+ (s, r) -> ser.readValRec(keyRec, s, r));
}
/** {@inheritDoc} */
@@ -301,11 +288,11 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Collection<R>> deleteAllAsync(@Nullable Transaction tx, @NotNull Collection<R> keyRecs) {
Objects.requireNonNull(keyRecs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (s, w) -> writeRecs(keyRecs, s, w, TuplePart.KEY),
- (schema, in) -> readRecs(schema, in, false, TuplePart.KEY),
+ (s, w) -> ser.writeRecs(tx, keyRecs, s, w, TuplePart.KEY),
+ (s, r) -> ser.readRecs(s, r, false, TuplePart.KEY),
Collections.emptyList());
}
@@ -319,11 +306,11 @@ public class ClientRecordView<R> implements RecordView<R> {
@Override
public @NotNull CompletableFuture<Collection<R>> deleteAllExactAsync(@Nullable Transaction tx, @NotNull Collection<R> recs) {
Objects.requireNonNull(recs);
- // TODO: Transactions IGNITE-15240
+
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL_EXACT,
- (s, w) -> writeRecs(recs, s, w, TuplePart.KEY_AND_VAL),
- this::readRecs,
+ (s, w) -> ser.writeRecs(tx, recs, s, w, TuplePart.KEY_AND_VAL),
+ (s, r) -> ser.readRecs(s, r, false, TuplePart.KEY_AND_VAL),
Collections.emptyList());
}
@@ -362,105 +349,4 @@ public class ClientRecordView<R> implements RecordView<R> {
) {
throw new UnsupportedOperationException();
}
-
- private void writeRec(@NotNull R rec, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
- out.packIgniteUuid(tbl.tableId());
- out.packInt(schema.version());
-
- Marshaller marshaller = schema.getMarshaller(recMapper, part);
- ClientMarshallerWriter writer = new ClientMarshallerWriter(out);
-
- try {
- marshaller.writeObject(rec, writer);
- } catch (MarshallerException e) {
- throw new IgniteClientException(e.getMessage(), e);
- }
- }
-
- private void writeRecs(@NotNull R rec, @NotNull R rec2, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
- out.packIgniteUuid(tbl.tableId());
- out.packInt(schema.version());
-
- Marshaller marshaller = schema.getMarshaller(recMapper, part);
- ClientMarshallerWriter writer = new ClientMarshallerWriter(out);
-
- try {
- marshaller.writeObject(rec, writer);
- marshaller.writeObject(rec2, writer);
- } catch (MarshallerException e) {
- throw new IgniteClientException(e.getMessage(), e);
- }
- }
-
- private void writeRecs(@NotNull Collection<R> recs, ClientSchema schema, ClientMessagePacker out, TuplePart part) {
- out.packIgniteUuid(tbl.tableId());
- out.packInt(schema.version());
- out.packInt(recs.size());
-
- Marshaller marshaller = schema.getMarshaller(recMapper, part);
- ClientMarshallerWriter writer = new ClientMarshallerWriter(out);
-
- try {
- for (R rec : recs) {
- marshaller.writeObject(rec, writer);
- }
- } catch (MarshallerException e) {
- throw new IgniteClientException(e.getMessage(), e);
- }
- }
-
- private Collection<R> readRecsNullable(ClientSchema schema, ClientMessageUnpacker in) {
- return readRecs(schema, in, true, TuplePart.KEY_AND_VAL);
- }
-
- private Collection<R> readRecs(ClientSchema schema, ClientMessageUnpacker in) {
- return readRecs(schema, in, false, TuplePart.KEY_AND_VAL);
- }
-
- private Collection<R> readRecs(ClientSchema schema, ClientMessageUnpacker in, boolean nullable, TuplePart part) {
- var cnt = in.unpackInt();
- var res = new ArrayList<R>(cnt);
-
- if (cnt == 0) {
- return res;
- }
-
- Marshaller marshaller = schema.getMarshaller(recMapper, part);
- var reader = new ClientMarshallerReader(in);
-
- try {
- for (int i = 0; i < cnt; i++) {
- if (nullable && !in.unpackBoolean()) {
- res.add(null);
- } else {
- res.add((R) marshaller.readObject(reader, null));
- }
- }
- } catch (MarshallerException e) {
- throw new IgniteClientException(e.getMessage(), e);
- }
-
- return res;
- }
-
- private R readValRec(@NotNull R keyRec, ClientSchema schema, ClientMessageUnpacker in) {
- if (oneColumnMode) {
- return keyRec;
- }
-
- Marshaller keyMarshaller = schema.getMarshaller(recMapper, TuplePart.KEY);
- Marshaller valMarshaller = schema.getMarshaller(recMapper, TuplePart.VAL);
-
- ClientMarshallerReader reader = new ClientMarshallerReader(in);
-
- try {
- var res = (R) valMarshaller.readObject(reader, null);
-
- keyMarshaller.copyObject(keyRec, res);
-
- return res;
- } catch (MarshallerException e) {
- throw new IgniteClientException(e.getMessage(), e);
- }
- }
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
index 84ef403..84c56dc 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
@@ -17,12 +17,6 @@
package org.apache.ignite.internal.client.table;
-import static org.apache.ignite.internal.client.proto.ClientMessageCommon.NO_VALUE;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
@@ -31,10 +25,11 @@ import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import org.apache.ignite.client.IgniteClientException;
+import org.apache.ignite.internal.client.PayloadOutputChannel;
import org.apache.ignite.internal.client.ReliableChannel;
-import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.ClientOp;
+import org.apache.ignite.internal.client.tx.ClientTransaction;
import org.apache.ignite.internal.tostring.IgniteToStringBuilder;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteUuid;
@@ -43,6 +38,7 @@ import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.Table;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.mapper.Mapper;
+import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -210,268 +206,38 @@ public class ClientTable implements Table {
}
/**
- * Writes {@link Tuple}.
- *
- * @param tuple Tuple.
- * @param schema Schema.
- * @param out Out.
- */
- public void writeTuple(
- @NotNull Tuple tuple,
- ClientSchema schema,
- ClientMessagePacker out
- ) {
- writeTuple(tuple, schema, out, false, false);
- }
-
- /**
- * Writes {@link Tuple}.
- *
- * @param tuple Tuple.
- * @param schema Schema.
- * @param out Out.
- * @param keyOnly Key only.
- */
- public void writeTuple(
- @NotNull Tuple tuple,
- ClientSchema schema,
- ClientMessagePacker out,
- boolean keyOnly
- ) {
- writeTuple(tuple, schema, out, keyOnly, false);
- }
-
- /**
- * Writes {@link Tuple}.
+ * Writes transaction, if present.
*
- * @param tuple Tuple.
- * @param schema Schema.
- * @param out Out.
- * @param keyOnly Key only.
- * @param skipHeader Skip header.
+ * @param tx Transaction.
+ * @param out Packer.
*/
- public void writeTuple(
- @NotNull Tuple tuple,
- ClientSchema schema,
- ClientMessagePacker out,
- boolean keyOnly,
- boolean skipHeader
- ) {
- if (!skipHeader) {
- out.packIgniteUuid(id);
- out.packInt(schema.version());
- }
-
- var columns = schema.columns();
- var count = keyOnly ? schema.keyColumnCount() : columns.length;
-
- for (var i = 0; i < count; i++) {
- var col = columns[i];
-
- Object v = tuple.valueOrDefault(col.name(), NO_VALUE);
-
- out.packObject(v);
- }
- }
-
- /**
- * Writes key and value {@link Tuple}.
- *
- * @param key Key tuple.
- * @param val Value tuple.
- * @param schema Schema.
- * @param out Out.
- * @param skipHeader Skip header.
- */
- public void writeKvTuple(
- @NotNull Tuple key,
- @Nullable Tuple val,
- ClientSchema schema,
- ClientMessagePacker out,
- boolean skipHeader
- ) {
- if (!skipHeader) {
- out.packIgniteUuid(id);
- out.packInt(schema.version());
- }
-
- var columns = schema.columns();
-
- for (var i = 0; i < columns.length; i++) {
- var col = columns[i];
-
- Object v = col.key()
- ? key.valueOrDefault(col.name(), NO_VALUE)
- : val != null
- ? val.valueOrDefault(col.name(), NO_VALUE)
- : NO_VALUE;
-
- out.packObject(v);
- }
- }
-
- /**
- * Writes pairs {@link Tuple}.
- *
- * @param pairs Key tuple.
- * @param schema Schema.
- * @param out Out.
- */
- public void writeKvTuples(Map<Tuple, Tuple> pairs, ClientSchema schema, ClientMessagePacker out) {
- out.packIgniteUuid(id);
- out.packInt(schema.version());
- out.packInt(pairs.size());
-
- for (Map.Entry<Tuple, Tuple> pair : pairs.entrySet()) {
- writeKvTuple(pair.getKey(), pair.getValue(), schema, out, true);
- }
- }
-
- /**
- * Writes {@link Tuple}'s.
- *
- * @param tuples Tuples.
- * @param schema Schema.
- * @param out Out.
- * @param keyOnly Key only.
- */
- public void writeTuples(
- @NotNull Collection<Tuple> tuples,
- ClientSchema schema,
- ClientMessagePacker out,
- boolean keyOnly
- ) {
- out.packIgniteUuid(id);
- out.packInt(schema.version());
- out.packInt(tuples.size());
-
- for (var tuple : tuples) {
- writeTuple(tuple, schema, out, keyOnly, true);
- }
- }
-
- private static Tuple readTuple(ClientSchema schema, ClientMessageUnpacker in, boolean keyOnly) {
- var tuple = new ClientTuple(schema);
-
- var colCnt = keyOnly ? schema.keyColumnCount() : schema.columns().length;
-
- for (var i = 0; i < colCnt; i++) {
- tuple.setInternal(i, in.unpackObject(schema.columns()[i].type()));
- }
-
- return tuple;
- }
-
- static Tuple readValueTuple(ClientSchema schema, ClientMessageUnpacker in, Tuple keyTuple) {
- var tuple = new ClientTuple(schema);
-
- for (var i = 0; i < schema.columns().length; i++) {
- ClientColumn col = schema.columns()[i];
-
- Object value = i < schema.keyColumnCount()
- ? keyTuple.value(col.name())
- : in.unpackObject(schema.columns()[i].type());
-
- tuple.setInternal(i, value);
- }
-
- return tuple;
- }
-
- static Tuple readValueTuple(ClientSchema schema, ClientMessageUnpacker in) {
- var keyColCnt = schema.keyColumnCount();
- var colCnt = schema.columns().length;
-
- var valTuple = new ClientTuple(schema, keyColCnt, schema.columns().length - 1);
-
- for (var i = keyColCnt; i < colCnt; i++) {
- ClientColumn col = schema.columns()[i];
- Object val = in.unpackObject(col.type());
-
- valTuple.setInternal(i - keyColCnt, val);
- }
-
- return valTuple;
- }
-
- static IgniteBiTuple<Tuple, Tuple> readKvTuple(ClientSchema schema, ClientMessageUnpacker in) {
- var keyColCnt = schema.keyColumnCount();
- var colCnt = schema.columns().length;
-
- var keyTuple = new ClientTuple(schema, 0, keyColCnt - 1);
- var valTuple = new ClientTuple(schema, keyColCnt, schema.columns().length - 1);
-
- for (var i = 0; i < colCnt; i++) {
- ClientColumn col = schema.columns()[i];
- Object val = in.unpackObject(col.type());
-
- if (i < keyColCnt) {
- keyTuple.setInternal(i, val);
- } else {
- valTuple.setInternal(i - keyColCnt, val);
+ public static void writeTx(@Nullable Transaction tx, PayloadOutputChannel out) {
+ if (tx == null) {
+ out.out().packNil();
+ } else {
+ ClientTransaction clientTx = getClientTx(tx);
+
+ if (clientTx.channel() != out.clientChannel()) {
+ throw new IgniteClientException("Transaction context has been lost due to connection errors.");
}
- }
-
- return new IgniteBiTuple<>(keyTuple, valTuple);
- }
-
- /**
- * Reads {@link Tuple} pairs.
- *
- * @param schema Schema.
- * @param in In.
- * @return Tuple pairs.
- */
- public Map<Tuple, Tuple> readKvTuplesNullable(ClientSchema schema, ClientMessageUnpacker in) {
- var cnt = in.unpackInt();
- Map<Tuple, Tuple> res = new HashMap<>(cnt);
-
- for (int i = 0; i < cnt; i++) {
- var hasValue = in.unpackBoolean();
-
- if (hasValue) {
- var pair = readKvTuple(schema, in);
- res.put(pair.get1(), pair.get2());
- }
+ out.out().packLong(clientTx.id());
}
-
- return res;
- }
-
- Collection<Tuple> readTuples(ClientSchema schema, ClientMessageUnpacker in) {
- return readTuples(schema, in, false);
}
- Collection<Tuple> readTuples(ClientSchema schema, ClientMessageUnpacker in, boolean keyOnly) {
- var cnt = in.unpackInt();
- var res = new ArrayList<Tuple>(cnt);
-
- for (int i = 0; i < cnt; i++) {
- res.add(readTuple(schema, in, keyOnly));
- }
-
- return res;
- }
-
- Collection<Tuple> readTuplesNullable(ClientSchema schema, ClientMessageUnpacker in) {
- var cnt = in.unpackInt();
- var res = new ArrayList<Tuple>(cnt);
-
- for (int i = 0; i < cnt; i++) {
- var tuple = in.unpackBoolean()
- ? readTuple(schema, in, false)
- : null;
-
- res.add(tuple);
+ private static ClientTransaction getClientTx(@NotNull Transaction tx) {
+ if (!(tx instanceof ClientTransaction)) {
+ throw new IgniteClientException("Unsupported transaction implementation: '"
+ + tx.getClass()
+ + "'. Use IgniteClient.transactions() to start transactions.");
}
- return res;
+ return (ClientTransaction) tx;
}
<T> CompletableFuture<T> doSchemaOutInOpAsync(
int opCode,
- BiConsumer<ClientSchema, ClientMessagePacker> writer,
+ BiConsumer<ClientSchema, PayloadOutputChannel> writer,
BiFunction<ClientSchema, ClientMessageUnpacker, T> reader
) {
return doSchemaOutInOpAsync(opCode, writer, reader, null);
@@ -479,26 +245,26 @@ public class ClientTable implements Table {
<T> CompletableFuture<T> doSchemaOutInOpAsync(
int opCode,
- BiConsumer<ClientSchema, ClientMessagePacker> writer,
+ BiConsumer<ClientSchema, PayloadOutputChannel> writer,
BiFunction<ClientSchema, ClientMessageUnpacker, T> reader,
T defaultValue
) {
return getLatestSchema()
.thenCompose(schema ->
ch.serviceAsync(opCode,
- w -> writer.accept(schema, w.out()),
+ w -> writer.accept(schema, w),
r -> readSchemaAndReadData(schema, r.in(), reader, defaultValue)))
.thenCompose(t -> loadSchemaAndReadData(t, reader));
}
<T> CompletableFuture<T> doSchemaOutOpAsync(
int opCode,
- BiConsumer<ClientSchema, ClientMessagePacker> writer,
+ BiConsumer<ClientSchema, PayloadOutputChannel> writer,
Function<ClientMessageUnpacker, T> reader) {
return getLatestSchema()
.thenCompose(schema ->
ch.serviceAsync(opCode,
- w -> writer.accept(schema, w.out()),
+ w -> writer.accept(schema, w),
r -> reader.apply(r.in())));
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
new file mode 100644
index 0000000..8bf7d30
--- /dev/null
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
@@ -0,0 +1,320 @@
+/*
+ * 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.table;
+
+import static org.apache.ignite.internal.client.proto.ClientMessageCommon.NO_VALUE;
+import static org.apache.ignite.internal.client.table.ClientTable.writeTx;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.ignite.internal.client.PayloadOutputChannel;
+import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.tx.Transaction;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Tuple serializer.
+ */
+class ClientTupleSerializer {
+ /** Table ID. */
+ private final IgniteUuid tableId;
+
+ /**
+ * Constructor.
+ *
+ * @param tableId Table id.
+ */
+ ClientTupleSerializer(IgniteUuid tableId) {
+ this.tableId = tableId;
+ }
+
+ /**
+ * Writes {@link Tuple}.
+ *
+ * @param tuple Tuple.
+ * @param schema Schema.
+ * @param out Out.
+ */
+ public void writeTuple(
+ @Nullable Transaction tx,
+ @NotNull Tuple tuple,
+ ClientSchema schema,
+ PayloadOutputChannel out
+ ) {
+ writeTuple(tx, tuple, schema, out, false, false);
+ }
+
+ /**
+ * Writes {@link Tuple}.
+ *
+ * @param tuple Tuple.
+ * @param schema Schema.
+ * @param out Out.
+ * @param keyOnly Key only.
+ */
+ public void writeTuple(
+ @Nullable Transaction tx,
+ @NotNull Tuple tuple,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ boolean keyOnly
+ ) {
+ writeTuple(tx, tuple, schema, out, keyOnly, false);
+ }
+
+ /**
+ * Writes {@link Tuple}.
+ *
+ * @param tuple Tuple.
+ * @param schema Schema.
+ * @param out Out.
+ * @param keyOnly Key only.
+ * @param skipHeader Skip header.
+ */
+ public void writeTuple(
+ @Nullable Transaction tx,
+ @NotNull Tuple tuple,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ boolean keyOnly,
+ boolean skipHeader
+ ) {
+ if (!skipHeader) {
+ out.out().packIgniteUuid(tableId);
+ writeTx(tx, out);
+ out.out().packInt(schema.version());
+ }
+
+ var columns = schema.columns();
+ var count = keyOnly ? schema.keyColumnCount() : columns.length;
+
+ for (var i = 0; i < count; i++) {
+ var col = columns[i];
+
+ Object v = tuple.valueOrDefault(col.name(), NO_VALUE);
+
+ out.out().packObject(v);
+ }
+ }
+
+ /**
+ * Writes key and value {@link Tuple}.
+ *
+ * @param key Key tuple.
+ * @param val Value tuple.
+ * @param schema Schema.
+ * @param out Out.
+ * @param skipHeader Skip header.
+ */
+ public void writeKvTuple(
+ @Nullable Transaction tx,
+ @NotNull Tuple key,
+ @Nullable Tuple val,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ boolean skipHeader
+ ) {
+ if (!skipHeader) {
+ out.out().packIgniteUuid(tableId);
+ writeTx(tx, out);
+ out.out().packInt(schema.version());
+ }
+
+ var columns = schema.columns();
+
+ for (var i = 0; i < columns.length; i++) {
+ var col = columns[i];
+
+ Object v = col.key()
+ ? key.valueOrDefault(col.name(), NO_VALUE)
+ : val != null
+ ? val.valueOrDefault(col.name(), NO_VALUE)
+ : NO_VALUE;
+
+ out.out().packObject(v);
+ }
+ }
+
+ /**
+ * Writes pairs {@link Tuple}.
+ *
+ * @param pairs Key tuple.
+ * @param schema Schema.
+ * @param out Out.
+ */
+ public void writeKvTuples(@Nullable Transaction tx, Map<Tuple, Tuple> pairs, ClientSchema schema, PayloadOutputChannel out) {
+ out.out().packIgniteUuid(tableId);
+ writeTx(tx, out);
+ out.out().packInt(schema.version());
+ out.out().packInt(pairs.size());
+
+ for (Map.Entry<Tuple, Tuple> pair : pairs.entrySet()) {
+ writeKvTuple(tx, pair.getKey(), pair.getValue(), schema, out, true);
+ }
+ }
+
+ /**
+ * Writes {@link Tuple}'s.
+ *
+ * @param tuples Tuples.
+ * @param schema Schema.
+ * @param out Out.
+ * @param keyOnly Key only.
+ */
+ public void writeTuples(
+ @Nullable Transaction tx,
+ @NotNull Collection<Tuple> tuples,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ boolean keyOnly
+ ) {
+ out.out().packIgniteUuid(tableId);
+ writeTx(tx, out);
+ out.out().packInt(schema.version());
+ out.out().packInt(tuples.size());
+
+ for (var tuple : tuples) {
+ writeTuple(tx, tuple, schema, out, keyOnly, true);
+ }
+ }
+
+ public static Tuple readTuple(ClientSchema schema, ClientMessageUnpacker in, boolean keyOnly) {
+ var tuple = new ClientTuple(schema);
+
+ var colCnt = keyOnly ? schema.keyColumnCount() : schema.columns().length;
+
+ for (var i = 0; i < colCnt; i++) {
+ tuple.setInternal(i, in.unpackObject(schema.columns()[i].type()));
+ }
+
+ return tuple;
+ }
+
+ public static Tuple readValueTuple(ClientSchema schema, ClientMessageUnpacker in, Tuple keyTuple) {
+ var tuple = new ClientTuple(schema);
+
+ for (var i = 0; i < schema.columns().length; i++) {
+ ClientColumn col = schema.columns()[i];
+
+ Object value = i < schema.keyColumnCount()
+ ? keyTuple.value(col.name())
+ : in.unpackObject(schema.columns()[i].type());
+
+ tuple.setInternal(i, value);
+ }
+
+ return tuple;
+ }
+
+ public static Tuple readValueTuple(ClientSchema schema, ClientMessageUnpacker in) {
+ var keyColCnt = schema.keyColumnCount();
+ var colCnt = schema.columns().length;
+
+ var valTuple = new ClientTuple(schema, keyColCnt, schema.columns().length - 1);
+
+ for (var i = keyColCnt; i < colCnt; i++) {
+ ClientColumn col = schema.columns()[i];
+ Object val = in.unpackObject(col.type());
+
+ valTuple.setInternal(i - keyColCnt, val);
+ }
+
+ return valTuple;
+ }
+
+ public static IgniteBiTuple<Tuple, Tuple> readKvTuple(ClientSchema schema, ClientMessageUnpacker in) {
+ var keyColCnt = schema.keyColumnCount();
+ var colCnt = schema.columns().length;
+
+ var keyTuple = new ClientTuple(schema, 0, keyColCnt - 1);
+ var valTuple = new ClientTuple(schema, keyColCnt, schema.columns().length - 1);
+
+ for (var i = 0; i < colCnt; i++) {
+ ClientColumn col = schema.columns()[i];
+ Object val = in.unpackObject(col.type());
+
+ if (i < keyColCnt) {
+ keyTuple.setInternal(i, val);
+ } else {
+ valTuple.setInternal(i - keyColCnt, val);
+ }
+ }
+
+ return new IgniteBiTuple<>(keyTuple, valTuple);
+ }
+
+ /**
+ * Reads {@link Tuple} pairs.
+ *
+ * @param schema Schema.
+ * @param in In.
+ * @return Tuple pairs.
+ */
+ public static Map<Tuple, Tuple> readKvTuplesNullable(ClientSchema schema, ClientMessageUnpacker in) {
+ var cnt = in.unpackInt();
+ Map<Tuple, Tuple> res = new HashMap<>(cnt);
+
+ for (int i = 0; i < cnt; i++) {
+ var hasValue = in.unpackBoolean();
+
+ if (hasValue) {
+ var pair = readKvTuple(schema, in);
+
+ res.put(pair.get1(), pair.get2());
+ }
+ }
+
+ return res;
+ }
+
+ public static Collection<Tuple> readTuples(ClientSchema schema, ClientMessageUnpacker in) {
+ return readTuples(schema, in, false);
+ }
+
+ public static Collection<Tuple> readTuples(ClientSchema schema, ClientMessageUnpacker in, boolean keyOnly) {
+ var cnt = in.unpackInt();
+ var res = new ArrayList<Tuple>(cnt);
+
+ for (int i = 0; i < cnt; i++) {
+ res.add(readTuple(schema, in, keyOnly));
+ }
+
+ return res;
+ }
+
+ public static Collection<Tuple> readTuplesNullable(ClientSchema schema, ClientMessageUnpacker in) {
+ var cnt = in.unpackInt();
+ var res = new ArrayList<Tuple>(cnt);
+
+ for (int i = 0; i < cnt; i++) {
+ var tuple = in.unpackBoolean()
+ ? readTuple(schema, in, false)
+ : null;
+
+ res.add(tuple);
+ }
+
+ return res;
+ }
+}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransaction.java b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransaction.java
new file mode 100644
index 0000000..c9b4f42
--- /dev/null
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransaction.java
@@ -0,0 +1,117 @@
+/*
+ * 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.tx;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.client.ClientChannel;
+import org.apache.ignite.internal.client.proto.ClientOp;
+import org.apache.ignite.tx.Transaction;
+import org.apache.ignite.tx.TransactionException;
+
+/**
+ * Client transaction.
+ */
+public class ClientTransaction implements Transaction {
+ /** Open state. */
+ private static final int STATE_OPEN = 0;
+
+ /** Committed state. */
+ private static final int STATE_COMMITTED = 1;
+
+ /** Rolled back state. */
+ private static final int STATE_ROLLED_BACK = 2;
+
+ /** Channel that the transaction belongs to. */
+ private final ClientChannel ch;
+
+ /** Transaction id. */
+ private final long id;
+
+ /** State. */
+ private final AtomicInteger state = new AtomicInteger(STATE_OPEN);
+
+ /**
+ * Constructor.
+ *
+ * @param ch Channel that the transaction belongs to.
+ * @param id Transaction id.
+ */
+ public ClientTransaction(ClientChannel ch, long id) {
+ this.ch = ch;
+ this.id = id;
+ }
+
+ /**
+ * Gets the id.
+ *
+ * @return Id.
+ */
+ public long id() {
+ return id;
+ }
+
+ /**
+ * Gets the associated channel.
+ *
+ * @return Channel.
+ */
+ public ClientChannel channel() {
+ return ch;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void commit() throws TransactionException {
+ commitAsync().join();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public CompletableFuture<Void> commitAsync() {
+ setState(STATE_COMMITTED);
+
+ return ch.serviceAsync(ClientOp.TX_COMMIT, w -> w.out().packLong(id), r -> null);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void rollback() throws TransactionException {
+ rollbackAsync().join();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public CompletableFuture<Void> rollbackAsync() {
+ setState(STATE_ROLLED_BACK);
+
+ return ch.serviceAsync(ClientOp.TX_ROLLBACK, w -> w.out().packLong(id), r -> null);
+ }
+
+ private void setState(int state) {
+ if (this.state.compareAndSet(STATE_OPEN, state)) {
+ return;
+ }
+
+ String message = this.state.get() == STATE_COMMITTED
+ ? "Transaction is already committed."
+ : "Transaction is already rolled back.";
+
+ throw new TransactionException(message);
+ }
+}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java
similarity index 54%
copy from modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
copy to modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java
index 972d02e..3c0f074 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java
@@ -15,59 +15,46 @@
* limitations under the License.
*/
-package org.apache.ignite.client.fakes;
+package org.apache.ignite.internal.client.tx;
-import java.util.Set;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.internal.processors.query.calcite.QueryProcessor;
-import org.apache.ignite.table.manager.IgniteTables;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.client.ReliableChannel;
+import org.apache.ignite.internal.client.proto.ClientOp;
import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.tx.Transaction;
/**
- * Fake Ignite.
+ * Client transactions implementation.
*/
-public class FakeIgnite implements Ignite {
+public class ClientTransactions implements IgniteTransactions {
+ /** Channel. */
+ private final ReliableChannel ch;
+
/**
- * Default constructor.
+ * Constructor.
+ *
+ * @param ch Channel.
*/
- public FakeIgnite() {
- super();
- }
-
- private final IgniteTables tables = new FakeIgniteTables();
-
- /** {@inheritDoc} */
- @Override
- public IgniteTables tables() {
- return tables;
- }
-
- public QueryProcessor queryEngine() {
- return new FakeIgniteQueryProcessor();
+ public ClientTransactions(ReliableChannel ch) {
+ this.ch = ch;
}
/** {@inheritDoc} */
@Override
- public IgniteTransactions transactions() {
- return null;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override public void setBaseline(Set<String> baselineNodes) {
+ public IgniteTransactions withTimeout(long timeout) {
+ // TODO: IGNITE-16193
throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
@Override
- public void close() {
- // No-op.
+ public Transaction begin() {
+ return beginAsync().join();
}
/** {@inheritDoc} */
@Override
- public String name() {
- return null;
+ public CompletableFuture<Transaction> beginAsync() {
+ return ch.serviceAsync(ClientOp.TX_BEGIN, w -> {}, r -> new ClientTransaction(r.clientChannel(), r.in().unpackLong()));
}
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/ClientTableTest.java b/modules/client/src/test/java/org/apache/ignite/client/ClientTableTest.java
index 3495e08..a7ad4a3 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/ClientTableTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/ClientTableTest.java
@@ -58,7 +58,7 @@ public class ClientTableTest extends AbstractClientTableTest {
table.upsert(null, tuple);
- Tuple key = tuple(123L);
+ Tuple key = tuple(DEFAULT_ID);
var resTuple = table.get(null, key);
assertEquals(DEFAULT_NAME, resTuple.stringValue("name"));
diff --git a/modules/client/src/test/java/org/apache/ignite/client/TestServer.java b/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
index 490dad9..76f8d63 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
@@ -70,7 +70,13 @@ public class TestServer implements AutoCloseable {
bootstrapFactory.start();
- module = new ClientHandlerModule(((FakeIgnite) ignite).queryEngine(), ignite.tables(), cfg, bootstrapFactory);
+ module = new ClientHandlerModule(
+ ((FakeIgnite) ignite).queryEngine(),
+ ignite.tables(),
+ ignite.transactions(),
+ cfg,
+ bootstrapFactory
+ );
module.start();
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
index 972d02e..ac7d58a 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
@@ -18,10 +18,12 @@
package org.apache.ignite.client.fakes;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.Ignite;
import org.apache.ignite.internal.processors.query.calcite.QueryProcessor;
import org.apache.ignite.table.manager.IgniteTables;
import org.apache.ignite.tx.IgniteTransactions;
+import org.apache.ignite.tx.Transaction;
/**
* Fake Ignite.
@@ -49,7 +51,22 @@ public class FakeIgnite implements Ignite {
/** {@inheritDoc} */
@Override
public IgniteTransactions transactions() {
- return null;
+ return new IgniteTransactions() {
+ @Override
+ public IgniteTransactions withTimeout(long timeout) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Transaction begin() {
+ return beginAsync().join();
+ }
+
+ @Override
+ public CompletableFuture<Transaction> beginAsync() {
+ throw new UnsupportedOperationException();
+ }
+ };
}
/**
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs
index 407fba0..690a1aa 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/Table/Table.cs
@@ -625,6 +625,7 @@ namespace Apache.Ignite.Internal.Table
var w = buf.GetMessageWriter();
w.Write(Id);
+ w.WriteNil(); // Transaction ID (TODO: IGNITE-16208).
w.Write(schema.Version);
w.Flush();
@@ -657,6 +658,7 @@ namespace Apache.Ignite.Internal.Table
bool keyOnly = false)
{
w.Write(Id);
+ w.WriteNil(); // Transaction ID (TODO: IGNITE-16208).
w.Write(schema.Version);
WriteTuple(ref w, schema, tuple, keyOnly);
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItThinClientConnectionTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
similarity index 58%
rename from modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItThinClientConnectionTest.java
rename to modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
index ef3ab59..bbb2123 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItThinClientConnectionTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
@@ -15,13 +15,13 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.runner.app;
+package org.apache.ignite.internal.runner.app.client;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
@@ -33,39 +33,46 @@ import org.apache.ignite.internal.ItUtils;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.schema.configuration.SchemaConfigurationConverter;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
+import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.schema.SchemaBuilders;
import org.apache.ignite.schema.definition.ColumnType;
import org.apache.ignite.schema.definition.TableDefinition;
-import org.apache.ignite.table.RecordView;
-import org.apache.ignite.table.Table;
-import org.apache.ignite.table.Tuple;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.extension.ExtendWith;
/**
- * Tests thin client connecting to a real server node.
+ * Thin client integration test base class.
*/
+@SuppressWarnings("ZeroLengthArrayAllocation")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@ExtendWith(WorkDirectoryExtension.class)
-public class ItThinClientConnectionTest extends IgniteAbstractTest {
- private static final String SCHEMA_NAME = "PUB";
+public abstract class ItAbstractThinClientTest extends IgniteAbstractTest {
+ protected static final String SCHEMA_NAME = "PUB";
- private static final String TABLE_NAME = "tbl1";
+ protected static final String TABLE_NAME = "tbl1";
+
+ protected static final String COLUMN_KEY = "key";
+
+ protected static final String COLUMN_VAL = "val";
- /** Nodes bootstrap configuration. */
private final Map<String, String> nodesBootstrapCfg = new LinkedHashMap<>();
private final List<Ignite> startedNodes = new ArrayList<>();
+ private IgniteClient client;
+
/**
* Before each.
*/
- @BeforeEach
- void setup(TestInfo testInfo) {
+ @BeforeAll
+ void beforeAll(TestInfo testInfo, @WorkDirectory Path workDir) {
+ this.workDir = workDir;
+
String node0Name = testNodeName(testInfo, 3344);
String node1Name = testNodeName(testInfo, 3345);
@@ -94,32 +101,15 @@ public class ItThinClientConnectionTest extends IgniteAbstractTest {
+ " }\n"
+ "}"
);
- }
- /**
- * After each.
- */
- @AfterEach
- void tearDown() throws Exception {
- IgniteUtils.closeAll(ItUtils.reverse(startedNodes));
- }
-
- /**
- * Check that thin client can connect to any server node and work with table API.
- */
- @Test
- void testThinClientConnectsToServerNodesAndExecutesBasicTableOperations() throws Exception {
nodesBootstrapCfg.forEach((nodeName, configStr) ->
startedNodes.add(IgnitionManager.start(nodeName, configStr, workDir.resolve(nodeName)))
);
- var keyCol = "key";
- var valCol = "val";
-
TableDefinition schTbl = SchemaBuilders.tableBuilder(SCHEMA_NAME, TABLE_NAME).columns(
- SchemaBuilders.column(keyCol, ColumnType.INT32).build(),
- SchemaBuilders.column(valCol, ColumnType.string()).asNullable(true).build()
- ).withPrimaryKey(keyCol).build();
+ SchemaBuilders.column(COLUMN_KEY, ColumnType.INT32).build(),
+ SchemaBuilders.column(COLUMN_VAL, ColumnType.string()).asNullable(true).build()
+ ).withPrimaryKey(COLUMN_KEY).build();
startedNodes.get(0).tables().createTable(schTbl.canonicalName(), tblCh ->
SchemaConfigurationConverter.convert(schTbl, tblCh)
@@ -127,47 +117,41 @@ public class ItThinClientConnectionTest extends IgniteAbstractTest {
.changePartitions(10)
);
- var addrs = new String[]{"127.0.0.1:"
- + ((InetSocketAddress) ((IgniteImpl) startedNodes.get(0)).clientHandlerModule().localAddress()).getPort()};
-
- for (var addr : addrs) {
- try (var client = IgniteClient.builder().addresses(addr).build()) {
- List<Table> tables = client.tables().tables();
- assertEquals(1, tables.size());
-
- Table table = tables.get(0);
- assertEquals(String.format("%s.%s", SCHEMA_NAME, TABLE_NAME), table.name());
-
- var tuple = Tuple.create().set(keyCol, 1).set(valCol, "Hello");
- var keyTuple = Tuple.create().set(keyCol, 1);
-
- RecordView<Tuple> recView = table.recordView();
+ client = IgniteClient.builder().addresses(getNodeAddresses().toArray(new String[0])).build();
+ }
- recView.upsert(null, tuple);
- assertEquals("Hello", recView.get(null, keyTuple).stringValue(valCol));
+ /**
+ * After each.
+ */
+ @AfterAll
+ void afterAll() throws Exception {
+ client.close();
- var kvView = table.keyValueView();
- assertEquals("Hello", kvView.get(null, keyTuple).stringValue(valCol));
+ IgniteUtils.closeAll(ItUtils.reverse(startedNodes));
+ }
- var pojoView = table.recordView(TestPojo.class);
- assertEquals("Hello", pojoView.get(null, new TestPojo(1)).val);
+ protected void stopNode() throws Exception {
+ startedNodes.remove(0).close();
+ }
- assertTrue(recView.delete(null, keyTuple));
- }
- }
+ protected String getNodeAddress() {
+ return getNodeAddresses().get(0);
}
- private static class TestPojo {
- public TestPojo() {
- //No-op.
- }
+ protected List<String> getNodeAddresses() {
+ List<String> res = new ArrayList<>(startedNodes.size());
+
+ for (Ignite ignite : startedNodes) {
+ SocketAddress addr = ((IgniteImpl) ignite).clientHandlerModule().localAddress();
+ int port = ((InetSocketAddress) addr).getPort();
- public TestPojo(int key) {
- this.key = key;
+ res.add("127.0.0.1:" + port);
}
- public int key;
+ return res;
+ }
- public String val;
+ protected IgniteClient client() {
+ return client;
}
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientConnectionTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientConnectionTest.java
new file mode 100644
index 0000000..7a48290
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientConnectionTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.runner.app.client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.table.RecordView;
+import org.apache.ignite.table.Table;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests thin client connecting to a real server node.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+public class ItThinClientConnectionTest extends ItAbstractThinClientTest {
+ /**
+ * Check that thin client can connect to any server node and work with table API.
+ */
+ @Test
+ void testThinClientConnectsToServerNodesAndExecutesBasicTableOperations() throws Exception {
+ for (var addr : getNodeAddresses()) {
+ try (var client = IgniteClient.builder().addresses(addr).build()) {
+ List<Table> tables = client.tables().tables();
+ assertEquals(1, tables.size());
+
+ Table table = tables.get(0);
+ assertEquals(String.format("%s.%s", SCHEMA_NAME, TABLE_NAME), table.name());
+
+ var tuple = Tuple.create().set(COLUMN_KEY, 1).set(COLUMN_VAL, "Hello");
+ var keyTuple = Tuple.create().set(COLUMN_KEY, 1);
+
+ RecordView<Tuple> recView = table.recordView();
+
+ recView.upsert(null, tuple);
+ assertEquals("Hello", recView.get(null, keyTuple).stringValue(COLUMN_VAL));
+
+ var kvView = table.keyValueView();
+ assertEquals("Hello", kvView.get(null, keyTuple).stringValue(COLUMN_VAL));
+
+ var pojoView = table.recordView(TestPojo.class);
+ assertEquals("Hello", pojoView.get(null, new TestPojo(1)).val);
+
+ assertTrue(recView.delete(null, keyTuple));
+ }
+ }
+ }
+
+ private static class TestPojo {
+ public TestPojo() {
+ //No-op.
+ }
+
+ public TestPojo(int key) {
+ this.key = key;
+ }
+
+ public int key;
+
+ public String val;
+ }
+}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java
new file mode 100644
index 0000000..f157f46
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java
@@ -0,0 +1,312 @@
+/*
+ * 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.runner.app.client;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.startsWith;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.client.IgniteClientException;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.RecordView;
+import org.apache.ignite.table.Table;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.mapper.Mapper;
+import org.apache.ignite.tx.Transaction;
+import org.apache.ignite.tx.TransactionException;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Thin client transactions integration test.
+ */
+public class ItThinClientTransactionsTest extends ItAbstractThinClientTest {
+ @Test
+ void testKvViewOperations() {
+ KeyValueView<Integer, String> kvView = kvView();
+
+ int key = 1;
+ kvView.put(null, key, "1");
+
+ Transaction tx = client().transactions().begin();
+ kvView.put(tx, key, "22");
+
+ assertTrue(kvView.contains(tx, key));
+ assertFalse(kvView.remove(tx, key, "1"));
+ assertFalse(kvView.putIfAbsent(tx, key, "111"));
+ assertEquals("22", kvView.get(tx, key));
+ assertEquals("22", kvView.getAndPut(tx, key, "33"));
+ assertEquals("33", kvView.getAndReplace(tx, key, "44"));
+ assertTrue(kvView.replace(tx, key, "55"));
+ assertEquals("55", kvView.getAndRemove(tx, key));
+ assertFalse(kvView.contains(tx, key));
+ assertFalse(kvView.remove(tx, key));
+
+ kvView.putAll(tx, Map.of(key, "6", 2, "7"));
+ assertEquals(2, kvView.getAll(tx, List.of(key, 2, 3)).size());
+
+ tx.rollback();
+ assertEquals("1", kvView.get(null, key));
+ }
+
+ @Test
+ void testKvViewBinaryOperations() {
+ KeyValueView<Tuple, Tuple> kvView = table().keyValueView();
+
+ Tuple key = key(1);
+ kvView.put(null, key, val("1"));
+
+ Transaction tx = client().transactions().begin();
+ kvView.put(tx, key, val("22"));
+
+ assertTrue(kvView.contains(tx, key));
+ assertFalse(kvView.remove(tx, key, val("1")));
+ assertFalse(kvView.putIfAbsent(tx, key, val("111")));
+ assertEquals(val("22"), kvView.get(tx, key));
+ assertEquals(val("22"), kvView.getAndPut(tx, key, val("33")));
+ assertEquals(val("33"), kvView.getAndReplace(tx, key, val("44")));
+ assertTrue(kvView.replace(tx, key, val("55")));
+ assertEquals(val("55"), kvView.getAndRemove(tx, key));
+ assertFalse(kvView.contains(tx, key));
+ assertFalse(kvView.remove(tx, key));
+
+ kvView.putAll(tx, Map.of(key, val("6"), key(2), val("7")));
+ assertEquals(2, kvView.getAll(tx, List.of(key, key(2), key(3))).size());
+
+ tx.rollback();
+ assertEquals(val("1"), kvView.get(null, key));
+ }
+
+ @Test
+ void testRecordViewOperations() {
+ RecordView<Rec> recordView = table().recordView(Mapper.of(Rec.class));
+ Rec key = rec(1, null);
+ recordView.upsert(null, rec(1, "1"));
+
+ Transaction tx = client().transactions().begin();
+ recordView.upsert(tx, rec(1, "22"));
+
+ assertFalse(recordView.deleteExact(tx, rec(1, "1")));
+ assertFalse(recordView.insert(tx, rec(1, "111")));
+ assertEquals(rec(1, "22"), recordView.get(tx, key));
+ assertEquals(rec(1, "22"), recordView.getAndUpsert(tx, rec(1, "33")));
+ assertEquals(rec(1, "33"), recordView.getAndReplace(tx, rec(1, "44")));
+ assertTrue(recordView.replace(tx, rec(1, "55")));
+ assertEquals(rec(1, "55"), recordView.getAndDelete(tx, key));
+ assertFalse(recordView.delete(tx, key));
+
+ recordView.upsertAll(tx, List.of(rec(1, "6"), rec(2, "7")));
+ assertEquals(2, recordView.getAll(tx, List.of(key, rec(2, null), rec(3, null))).size());
+
+ tx.rollback();
+ assertEquals(rec(1, "1"), recordView.get(null, key));
+ }
+
+ @Test
+ void testRecordViewBinaryOperations() {
+ RecordView<Tuple> recordView = table().recordView();
+
+ Tuple key = key(1);
+ recordView.upsert(null, kv(1, "1"));
+
+ Transaction tx = client().transactions().begin();
+ recordView.upsert(tx, kv(1, "22"));
+
+ assertFalse(recordView.deleteExact(tx, kv(1, "1")));
+ assertFalse(recordView.insert(tx, kv(1, "111")));
+ assertEquals(kv(1, "22"), recordView.get(tx, key));
+ assertEquals(kv(1, "22"), recordView.getAndUpsert(tx, kv(1, "33")));
+ assertEquals(kv(1, "33"), recordView.getAndReplace(tx, kv(1, "44")));
+ assertTrue(recordView.replace(tx, kv(1, "55")));
+ assertEquals(kv(1, "55"), recordView.getAndDelete(tx, key));
+ assertFalse(recordView.delete(tx, key));
+
+ recordView.upsertAll(tx, List.of(kv(1, "6"), kv(2, "7")));
+ assertEquals(2, recordView.getAll(tx, List.of(key, key(2), key(3))).size());
+
+ tx.rollback();
+ assertEquals(kv(1, "1"), recordView.get(null, key));
+ }
+
+ @Test
+ void testCommitUpdatesData() {
+ KeyValueView<Integer, String> kvView = kvView();
+ kvView.put(null, 1, "1");
+
+ Transaction tx = client().transactions().begin();
+ assertEquals("1", kvView.get(null, 1));
+ assertEquals("1", kvView.get(tx, 1));
+
+ kvView.put(tx, 1, "2");
+ assertEquals("2", kvView.get(tx, 1));
+
+ tx.commit();
+ assertEquals("2", kvView.get(null, 1));
+ }
+
+ @Test
+ void testRollbackDoesNotUpdateData() {
+ KeyValueView<Integer, String> kvView = kvView();
+ kvView.put(null, 1, "1");
+
+ Transaction tx = client().transactions().begin();
+ assertEquals("1", kvView.get(null, 1));
+ assertEquals("1", kvView.get(tx, 1));
+
+ kvView.put(tx, 1, "2");
+ assertEquals("2", kvView.get(tx, 1));
+
+ tx.rollback();
+ assertEquals("1", kvView.get(null, 1));
+ }
+
+ @Test
+ void testAccessLockedKeyTimesOut() {
+ KeyValueView<Integer, String> kvView = kvView();
+
+ Transaction tx = client().transactions().begin();
+ kvView.put(tx, -100, "1");
+
+ var ex = assertThrows(CompletionException.class, () -> kvView.get(null, -100));
+ assertThat(ex.getCause().getMessage(), containsString("TimeoutException"));
+
+ tx.rollback();
+ }
+
+ @Test
+ void testCommitRollbackSameTxThrows() {
+ Transaction tx = client().transactions().begin();
+ tx.commit();
+
+ TransactionException ex = assertThrows(TransactionException.class, tx::rollback);
+ assertEquals("Transaction is already committed.", ex.getMessage());
+ }
+
+ @Test
+ void testRollbackCommitSameTxThrows() {
+ Transaction tx = client().transactions().begin();
+ tx.rollback();
+
+ TransactionException ex = assertThrows(TransactionException.class, tx::commit);
+ assertEquals("Transaction is already rolled back.", ex.getMessage());
+ }
+
+ @Test
+ void testCustomTransactionInterfaceThrows() {
+ var tx = new Transaction() {
+ @Override
+ public void commit() throws TransactionException {
+ }
+
+ @Override
+ public CompletableFuture<Void> commitAsync() {
+ return null;
+ }
+
+ @Override
+ public void rollback() throws TransactionException {
+ }
+
+ @Override
+ public CompletableFuture<Void> rollbackAsync() {
+ return null;
+ }
+ };
+
+ var ex = assertThrows(CompletionException.class, () -> kvView().put(tx, 1, "1"));
+
+ String expected = "Unsupported transaction implementation: "
+ + "'class org.apache.ignite.internal.runner.app.client.ItThinClientTransactionsTest";
+
+ assertThat(ex.getCause().getMessage(), startsWith(expected));
+ }
+
+ @Test
+ void testTransactionFromAnotherChannelThrows() throws Exception {
+ Transaction tx = client().transactions().begin();
+
+ try (IgniteClient client2 = IgniteClient.builder().addresses(getNodeAddress()).build()) {
+ RecordView<Tuple> recordView = client2.tables().tables().get(0).recordView();
+
+ var ex = assertThrows(CompletionException.class, () -> recordView.upsert(tx, Tuple.create()));
+ var iex = (IgniteClientException) ex.getCause();
+
+ assertEquals("Transaction context has been lost due to connection errors.", iex.getMessage());
+ }
+ }
+
+ private KeyValueView<Integer, String> kvView() {
+ return table().keyValueView(Mapper.of(Integer.class), Mapper.of(String.class));
+ }
+
+ private Table table() {
+ return client().tables().tables().get(0);
+ }
+
+ private Tuple val(String v) {
+ return Tuple.create().set(COLUMN_VAL, v);
+ }
+
+ private Tuple key(Integer k) {
+ return Tuple.create().set(COLUMN_KEY, k);
+ }
+
+ private Tuple kv(Integer k, String v) {
+ return Tuple.create().set(COLUMN_KEY, k).set(COLUMN_VAL, v);
+ }
+
+ private Rec rec(int key, String val) {
+ var r = new Rec();
+
+ r.key = key;
+ r.val = val;
+
+ return r;
+ }
+
+ private static class Rec {
+ public int key;
+ public String val;
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Rec rec = (Rec) o;
+ return key == rec.key && Objects.equals(val, rec.val);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(key, val);
+ }
+ }
+}
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 0bd161c..fc007f4 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -50,6 +50,7 @@ import org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.internal.table.distributed.TableTxManagerImpl;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.impl.HeapLockManager;
+import org.apache.ignite.internal.tx.impl.IgniteTransactionsImpl;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.VaultService;
import org.apache.ignite.internal.vault.persistence.PersistentVaultService;
@@ -221,6 +222,7 @@ public class IgniteImpl implements Ignite {
clientHandlerModule = new ClientHandlerModule(
qryEngine,
distributedTblMgr,
+ new IgniteTransactionsImpl(txManager),
nodeCfgMgr.configurationRegistry(),
nettyBootstrapFactory
);
@@ -356,7 +358,7 @@ public class IgniteImpl implements Ignite {
/** {@inheritDoc} */
@Override
public IgniteTransactions transactions() {
- return null;
+ return new IgniteTransactionsImpl(txManager);
}
/** {@inheritDoc} */
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteTransactionsImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteTransactionsImpl.java
index f9d34f6..6c40462 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteTransactionsImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteTransactionsImpl.java
@@ -18,13 +18,9 @@
package org.apache.ignite.internal.tx.impl;
import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.tx.IgniteTransactions;
import org.apache.ignite.tx.Transaction;
-import org.apache.ignite.tx.TransactionException;
/**
* Transactions facade implementation.
@@ -44,7 +40,8 @@ public class IgniteTransactionsImpl implements IgniteTransactions {
/** {@inheritDoc} */
@Override
public IgniteTransactions withTimeout(long timeout) {
- return null;
+ // TODO: IGNITE-15936.
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
@@ -58,35 +55,4 @@ public class IgniteTransactionsImpl implements IgniteTransactions {
public CompletableFuture<Transaction> beginAsync() {
return CompletableFuture.completedFuture(txManager.begin());
}
-
- /** {@inheritDoc} */
- @Override
- public void runInTransaction(Consumer<Transaction> clo) throws TransactionException {
- runInTransaction(tx -> {
- clo.accept(tx);
- return null;
- });
- }
-
- /** {@inheritDoc} */
- @Override
- public <T> T runInTransaction(Function<Transaction, T> clo) throws TransactionException {
- InternalTransaction tx = txManager.begin();
-
- try {
- T ret = clo.apply(tx);
-
- tx.commit();
-
- return ret;
- } catch (Throwable t) {
- try {
- tx.rollback(); // Try rolling back on user exception.
- } catch (Exception e) {
- t.addSuppressed(e);
- }
-
- throw t;
- }
- }
}