You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mp...@apache.org on 2026/02/18 18:18:56 UTC
(ignite-3) 01/01: IGNITE-27789 Remove file transfer module
This is an automated email from the ASF dual-hosted git repository.
mpochatkin pushed a commit to branch IGNITE-27789
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 4d0ad12c11bc0a1f899d382c6cb99542529ea19a
Author: Pochatkin Mikhail <m....@gmail.com>
AuthorDate: Wed Feb 18 21:18:34 2026 +0300
IGNITE-27789 Remove file transfer module
---
.../ignite/teamcity/ApacheIgnite3Teamcity.kt | 5 -
.../operations/handle-exceptions.md | 2 -
.../reference/node-configuration.md | 11 -
.../administrators-guide/config/node-config.md | 11 -
.../java/org/apache/ignite/lang/ErrorGroups.java | 7 +-
modules/file-transfer/README.md | 26 -
modules/file-transfer/build.gradle | 47 --
.../internal/network/file/ItFileTransferTest.java | 526 -----------------
.../ignite/internal/network/file/TestCluster.java | 193 ------
.../ignite/internal/network/file/Channel.java | 34 --
.../internal/network/file/ChunkedFileReader.java | 119 ----
.../internal/network/file/ChunkedFileWriter.java | 112 ----
.../network/file/FileChunkMessagesStream.java | 165 ------
.../ignite/internal/network/file/FileConsumer.java | 38 --
.../ignite/internal/network/file/FileProvider.java | 36 --
.../ignite/internal/network/file/FileReceiver.java | 196 ------
.../ignite/internal/network/file/FileSender.java | 266 ---------
.../file/FileTransferChannelTypeModule.java | 31 -
.../network/file/FileTransferMessagesHandler.java | 169 ------
.../internal/network/file/FileTransferService.java | 69 ---
.../network/file/FileTransferServiceImpl.java | 654 ---------------------
.../network/file/TransferredFilesCollector.java | 35 --
.../file/exception/FileTransferException.java | 46 --
.../file/exception/FileValidationException.java | 36 --
.../network/file/messages/FileChunkMessage.java | 56 --
.../network/file/messages/FileChunkResponse.java | 36 --
.../network/file/messages/FileDownloadRequest.java | 42 --
.../file/messages/FileDownloadResponse.java | 36 --
.../internal/network/file/messages/FileHeader.java | 68 ---
.../network/file/messages/FileTransferError.java | 66 ---
.../file/messages/FileTransferErrorMessage.java | 42 --
.../file/messages/FileTransferInitMessage.java | 50 --
.../file/messages/FileTransferInitResponse.java | 36 --
.../file/messages/FileTransferMessageType.java | 76 ---
.../internal/network/file/messages/Identifier.java | 31 -
.../network/file/ChunkedFileReaderTest.java | 86 ---
.../network/file/ChunkedFileWriterTest.java | 81 ---
.../network/file/FileChunkMessagesStreamTest.java | 121 ----
.../internal/network/file/FileReceiverTest.java | 174 ------
.../internal/network/file/FileSenderTest.java | 296 ----------
.../network/file/FileTransferServiceImplTest.java | 311 ----------
.../network/file/TestMessagingService.java | 92 ---
.../internal/network/file/TestTopologyService.java | 92 ---
.../internal/network/file/FileGenerator.java | 46 --
.../internal/network/file/PathAssertions.java | 44 --
.../FileTransferConfigurationSchema.java | 50 --
.../configuration/NetworkConfigurationSchema.java | 3 -
settings.gradle | 2 -
48 files changed, 2 insertions(+), 4769 deletions(-)
diff --git a/.teamcity/org/apache/ignite/teamcity/ApacheIgnite3Teamcity.kt b/.teamcity/org/apache/ignite/teamcity/ApacheIgnite3Teamcity.kt
index 9574c8e3b93..7aa24f96798 100644
--- a/.teamcity/org/apache/ignite/teamcity/ApacheIgnite3Teamcity.kt
+++ b/.teamcity/org/apache/ignite/teamcity/ApacheIgnite3Teamcity.kt
@@ -79,10 +79,6 @@ class ApacheIgnite3Teamcity {
"Examples",
"ignite-examples"
)
- private val IGNITE__FILE_TRANSFER = GradleModule(
- "File Transfer",
- "ignite-file-transfer"
- )
private val IGNITE__FILE_IO = GradleModule(
"File IO",
"ignite-file-io"
@@ -291,7 +287,6 @@ class ApacheIgnite3Teamcity {
IGNITE__CONFIGURATION_ANNOTATION_PROCESSOR,
IGNITE__DISTRIBUTION_ZONES,
IGNITE__EXAMPLES,
- IGNITE__FILE_TRANSFER,
IGNITE__INDEX,
IGNITE__JDBC,
IGNITE__METASTORAGE,
diff --git a/docs/docs/configure-and-operate/operations/handle-exceptions.md b/docs/docs/configure-and-operate/operations/handle-exceptions.md
index 73b47fa0016..fccc1714d00 100644
--- a/docs/docs/configure-and-operate/operations/handle-exceptions.md
+++ b/docs/docs/configure-and-operate/operations/handle-exceptions.md
@@ -165,8 +165,6 @@ When the exception happens, Apache Ignite provides a UUID of the specific except
|---|---|---|
| `IGN-NETWORK-1` | The node with the specified ID is not in the physical topology. | Check the error message and node ID. Update node ID if it is incorrect. If the node is offline, check why and bring it back online. |
| `IGN-NETWORK-2` | Port is already in use. | Most likely, Apache Ignite tried to access the port occupied by a different process. Change the port or free it, and then restart the node. |
-| `IGN-NETWORK-3` | File transfer error. | Check network connectivity and retry the file transfer. |
-| `IGN-NETWORK-4` | File validation error. | Ensure the file is not corrupted. |
| `IGN-NETWORK-5` | Recipient node has left the physical topology. | Check the error message. The node is unavailable and need to be brought back into the cluster. |
| `IGN-NETWORK-6` | Could not resolve address. Most likely, the IP address specified in the operation is not available locally. | Change the node configuration to make the address available or use a different IP address. |
diff --git a/docs/docs/configure-and-operate/reference/node-configuration.md b/docs/docs/configure-and-operate/reference/node-configuration.md
index a65c9a90c30..4576497fa3e 100644
--- a/docs/docs/configure-and-operate/reference/node-configuration.md
+++ b/docs/docs/configure-and-operate/reference/node-configuration.md
@@ -128,12 +128,6 @@ Example configuration with a `STATIC` node finder:
{
"ignite" : {
"network" : {
- "fileTransfer" : {
- "chunkSizeBytes" : 1048576,
- "maxConcurrentRequests" : 4,
- "responseTimeoutMillis" : 10000,
- "threadPoolSize" : 8
- },
"inbound" : {
"soBacklog" : 128,
"soKeepAlive" : true,
@@ -205,11 +199,6 @@ To switch to a `MULTICAST` node finder, update the `nodeFinder` section in your
| Property | Default | Description | Changeable | Requires Restart | Acceptable Values |
|----------|---------|-------------|------------|------------------|-------------------|
-| fileTransfer | | File transfer configuration. | | | |
-| fileTransfer.chunkSizeBytes | 1048576 | Chunk size in bytes. | Yes | Yes | 1 - Integer.MAX_VALUE |
-| fileTransfer.maxConcurrentRequests | 4 | Maximum number of concurrent requests. | Yes | Yes | 1 - Integer.MAX_VALUE |
-| fileTransfer.responseTimeoutMillis | 10000 | Node response timeout during file transfer. | Yes | Yes | 0 - inf |
-| fileTransfer.threadPoolSize | 8 | File sender thread pool size. | Yes | Yes | 1 - Integer.MAX_VALUE |
| inbound | | Server socket configuration. See [TCP documentation](https://man7.org/linux/man-pages/man7/tcp.7.html) and [socket documentation](https://man7.org/linux/man-pages/man7/socket.7.html) for more information. | | | |
| inbound.soBacklog | 128 | The size of the backlog. | Yes | Yes | 0 - Integer.MAX_VALUE |
| inbound.soKeepAlive | true | Defines if the keep-alive packets are allowed. | Yes | Yes | true, false |
diff --git a/docs/versioned_docs/version-3.0.0/administrators-guide/config/node-config.md b/docs/versioned_docs/version-3.0.0/administrators-guide/config/node-config.md
index 1d0af5464b5..f1840e1df35 100644
--- a/docs/versioned_docs/version-3.0.0/administrators-guide/config/node-config.md
+++ b/docs/versioned_docs/version-3.0.0/administrators-guide/config/node-config.md
@@ -146,12 +146,6 @@ In Apache Ignite 3, you can choose between two node discovery types. With `STATI
{
"ignite" : {
"network" : {
- "fileTransfer" : {
- "chunkSizeBytes" : 1048576,
- "maxConcurrentRequests" : 4,
- "responseTimeoutMillis" : 10000,
- "threadPoolSize" : 8
- },
"inbound" : {
"soBacklog" : 128,
"soKeepAlive" : true,
@@ -223,11 +217,6 @@ In Apache Ignite 3, you can choose between two node discovery types. With `STATI
| Property | Default | Description | Changeable | Requires Restart | Acceptable Values |
|----------|---------|-------------|------------|------------------|-------------------|
-| fileTransfer | | File transfer configuration. | | | |
-| fileTransfer.chunkSizeBytes | 1048576 | Chunk size in bytes. | Yes | Yes | 1 - Integer.MAX_VALUE |
-| fileTransfer.maxConcurrentRequests | 4 | Maximum number of concurrent requests. | Yes | Yes | 1 - Integer.MAX_VALUE |
-| fileTransfer.responseTimeoutMillis | 10000 | Node response timeout during file transfer. | Yes | Yes | 0 - inf |
-| fileTransfer.threadPoolSize | 8 | File sender thread pool size. | Yes | Yes | 1 - Integer.MAX_VALUE |
| inbound | | Server socket configuration. See [TCP documentation](https://man7.org/linux/man-pages/man7/tcp.7.html) and [socket documentation](https://man7.org/linux/man-pages/man7/socket.7.html) for more information. | | | |
| inbound.soBacklog | 128 | The size of the backlog. | Yes | Yes | 0 - Integer.MAX_VALUE |
| inbound.soKeepAlive | true | Defines if the keep-alive packets are allowed. | Yes | Yes | true, false |
diff --git a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index 25a7c7048ab..c0aa2ffb2cf 100755
--- a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -539,11 +539,8 @@ public class ErrorGroups {
/** Address or port bind error. */
public static final int BIND_ERR = NETWORK_ERR_GROUP.registerErrorCode((short) 2);
- /** File transfer error. */
- public static final int FILE_TRANSFER_ERR = NETWORK_ERR_GROUP.registerErrorCode((short) 3);
-
- /** File validation error. */
- public static final int FILE_VALIDATION_ERR = NETWORK_ERR_GROUP.registerErrorCode((short) 4);
+ // Error codes 3 (FILE_TRANSFER_ERR) and 4 (FILE_VALIDATION_ERR) were removed along with the file-transfer module.
+ // These codes are reserved forever and must not be reused.
/** Recipient node has left the physical topology. */
public static final int RECIPIENT_LEFT_ERR = NETWORK_ERR_GROUP.registerErrorCode((short) 5);
diff --git a/modules/file-transfer/README.md b/modules/file-transfer/README.md
deleted file mode 100644
index f5ff2855be5..00000000000
--- a/modules/file-transfer/README.md
+++ /dev/null
@@ -1,26 +0,0 @@
-# Ignite File Transfer
-
-This module provides a way to transfer files between the host and the target.
-
-## Usage
-
-The module provides the `org.apache.ignite.internal.network.file.FileTransferService` class that can be used to transfer files between the
-host and the target.
-
-### Identifier
-
-For all files that are going to be transferred, identifiers must be provided.
-The identifier is represented by the `org.apache.ignite.internal.network.file.messages.Identifier` class.
-The identifier allows fetching files from the target and uploading files to the target.
-
-### Downloading files from the target
-
-For downloading files from the target, the target should have a `FileTransferService` instance running
-and a registered `org.apache.ignite.internal.network.file.FileProvider` for the files that are to be downloaded.
-The `FileProvider` is a function that accepts an `Identifier` object and returns a collection of `File` objects.
-
-### Uploading files to the target
-
-For uploading files to the target, the target should have a `FileTransferService` instance running
-and a registered `org.apache.ignite.internal.network.file.FileConsumer` for the files that are to be uploaded.
-The `FileConsumer` is a function that accepts an `Identifier` object and a collection of `File` objects.
diff --git a/modules/file-transfer/build.gradle b/modules/file-transfer/build.gradle
deleted file mode 100644
index add56b3c0b0..00000000000
--- a/modules/file-transfer/build.gradle
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-apply from: "$rootDir/buildscripts/java-core.gradle"
-apply from: "$rootDir/buildscripts/publishing.gradle"
-apply from: "$rootDir/buildscripts/java-junit5.gradle"
-apply from: "$rootDir/buildscripts/java-integration-test.gradle"
-apply from: "$rootDir/buildscripts/java-test-fixtures.gradle"
-
-dependencies {
- annotationProcessor project(':ignite-network-annotation-processor')
- annotationProcessor libs.auto.service
-
- implementation project(':ignite-core')
- implementation project(':ignite-network')
- implementation project(':ignite-configuration-api')
- implementation libs.jetbrains.annotations
- implementation libs.auto.service.annotations
-
- testImplementation testFixtures(project)
- testImplementation testFixtures(project(':ignite-core'))
- testImplementation libs.awaitility
-
- integrationTestImplementation testFixtures(project)
- integrationTestImplementation testFixtures(project(':ignite-core'))
- integrationTestImplementation testFixtures(project(':ignite-network'))
- integrationTestImplementation testFixtures(project(':ignite-configuration'))
- integrationTestImplementation libs.awaitility
-
- testFixturesImplementation testFixtures(project(':ignite-core'))
-}
-
-description = 'ignite-file-transfer'
diff --git a/modules/file-transfer/src/integrationTest/java/org/apache/ignite/internal/network/file/ItFileTransferTest.java b/modules/file-transfer/src/integrationTest/java/org/apache/ignite/internal/network/file/ItFileTransferTest.java
deleted file mode 100644
index 692295fe286..00000000000
--- a/modules/file-transfer/src/integrationTest/java/org/apache/ignite/internal/network/file/ItFileTransferTest.java
+++ /dev/null
@@ -1,526 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static java.util.stream.Collectors.toList;
-import static org.apache.ignite.internal.network.file.FileGenerator.randomFile;
-import static org.apache.ignite.internal.network.file.PathAssertions.namesAndContentEquals;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowWithCauseOrSuppressed;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
-import static org.apache.ignite.internal.util.CompletableFutures.emptyListCompletedFuture;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-import static org.awaitility.Awaitility.await;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.emptyIterable;
-import static org.hamcrest.Matchers.not;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-import java.io.IOException;
-import java.nio.file.AccessDeniedException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.stream.Stream;
-import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
-import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.network.configuration.FileTransferConfiguration;
-import org.apache.ignite.internal.network.file.TestCluster.Node;
-import org.apache.ignite.internal.network.file.exception.FileTransferException;
-import org.apache.ignite.internal.network.file.messages.FileTransferFactory;
-import org.apache.ignite.internal.network.file.messages.Identifier;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInfo;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-/**
- * Integration tests for file transferring.
- */
-@ExtendWith(ConfigurationExtension.class)
-@ExtendWith(WorkDirectoryExtension.class)
-public class ItFileTransferTest extends BaseIgniteAbstractTest {
- private static final String DOWNLOADS_DIR = "downloads";
-
- @WorkDirectory
- private Path workDir;
-
- @InjectConfiguration("mock.chunkSizeBytes=1024")
- private FileTransferConfiguration configuration;
-
- private TestCluster cluster;
-
- private Path sourceDir;
-
- private final FileTransferFactory messageFactory = new FileTransferFactory();
-
- @BeforeEach
- void setUp(TestInfo testInfo) throws InterruptedException, IOException {
- sourceDir = Files.createDirectories(workDir.resolve("source"));
-
- cluster = new TestCluster(2, configuration, workDir, testInfo);
- cluster.startAwait();
- }
-
- @AfterEach
- void tearDown() throws Exception {
- cluster.shutdown();
- }
-
- @Test
- void downloadToEmptyDir() {
- // Generate files on the source node.
- Node sourceNode = cluster.members.get(0);
-
- int chunkSize = configuration.value().chunkSizeBytes();
- Path file1 = randomFile(sourceDir, chunkSize);
- Path file2 = randomFile(sourceDir, chunkSize + 1);
- Path file3 = randomFile(sourceDir, chunkSize * 2);
- Path file4 = randomFile(sourceDir, chunkSize * 2);
-
- // Add file provider to the source node.
- List<Path> files = List.of(file1, file2, file3, file4);
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> completedFuture(files)
- );
-
- // Download files on the target node from the source node.
- Node targetNode = cluster.members.get(1);
- CompletableFuture<List<Path>> downloadedFilesFuture = targetNode.fileTransferService().download(
- sourceNode.nodeName(),
- messageFactory.identifier().build(),
- targetNode.workDir().resolve(DOWNLOADS_DIR)
- );
-
- // Check that files were downloaded.
- assertThat(
- downloadedFilesFuture,
- willBe(namesAndContentEquals(files))
- );
-
- // Check that files were downloaded to the correct directory.
- assertDownloadsDirContainsFiles(targetNode, files);
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- void downloadToNonEmptyDir() throws IOException {
- // Generate files on the source node.
- Node sourceNode = cluster.members.get(0);
-
- int chunkSize = configuration.value().chunkSizeBytes();
- Path file1 = randomFile(sourceDir, chunkSize);
- Path file2 = randomFile(sourceDir, chunkSize + 1);
- Path file3 = randomFile(sourceDir, chunkSize * 2);
- Path file4 = randomFile(sourceDir, chunkSize * 2);
-
- // Add file provider to the source node.
- List<Path> files = List.of(file1, file2, file3, file4);
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> completedFuture(files)
- );
-
- Node targetNode = cluster.members.get(1);
-
- // Create downloads directory.
- Path targetNodeDownloadsDir = targetNode.workDir().resolve(DOWNLOADS_DIR);
- Files.createDirectories(targetNodeDownloadsDir);
-
- // Create some files in the downloads directory.
- Path file5 = randomFile(targetNodeDownloadsDir, chunkSize);
- Path file6 = randomFile(targetNodeDownloadsDir, chunkSize + 1);
-
- // Download files on the target node from the source node.
- CompletableFuture<List<Path>> downloadedFilesFuture = targetNode.fileTransferService().download(
- sourceNode.nodeName(),
- messageFactory.identifier().build(),
- targetNodeDownloadsDir
- );
-
- // Check that files were downloaded.
- assertThat(
- downloadedFilesFuture,
- willBe(namesAndContentEquals(files))
- );
-
- // Check that files were downloaded to the correct directory.
- assertDownloadsDirContainsFiles(targetNode, files);
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
-
- // Check that previous files in the downloads directory were deleted.
- try (Stream<Path> stream = Files.list(targetNodeDownloadsDir)) {
- List<String> filesInDownloadsDir = stream.map(Path::getFileName)
- .map(Path::toString)
- .collect(toList());
- assertThat(filesInDownloadsDir, not(containsInAnyOrder(file5.getFileName().toString(), file6.getFileName().toString())));
- }
- }
-
- @Test
- @DisabledOnOs(OS.WINDOWS) // Windows doesn't support posix file permissions.
- void downloadNonReadableFiles() {
- // Generate files on the source node.
- Node sourceNode = cluster.members.get(0);
-
- int chunkSize = configuration.value().chunkSizeBytes();
- Path file1 = randomFile(sourceDir, chunkSize - 1);
- Path file2 = randomFile(sourceDir, chunkSize + 1);
- Path file3 = randomFile(sourceDir, chunkSize * 2);
-
- // Make file2 non-readable.
- assertTrue(file2.toFile().setReadable(false));
-
- // Add file provider to the source node.
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> completedFuture(List.of(file1, file2, file3))
- );
-
- // Download files on the target node from the source node.
- Node targetNode = cluster.members.get(1);
- CompletableFuture<List<Path>> download = targetNode.fileTransferService().download(
- sourceNode.nodeName(),
- messageFactory.identifier().build(),
- targetNode.workDir().resolve(DOWNLOADS_DIR)
- );
-
- // Check that files transfer failed.
- assertThat(download, willThrow(FileTransferException.class, "Failed to create a file transfer stream"));
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- void downloadFilesWhenProviderThrowsException() {
- // Add file provider to the source node that throws an exception.
- Node sourceNode = cluster.members.get(0);
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> failedFuture(new RuntimeException("Test exception"))
- );
-
- // Download files on the target node from the source node.
- Node targetNode = cluster.members.get(1);
- CompletableFuture<List<Path>> download = targetNode.fileTransferService().download(
- sourceNode.nodeName(),
- messageFactory.identifier().build(),
- targetNode.workDir().resolve(DOWNLOADS_DIR)
- );
-
- // Check that files transfer failed.
- assertThat(download, willThrowWithCauseOrSuppressed(FileTransferException.class));
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- @DisabledOnOs(OS.WINDOWS) // Windows doesn't support posix file permissions.
- void downloadFilesWhenDoNotHaveAccessToWrite() {
- // Generate files on the source node.
- Node sourceNode = cluster.members.get(0);
-
- int chunkSize = configuration.value().chunkSizeBytes();
- Path file1 = randomFile(sourceDir, 0);
- Path file2 = randomFile(sourceDir, chunkSize + 1);
- Path file3 = randomFile(sourceDir, chunkSize * 2);
- Path file4 = randomFile(sourceDir, chunkSize * 2);
-
- // Add file provider to the source node.
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> completedFuture(List.of(file1, file2, file3, file4))
- );
-
- // Make work directory not writable on the target node.
- Node targetNode = cluster.members.get(1);
- assertTrue(targetNode.workDir().toFile().setWritable(false));
-
- // Download files on the target node from the source node.
- CompletableFuture<List<Path>> downloadedFilesFuture = targetNode.fileTransferService().download(
- sourceNode.nodeName(),
- messageFactory.identifier().build(),
- targetNode.workDir().resolve(DOWNLOADS_DIR)
- );
-
- // Check that files transfer failed.
- assertThat(
- downloadedFilesFuture,
- willThrowWithCauseOrSuppressed(AccessDeniedException.class)
- );
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- void downloadEmptyFileList() {
- // Set empty file list provider on the source node.
- Node sourceNode = cluster.members.get(0);
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> emptyListCompletedFuture()
- );
-
- // Download files on the target node from the source node.
- Node targetNode = cluster.members.get(1);
- CompletableFuture<List<Path>> downloadedFiles = targetNode.fileTransferService().download(
- sourceNode.nodeName(),
- messageFactory.identifier().build(),
- targetNode.workDir().resolve(DOWNLOADS_DIR)
-
- );
-
- // Check that files transfer failed.
- assertThat(downloadedFiles, willThrow(FileTransferException.class, "No files to download"));
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- void upload() {
- // Generate files on the source node.
- Node sourceNode = cluster.members.get(0);
-
- int chunkSize = configuration.value().chunkSizeBytes();
- Path file1 = randomFile(sourceDir, chunkSize);
- Path file2 = randomFile(sourceDir, chunkSize - 1);
- Path file3 = randomFile(sourceDir, chunkSize + 1);
- Path file4 = randomFile(sourceDir, chunkSize * 2);
-
- // Add file provider to the source node.
- List<Path> files = List.of(file1, file2, file3, file4);
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> completedFuture(files)
- );
-
- Node targetNode = cluster.members.get(1);
-
- // The future that will be completed when the consumer is called.
- CompletableFuture<Void> uploadedFilesFuture = new CompletableFuture<>();
- targetNode.fileTransferService().addFileConsumer(Identifier.class, ((identifier, uploadedFiles) -> {
- return uploadedFilesFuture.completeAsync(() -> {
- assertThat(uploadedFiles, namesAndContentEquals(files));
- return null;
- });
- }));
-
- // Upload files to the target node from the source node.
- CompletableFuture<Void> upload = sourceNode.fileTransferService()
- .upload(targetNode.nodeName(), messageFactory.identifier().build());
-
- // Check that transfer was successful.
- assertThat(
- upload,
- CompletableFutureMatcher.willCompleteSuccessfully()
- );
-
- // Check the consumer was called.
- assertThat(
- uploadedFilesFuture,
- CompletableFutureMatcher.willCompleteSuccessfully()
- );
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- void uploadEmptyFileList() {
- // Set empty file list provider on the source node.
- Node sourceNode = cluster.members.get(0);
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> emptyListCompletedFuture()
- );
-
- // Upload files to the target node from the source node.
- Node targetNode = cluster.members.get(1);
- CompletableFuture<Void> uploaded = sourceNode.fileTransferService()
- .upload(targetNode.nodeName(), messageFactory.identifier().build());
-
- // Check that files transfer failed.
- assertThat(
- uploaded,
- willThrow(FileTransferException.class, "No files to upload")
- );
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- @DisabledOnOs(OS.WINDOWS) // Windows doesn't support posix file permissions.
- void uploadNonReadableFiles() {
- // Generate files on the source node.
- Node sourceNode = cluster.members.get(0);
-
- int chunkSize = configuration.value().chunkSizeBytes();
- Path file1 = randomFile(sourceDir, chunkSize - 1);
- Path file2 = randomFile(sourceDir, chunkSize + 1);
- Path file3 = randomFile(sourceDir, chunkSize * 2);
-
- // Make file2 non-readable.
- assertTrue(file2.toFile().setReadable(false));
-
- // Add file provider to the source node.
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> completedFuture(List.of(file1, file2, file3))
- );
-
- // Set file consumer on the target node.
- Node targetNode = cluster.members.get(1);
-
- // The future will be completed when the file consumer is called.
- CompletableFuture<List<Path>> uploadedFilesFuture = new CompletableFuture<>();
- targetNode.fileTransferService().addFileConsumer(Identifier.class, ((identifier, uploadedFiles) -> {
- uploadedFilesFuture.complete(uploadedFiles);
- return nullCompletedFuture();
- }));
-
- // Upload files to the target node from the source node.
- Identifier identifier = messageFactory.identifier().build();
- CompletableFuture<Void> upload = sourceNode.fileTransferService().upload(targetNode.nodeName(), identifier);
-
- // Check that files transfer failed.
- assertThat(
- upload,
- willThrow(FileTransferException.class, "Failed to create a file transfer stream")
- );
-
- // Check the consumer was not called.
- assertFalse(uploadedFilesFuture.isDone());
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- void uploadFilesWhenProviderThrowsException() {
- // Set file provider on the source node that throws an exception.
- Node sourceNode = cluster.members.get(0);
-
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> failedFuture(new RuntimeException("Test exception"))
- );
-
- // Set file consumer on the target node.
- Node targetNode = cluster.members.get(1);
-
- // The future will be completed when the file consumer is called.
- CompletableFuture<List<Path>> uploadedFilesFuture = new CompletableFuture<>();
- targetNode.fileTransferService().addFileConsumer(Identifier.class, ((identifier, uploadedFiles) -> {
- uploadedFilesFuture.complete(uploadedFiles);
- return nullCompletedFuture();
- }));
-
- // Upload files to the target node from the source node.
- CompletableFuture<Void> uploaded = sourceNode.fileTransferService()
- .upload(targetNode.nodeName(), messageFactory.identifier().build());
-
- // Check that files transfer failed.
- assertThat(
- uploaded,
- willThrow(RuntimeException.class, "Test exception")
- );
-
- // Check the consumer was not called.
- assertFalse(uploadedFilesFuture.isDone());
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- @Test
- @DisabledOnOs(OS.WINDOWS) // Windows doesn't support posix file permissions.
- void uploadFilesWhenDoNotHaveAccessToWrite() {
- // Generate files on the source node.
- Node sourceNode = cluster.members.get(0);
-
- int chunkSize = configuration.value().chunkSizeBytes();
- Path file1 = randomFile(sourceDir, 0);
- Path file2 = randomFile(sourceDir, chunkSize + 1);
- Path file3 = randomFile(sourceDir, chunkSize * 2);
- Path file4 = randomFile(sourceDir, chunkSize * 2);
-
- // Add file provider to the source node.
- sourceNode.fileTransferService().addFileProvider(
- Identifier.class,
- req -> completedFuture(List.of(file1, file2, file3, file4))
- );
-
- // Make work directory not writable on the target node.
- Node targetNode = cluster.members.get(1);
- assertTrue(targetNode.workDir().toFile().setWritable(false));
-
- // Upload files to the target node from the source node.
- CompletableFuture<Void> uploaded = sourceNode.fileTransferService().upload(
- targetNode.nodeName(), messageFactory.identifier().build()
- );
-
- // Check that files transfer failed.
- assertThat(uploaded, willThrow(FileTransferException.class, "Failed to upload files:"));
-
- // Check temporary files were deleted.
- assertTemporaryFilesWereDeleted(targetNode);
- }
-
- private static void assertTemporaryFilesWereDeleted(Node targetNode) {
- await().until(() -> {
- try (Stream<Path> stream = Files.list(targetNode.workDir())) {
- return stream.map(Path::getFileName)
- .map(Path::toString)
- .filter(path -> !path.equals(DOWNLOADS_DIR))
- .collect(toList());
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }, emptyIterable());
- }
-
- private static void assertDownloadsDirContainsFiles(Node node, List<Path> expectedFiles) {
- try (Stream<Path> stream = Files.list(node.workDir().resolve(DOWNLOADS_DIR))) {
- assertThat(stream.collect(toList()), namesAndContentEquals(expectedFiles));
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-}
diff --git a/modules/file-transfer/src/integrationTest/java/org/apache/ignite/internal/network/file/TestCluster.java b/modules/file-transfer/src/integrationTest/java/org/apache/ignite/internal/network/file/TestCluster.java
deleted file mode 100644
index 8cbf8e1ac03..00000000000
--- a/modules/file-transfer/src/integrationTest/java/org/apache/ignite/internal/network/file/TestCluster.java
+++ /dev/null
@@ -1,193 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static java.util.stream.Collectors.toUnmodifiableList;
-import static org.apache.ignite.internal.network.utils.ClusterServiceTestUtils.clusterService;
-import static org.apache.ignite.internal.network.utils.ClusterServiceTestUtils.findLocalAddresses;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
-import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
-import static org.apache.ignite.internal.util.IgniteUtils.startAsync;
-import static org.apache.ignite.internal.util.IgniteUtils.stopAsync;
-import static org.hamcrest.MatcherAssert.assertThat;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Stream;
-import org.apache.ignite.internal.manager.ComponentContext;
-import org.apache.ignite.internal.manager.IgniteComponent;
-import org.apache.ignite.internal.network.ClusterService;
-import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.network.NodeFinder;
-import org.apache.ignite.internal.network.StaticNodeFinder;
-import org.apache.ignite.internal.network.TopologyEventHandler;
-import org.apache.ignite.internal.network.configuration.FileTransferConfiguration;
-import org.apache.ignite.network.NetworkAddress;
-import org.junit.jupiter.api.TestInfo;
-
-/**
- * Test cluster.
- */
-public class TestCluster {
- /** Startup timeout in seconds. */
- private static final int STARTUP_TIMEOUT = 3;
-
- /** Members of the cluster. */
- final List<Node> members;
-
- /** Latch that is locked until all members are visible in the topology. */
- private final CountDownLatch startupLatch;
-
- /** Node finder. */
- private final NodeFinder nodeFinder;
-
- private final FileTransferConfiguration configuration;
-
- /**
- * Creates a test cluster with the given amount of members.
- *
- * @param numOfNodes Amount of cluster members.
- * @param testInfo Test info.
- */
- TestCluster(int numOfNodes, FileTransferConfiguration configuration, Path workDir, TestInfo testInfo) {
- this.startupLatch = new CountDownLatch(numOfNodes - 1);
-
- int initialPort = 3344;
-
- List<NetworkAddress> addresses = findLocalAddresses(initialPort, initialPort + numOfNodes);
-
- this.nodeFinder = new StaticNodeFinder(addresses);
- this.configuration = configuration;
-
- var isInitial = new AtomicBoolean(true);
-
- this.members = addresses.stream()
- .map(addr -> startNode(workDir, testInfo, addr, isInitial.getAndSet(false)))
- .collect(toUnmodifiableList());
- }
-
- /**
- * Start cluster node.
- *
- * @param testInfo Test info.
- * @param addr Node address.
- * @param initial Whether this node is the first one.
- * @return Started cluster node.
- */
- private Node startNode(
- Path workDir, TestInfo testInfo, NetworkAddress addr, boolean initial
- ) {
- ClusterService clusterSvc = clusterService(testInfo, addr.port(), nodeFinder);
-
- if (initial) {
- clusterSvc.topologyService().addEventHandler(new TopologyEventHandler() {
- @Override
- public void onAppeared(InternalClusterNode member) {
- startupLatch.countDown();
- }
- });
- }
-
- try {
- Path nodeDir = Files.createDirectory(workDir.resolve("node-" + clusterSvc.nodeName()));
- FileTransferServiceImpl fileTransferringService = new FileTransferServiceImpl(
- clusterSvc.nodeName(),
- clusterSvc.topologyService(),
- clusterSvc.messagingService(),
- configuration,
- nodeDir
- );
- return new Node(nodeDir, clusterSvc, fileTransferringService);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * Starts and waits for the cluster to come up.
- *
- * @throws InterruptedException If failed.
- * @throws AssertionError If the cluster was unable to start in 3 seconds.
- */
- void startAwait() throws InterruptedException {
- members.forEach(Node::start);
-
- if (!startupLatch.await(STARTUP_TIMEOUT, TimeUnit.SECONDS)) {
- throw new AssertionError("Cluster was unable to start in " + STARTUP_TIMEOUT + " seconds");
- }
- }
-
- /**
- * Stops the cluster.
- */
- void shutdown() throws Exception {
- closeAll(members.stream().map(it -> it::stop));
- }
-
- /**
- * Cluster node.
- */
- public static class Node {
- private final Path workDir;
- private final ClusterService clusterService;
- private final FileTransferService fileTransferService;
- private final List<IgniteComponent> components;
-
- /**
- * Constructor.
- *
- * @param workDir Work directory.
- * @param clusterService Cluster service.
- * @param fileTransferService File transferring service.
- */
- public Node(Path workDir, ClusterService clusterService, FileTransferService fileTransferService) {
- this.workDir = workDir;
- this.clusterService = clusterService;
- this.fileTransferService = fileTransferService;
- this.components = List.of(clusterService, fileTransferService);
- }
-
- public String nodeName() {
- return clusterService.nodeName();
- }
-
- public Path workDir() {
- return workDir;
- }
-
- public FileTransferService fileTransferService() {
- return fileTransferService;
- }
-
- void start() {
- assertThat(startAsync(new ComponentContext(), components), willCompleteSuccessfully());
- }
-
- void stop() throws Exception {
- closeAll(Stream.concat(
- components.stream().map(c -> c::beforeNodeStop),
- Stream.of(() -> assertThat(stopAsync(new ComponentContext(), components), willCompleteSuccessfully()))
- ));
- }
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/Channel.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/Channel.java
deleted file mode 100644
index 2a7a395315b..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/Channel.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import org.apache.ignite.internal.network.ChannelType;
-
-/**
- * Channel types used by the file transfer protocol.
- */
-final class Channel {
- /**
- * File transfer channel.
- */
- static final ChannelType FILE_TRANSFER_CHANNEL = new ChannelType((short) 3, "FileTransfer");
-
- private Channel() {
- // No-op.
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/ChunkedFileReader.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/ChunkedFileReader.java
deleted file mode 100644
index 88cef4c2cee..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/ChunkedFileReader.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.io.BufferedInputStream;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.nio.file.Files;
-
-/**
- * Chunked file reader. Reads the file in chunks. Each chunk has a fixed size. The last chunk may be smaller than the chunk size. If the
- * file size is less than the chunk size, only one chunk will be read. The reader is not thread-safe.
- */
-class ChunkedFileReader implements AutoCloseable {
- private final int chunkSize;
-
- private final BufferedInputStream stream;
-
- private int nextChunkNumber = 0;
-
- private boolean hasNextChunk;
-
- /**
- * Constructor.
- *
- * @param chunkSize Chunk size.
- * @param stream Random access file.
- */
- private ChunkedFileReader(int chunkSize, BufferedInputStream stream) throws IOException {
- this.chunkSize = chunkSize;
- this.stream = stream;
- this.hasNextChunk = stream.available() > 0;
- }
-
- /**
- * Opens a file for reading.
- *
- * @param file File.
- * @param chunkSize Chunk size.
- * @return Chunked file reader.
- * @throws FileNotFoundException If the file does not exist.
- */
- static ChunkedFileReader open(File file, int chunkSize) throws IOException {
- return new ChunkedFileReader(chunkSize, new BufferedInputStream(Files.newInputStream(file.toPath())));
- }
-
- /**
- * Returns {@code false} if there are no more chunks to read. Otherwise, returns {@code true}. Does not change the state of the reader.
- *
- * @return {@code false} if there are no more chunks to read. Otherwise, returns {@code true}.
- */
- boolean hasNextChunk() {
- return hasNextChunk;
- }
-
- /**
- * Reads the next chunk. If there are no more chunks to read, throws an exception. If the last chunk is read successfully, closes the
- * file.
- *
- * @return Chunk data.
- * @throws IOException If an I/O error occurs.
- */
- byte[] readNextChunk() throws IOException {
- if (!hasNextChunk()) {
- throw new IOException("No more chunks to read");
- }
-
- byte[] data = stream.readNBytes(chunkSize);
-
- nextChunkNumber++;
- hasNextChunk = stream.available() > 0;
-
- if (!hasNextChunk) {
- stream.close();
- }
-
- return data;
- }
-
- /**
- * Returns the number of the next chunk to read. Does not change the state of the reader.
- *
- * @return The number of the next chunk to read.
- * @throws IOException If there are no more chunks to read.
- */
- int nextChunkNumber() throws IOException {
- if (!hasNextChunk()) {
- throw new IOException("No more chunks to read");
- }
-
- return nextChunkNumber;
- }
-
- /**
- * Closes the file.
- *
- * @throws IOException If an I/O error occurs.
- */
- @Override
- public void close() throws IOException {
- stream.close();
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/ChunkedFileWriter.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/ChunkedFileWriter.java
deleted file mode 100644
index 30653709a70..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/ChunkedFileWriter.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.io.BufferedOutputStream;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import org.apache.ignite.internal.network.file.exception.FileValidationException;
-import org.apache.ignite.internal.network.file.messages.FileChunkMessage;
-
-/**
- * Chunked file writer. Writes chunks to a file. Checks that the file size and chunk numbers are valid.
- */
-class ChunkedFileWriter implements AutoCloseable {
- private final BufferedOutputStream stream;
-
- private final long expectedFileLength;
-
- private long bytesWritten = 0;
-
- private int expectedNextChunkNumber = 0;
-
- /**
- * Lock to synchronize access to the file. We don't write to the file simultaneously, but the next chunk number and bytes written can be
- * updated from different threads.
- */
- private final Lock lock = new ReentrantLock();
-
- private ChunkedFileWriter(BufferedOutputStream stream, long expectedFileLength) {
- this.stream = stream;
- this.expectedFileLength = expectedFileLength;
- }
-
- /**
- * Opens a file with known size for writing.
- *
- * @param file File path.
- * @param expectedFileLength Expected file size.
- * @return Chunked file writer.
- * @throws FileNotFoundException If the file is not found.
- */
- static ChunkedFileWriter open(File file, long expectedFileLength) throws IOException {
- return new ChunkedFileWriter(new BufferedOutputStream(Files.newOutputStream(file.toPath())), expectedFileLength);
- }
-
- /**
- * Writes a chunk to the file. The chunk number must be equal to the next expected chunk number. Closes the file if the last chunk is
- * written. Returns true if the last chunk is written.
- *
- * @param chunk Chunk to write.
- * @return True if the last chunk is written. False otherwise.
- * @throws IOException If an I/O error occurs.
- * @throws FileValidationException If the chunk number or file size is invalid or expected file size is exceeded.
- */
- boolean write(FileChunkMessage chunk) throws IOException {
- lock.lock();
- try {
- if (chunk.number() != expectedNextChunkNumber) {
- throw new FileValidationException(
- "Chunk number mismatch: expected " + expectedNextChunkNumber + ", actual " + chunk.number()
- );
- }
-
- stream.write(chunk.data());
- expectedNextChunkNumber++;
- bytesWritten += chunk.data().length;
-
- if (bytesWritten > expectedFileLength) {
- throw new FileValidationException("File size mismatch: expected " + expectedFileLength + ", actual " + bytesWritten);
- }
-
- if (bytesWritten == expectedFileLength) {
- close();
-
- return true;
- } else {
- return false;
- }
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Closes the file.
- *
- * @throws IOException If an I/O error occurs.
- */
- @Override
- public void close() throws IOException {
- stream.close();
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileChunkMessagesStream.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileChunkMessagesStream.java
deleted file mode 100644
index 39972007e8d..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileChunkMessagesStream.java
+++ /dev/null
@@ -1,165 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.nio.file.Path;
-import java.util.Iterator;
-import java.util.UUID;
-import java.util.function.Consumer;
-import org.apache.ignite.internal.network.file.messages.FileChunkMessage;
-import org.apache.ignite.internal.network.file.messages.FileTransferFactory;
-
-/**
- * A stream of {@link FileChunkMessage} to send file.
- */
-public class FileChunkMessagesStream implements Iterable<FileChunkMessage>, AutoCloseable {
- private final UUID transferId;
-
- private final Path path;
-
- private final ChunkedFileReader reader;
-
- private boolean closed = false;
-
- private final FileTransferFactory messageFactory = new FileTransferFactory();
-
- /**
- * Creates a new stream of messages to send files.
- *
- * @param transferId the id of the stream.
- * @param reader the reader of the file to send.
- */
- private FileChunkMessagesStream(
- UUID transferId,
- Path path,
- ChunkedFileReader reader
- ) {
- this.transferId = transferId;
- this.path = path;
- this.reader = reader;
- }
-
- /**
- * Creates a new stream of messages to send file.
- *
- * @param chunkSize the size of the chunks to send.
- * @param transferId the id of the transfer.
- * @param path the path of the file to send.
- * @return a new stream of messages to send files.
- * @throws IOException if an I/O error occurs.
- */
- public static FileChunkMessagesStream fromPath(
- int chunkSize,
- UUID transferId,
- Path path
- ) throws IOException {
- if (chunkSize <= 0) {
- throw new IllegalArgumentException("Chunk size must be positive");
- }
-
- return new FileChunkMessagesStream(transferId, path, ChunkedFileReader.open(path.toFile(), chunkSize));
- }
-
- /**
- * Returns true if there are more messages to send.
- *
- * @return true if there are more messages to send.
- */
- boolean hasNextMessage() throws IOException {
- // Check that the stream is not closed and the reader has more chunks.
- return !closed && reader.hasNextChunk();
- }
-
- /**
- * Returns the next message to send.
- *
- * @return the next message to send.
- * @throws IOException if an I/O error occurs.
- * @throws IllegalStateException if there are no more messages to send.
- */
- FileChunkMessage nextMessage() throws IOException {
- if (hasNextMessage()) {
- return nextChunk();
- } else {
- throw new IllegalStateException("There are no more messages to send");
- }
- }
-
- /**
- * Returns the next chunk of the current file. Throws an exception if the current file is finished.
- *
- * @return the next chunk of the current file.
- * @throws IOException if an I/O error occurs.
- * @throws IllegalStateException if the current file is finished.
- */
- private FileChunkMessage nextChunk() throws IOException {
- return messageFactory.fileChunkMessage()
- .transferId(transferId)
- .fileName(path.getFileName().toString())
- .number(reader.nextChunkNumber())
- .data(reader.readNextChunk())
- .build();
- }
-
- public Path path() {
- return path;
- }
-
- /**
- * Closes the stream.
- */
- @Override
- public void close() throws IOException {
- if (!closed) {
- closed = true;
- reader.close();
- }
- }
-
- @Override
- public Iterator<FileChunkMessage> iterator() {
- return new Iterator<>() {
- @Override
- public boolean hasNext() {
- try {
- return hasNextMessage();
- } catch (IOException e) {
- throw new UncheckedIOException(e);
- }
- }
-
- @Override
- public FileChunkMessage next() {
- try {
- return nextMessage();
- } catch (IOException e) {
- throw new UncheckedIOException(e);
- }
- }
- };
- }
-
- @Override
- public void forEach(Consumer<? super FileChunkMessage> action) {
- for (FileChunkMessage message : this) {
- action.accept(message);
- }
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileConsumer.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileConsumer.java
deleted file mode 100644
index 62dbbbd2a79..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileConsumer.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.network.file.messages.Identifier;
-
-/**
- * Consumes the uploaded files.
- */
-public interface FileConsumer<I extends Identifier> {
- /**
- * Consumes the list of paths to the uploaded files. The paths are temporary and will be deleted after the method returns. If there is a
- * need to keep the files, they should be copied to a different location.
- *
- * @param identifier Uploaded files identifier.
- * @param uploadedFiles List of paths to the uploaded files.
- * @return Future that will be completed when the consumption is finished.
- */
- CompletableFuture<Void> consume(I identifier, List<Path> uploadedFiles);
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileProvider.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileProvider.java
deleted file mode 100644
index 7df505a20f0..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileProvider.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.network.file.messages.Identifier;
-
-/**
- * Files provider.
- */
-public interface FileProvider<I extends Identifier> {
- /**
- * Returns a collection of paths to files for the given identifier.
- *
- * @param identifier Files identifier.
- * @return Collection of paths to files.
- */
- CompletableFuture<List<Path>> files(I identifier);
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileReceiver.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileReceiver.java
deleted file mode 100644
index 5a02a52fdfe..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileReceiver.java
+++ /dev/null
@@ -1,196 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.nio.file.Path;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Supplier;
-import org.apache.ignite.internal.network.file.exception.FileTransferException;
-import org.apache.ignite.internal.network.file.messages.FileChunkMessage;
-import org.apache.ignite.internal.network.file.messages.FileHeader;
-import org.apache.ignite.internal.util.RefCountedObjectPool;
-
-/**
- * File receiver.
- */
-class FileReceiver {
- /**
- * Map from transfer id to file transfer handler.
- */
- private final Map<UUID, FileTransferMessagesHandler> transferIdToHandler = new ConcurrentHashMap<>();
-
- /**
- * Map from sender consistent id to transfer ids.
- */
- private final Map<String, Set<UUID>> senderConsistentIdToTransferIds = new ConcurrentHashMap<>();
-
- /**
- * Map from sender consistent id to lock.
- */
- private final RefCountedObjectPool<String, ReentrantLock> senderConsistentIdToLock = new RefCountedObjectPool<>();
-
- /**
- * Registers file transfer.
- *
- * @param senderConsistentId Sender consistent id.
- * @param transferId Transfer id.
- * @return Future that will be completed when file transfer is finished.
- */
- TransferredFilesCollector registerTransfer(
- String senderConsistentId,
- UUID transferId,
- List<FileHeader> headers,
- Path handlerDir
- ) {
- return doInLock(senderConsistentId, () -> registerTransfer0(senderConsistentId, transferId, headers, handlerDir));
- }
-
- private TransferredFilesCollector registerTransfer0(
- String senderConsistentId,
- UUID transferId,
- List<FileHeader> headers,
- Path handlerDir
- ) {
- FileTransferMessagesHandler handler = new FileTransferMessagesHandler(headers, handlerDir);
-
- transferIdToHandler.put(transferId, handler);
-
- senderConsistentIdToTransferIds.compute(senderConsistentId, (k, v) -> {
- if (v == null) {
- v = new HashSet<>();
- }
- v.add(transferId);
- return v;
- });
-
- // De-register transfer on completion.
- handler.collectedFiles().whenComplete((files, throwable) -> deregisterTransfer(senderConsistentId, transferId));
-
- return handler;
- }
-
- /**
- * De-registers file transfer.
- *
- * @param senderConsistentId Sender consistent id.
- * @param transferId Transfer id.
- */
- private void deregisterTransfer(String senderConsistentId, UUID transferId) {
- doInLock(senderConsistentId, () -> deregisterTransfer0(senderConsistentId, transferId));
- }
-
- private void deregisterTransfer0(String senderConsistentId, UUID transferId) {
- transferIdToHandler.remove(transferId);
-
- senderConsistentIdToTransferIds.compute(senderConsistentId, (k, v) -> {
- if (v != null) {
- v.remove(transferId);
- if (v.isEmpty()) {
- return null;
- }
- }
- return v;
- });
- }
-
- /**
- * Cancels all transfers from sender.
- *
- * @param senderConsistentId Sender consistent id.
- */
- void cancelTransfersFromSender(String senderConsistentId) {
- doInLock(senderConsistentId, () -> cancelTransfersFromSender0(senderConsistentId));
- }
-
- private void cancelTransfersFromSender0(String senderConsistentId) {
- Set<UUID> uuids = senderConsistentIdToTransferIds.remove(senderConsistentId);
- if (uuids != null) {
- uuids.forEach(uuid -> {
- transferIdToHandler.get(uuid).handleFileTransferError(new FileTransferException("Transfer was cancelled"));
- });
- }
- }
-
- /**
- * Cancels transfer.
- *
- * @param transferId Transfer id.
- * @param error Error message.
- */
- void cancelTransfer(UUID transferId, Throwable error) {
- FileTransferMessagesHandler handler = transferIdToHandler.get(transferId);
- if (handler == null) {
- throw new FileTransferException("Handler is not found for unknown transferId: " + transferId);
- } else {
- handler.handleFileTransferError(error);
- }
- }
-
- /**
- * Receives file chunk.
- *
- * @param chunk File chunk.
- */
- void receiveFileChunk(FileChunkMessage chunk) {
- FileTransferMessagesHandler handler = transferIdToHandler.get(chunk.transferId());
- if (handler == null) {
- throw new FileTransferException("Handler is not found for unknown transferId: " + chunk.transferId());
- } else {
- handler.handleFileChunk(chunk);
- }
- }
-
- /**
- * Acquires lock for sender consistent id and executes supplier.
- *
- * @param senderConsistentId Sender consistent id.
- * @param supplier Supplier.
- */
- private <V> V doInLock(String senderConsistentId, Supplier<V> supplier) {
- Lock lock = senderConsistentIdToLock.acquire(senderConsistentId, ignored -> new ReentrantLock());
- lock.lock();
- try {
- return supplier.get();
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Acquires lock for sender consistent id and executes runnable.
- *
- * @param senderConsistentId Sender consistent id.
- * @param runnable Runnable.
- */
- private void doInLock(String senderConsistentId, Runnable runnable) {
- Lock lock = senderConsistentIdToLock.acquire(senderConsistentId, ignored -> new ReentrantLock());
- lock.lock();
- try {
- runnable.run();
- } finally {
- lock.unlock();
- }
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileSender.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileSender.java
deleted file mode 100644
index 43ed3ec1066..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileSender.java
+++ /dev/null
@@ -1,266 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static java.util.concurrent.CompletableFuture.allOf;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static java.util.concurrent.CompletableFuture.supplyAsync;
-import static org.apache.ignite.internal.network.file.Channel.FILE_TRANSFER_CHANNEL;
-import static org.apache.ignite.internal.network.file.messages.FileTransferError.toException;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-
-import java.io.IOException;
-import java.nio.file.Path;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Queue;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-import org.apache.ignite.internal.network.MessagingService;
-import org.apache.ignite.internal.network.file.exception.FileTransferException;
-import org.apache.ignite.internal.network.file.messages.FileChunkResponse;
-
-/**
- * A class that sends files to a node. It uses a rate limiter to limit the bandwidth used. It also uses a thread pool to send the files in
- * parallel.
- */
-class FileSender {
- /**
- * The size of the chunks to send.
- */
- private final int chunkSize;
-
- /**
- * Limits the bandwidth used to send files. The rate limiter is acquired for each file. If the rate limiter is not available, the file
- * is placed in the {@link FileSender#queue}.
- */
- private final Semaphore rateLimiter;
-
- /**
- * The timeout for the response from the node.
- */
- private final long responseTimeout;
-
- /**
- * The messaging service used to send the files.
- */
- private final MessagingService messagingService;
-
- /**
- * The thread pool used to send the files in parallel.
- */
- private final ExecutorService executorService;
-
- /**
- * A queue of files to send. The files are placed in the queue when the rate limiter is not available. The files are sent when the rate
- * limiter is available.
- */
- private final Queue<FileTransfer> queue = new LinkedList<>();
-
- /**
- * We need this lock to avoid stuck transfers in the queue, when one thread has a permit and sees the queue is empty and another thread
- * can't acquire a permit and places a transfer in the queue. As a result, the first thread will release the permit and the transfer
- * will not be sent until the next file transfer.
- */
- private final Object lock = new Object();
-
- FileSender(
- int chunkSize,
- Semaphore rateLimiter,
- long responseTimeout,
- MessagingService messagingService,
- ExecutorService executorService
- ) {
- this.chunkSize = chunkSize;
- this.rateLimiter = rateLimiter;
- this.responseTimeout = responseTimeout;
- this.messagingService = messagingService;
- this.executorService = executorService;
- }
-
- /**
- * Creates {@link FileTransfer} objects for each file and places them in the queue. The files will be sent when the rate limiter is
- * available.
- *
- * @param targetNodeConsistentId The consistent ID of the node to send the files to.
- * @param transferId The ID of the transfer.
- * @param paths The paths of the files to send.
- * @return A future that will be completed when the transfer is complete.
- */
- CompletableFuture<Void> send(String targetNodeConsistentId, UUID transferId, List<Path> paths) {
- // It doesn't make sense to continue file transfer if there is a failure in one of the files.
- AtomicBoolean shouldBeCancelled = new AtomicBoolean(false);
-
- List<FileTransfer> transfers = paths.stream()
- .map(path -> new FileTransfer(targetNodeConsistentId, transferId, path, shouldBeCancelled))
- .collect(Collectors.toList());
-
- CompletableFuture[] results = transfers.stream()
- .map(this::processTransferAsync)
- .map(it -> it.whenComplete((v, e) -> {
- if (e != null) {
- shouldBeCancelled.set(true);
- }
- }))
- .toArray(CompletableFuture[]::new);
-
- return allOf(results);
- }
-
- /**
- * Processes the given transfer. If the rate limiter is not available, the transfer will be processed later.
- *
- * @param transfer The transfer to process.
- */
- private CompletableFuture<Void> processTransferAsync(FileTransfer transfer) {
- synchronized (lock) {
- if (rateLimiter.tryAcquire()) {
- processTransferWithNextAsync(transfer);
- } else {
- queue.add(transfer);
- }
- }
- return transfer.result;
- }
-
- /**
- * Process the given transfer and then process the next transfer if there is a next transfer.
- *
- * @return A future that will be completed when the transfer is complete and queued transfers are processed.
- */
- private CompletableFuture<Void> processTransferWithNextAsync(FileTransfer transfer) {
- return sendTransfer(transfer)
- .thenComposeAsync(v -> {
- synchronized (lock) {
- FileTransfer nextTransfer = queue.poll();
-
- // If there is a next transfer, process it.
- // Otherwise, release the rate limiter.
- if (nextTransfer != null) {
- return processTransferWithNextAsync(nextTransfer);
- } else {
- rateLimiter.release();
- return nullCompletedFuture();
- }
- }
- }, executorService);
- }
-
- /**
- * Sends the file to the node with the given consistent id.
- *
- * @param transfer The transfer to send.
- * @return A future that will be completed when the file is sent. The future will be completed successfully always, even if the file is
- * not sent.
- */
- private CompletableFuture<Void> sendTransfer(FileTransfer transfer) {
- return sendFile(transfer.receiverConsistentId, transfer.transferId, transfer.path, transfer.shouldBeCancelled)
- .handle((v, e) -> {
- if (e == null) {
- transfer.result.complete(null);
- } else {
- transfer.result.completeExceptionally(e);
- }
- return null;
- });
- }
-
- /**
- * Sends the file to the node with the given consistent id.
- *
- * @param receiverConsistentId The consistent id of the node to send the file to.
- * @param id The id of the file transfer.
- * @param path The path of the file to send.
- * @return A future that will be completed when the file is sent.
- */
- private CompletableFuture<Void> sendFile(String receiverConsistentId, UUID id, Path path, AtomicBoolean shouldBeCancelled) {
- if (path.toFile().length() == 0) {
- return nullCompletedFuture();
- } else {
- return supplyAsync(() -> {
- try {
- FileChunkMessagesStream stream = FileChunkMessagesStream.fromPath(chunkSize, id, path);
- return sendMessagesFromStream(receiverConsistentId, stream, shouldBeCancelled)
- .whenComplete((v, e) -> {
- try {
- stream.close();
- } catch (IOException ex) {
- throw new FileTransferException("Failed to close the file transfer stream", ex);
- }
- });
- } catch (IOException e) {
- throw new FileTransferException("Failed to create a file transfer stream", e);
- }
- }, executorService).thenCompose(Function.identity());
- }
- }
-
- /**
- * Sends the next message in the stream. If there are no more messages, the future will be completed.
- *
- * @param receiverConsistentId The consistent id of the node to send the files to.
- * @param stream The stream of messages to send.
- * @return A future that will be completed when the next message is sent.
- */
- private CompletableFuture<Void> sendMessagesFromStream(
- String receiverConsistentId,
- FileChunkMessagesStream stream,
- AtomicBoolean shouldBeCancelled
- ) {
- try {
- if (stream.hasNextMessage() && !shouldBeCancelled.get()) {
- return messagingService.invoke(receiverConsistentId, FILE_TRANSFER_CHANNEL, stream.nextMessage(), responseTimeout)
- .thenApply(FileChunkResponse.class::cast)
- .thenComposeAsync(
- ack -> {
- if (ack.error() != null) {
- return failedFuture(toException(ack.error()));
- } else {
- return sendMessagesFromStream(receiverConsistentId, stream, shouldBeCancelled);
- }
- },
- executorService
- );
- } else {
- return nullCompletedFuture();
- }
- } catch (IOException e) {
- return failedFuture(e);
- }
- }
-
- private static class FileTransfer {
- private final String receiverConsistentId;
- private final UUID transferId;
- private final Path path;
- private final AtomicBoolean shouldBeCancelled;
- private final CompletableFuture<Void> result = new CompletableFuture<>();
-
- private FileTransfer(String receiverConsistentId, UUID transferId, Path path, AtomicBoolean shouldBeCancelled) {
- this.receiverConsistentId = receiverConsistentId;
- this.transferId = transferId;
- this.path = path;
- this.shouldBeCancelled = shouldBeCancelled;
- }
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferChannelTypeModule.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferChannelTypeModule.java
deleted file mode 100644
index 978906bac03..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferChannelTypeModule.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import com.google.auto.service.AutoService;
-import org.apache.ignite.internal.network.ChannelTypeModule;
-import org.apache.ignite.internal.network.ChannelTypeRegistrar;
-
-/** {@link ChannelTypeModule} for file-transfer module. */
-@AutoService(ChannelTypeModule.class)
-public class FileTransferChannelTypeModule implements ChannelTypeModule {
- @Override
- public void register(ChannelTypeRegistrar channelTypeRegistrar) {
- channelTypeRegistrar.register(Channel.FILE_TRANSFER_CHANNEL);
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferMessagesHandler.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferMessagesHandler.java
deleted file mode 100644
index bd0979d4f9f..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferMessagesHandler.java
+++ /dev/null
@@ -1,169 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.apache.ignite.internal.network.file.exception.FileTransferException;
-import org.apache.ignite.internal.network.file.messages.FileChunkMessage;
-import org.apache.ignite.internal.network.file.messages.FileHeader;
-
-/**
- * Handler for file transfer messages.
- */
-class FileTransferMessagesHandler implements TransferredFilesCollector {
- private final int filesCount;
-
- private final Path dir;
-
- private final AtomicInteger filesComplete = new AtomicInteger(0);
-
- private final CompletableFuture<List<Path>> result = new CompletableFuture<>();
-
- private final Map<String, ChunkedFileWriter> fileNameToWriter = new ConcurrentHashMap<>();
-
- /**
- * Constructor.
- *
- * @param dir Directory to write files to.
- */
- FileTransferMessagesHandler(List<FileHeader> headers, Path dir) {
- this.filesCount = headers.size();
- this.dir = dir;
- headers.forEach(this::handleFileHeader);
- }
-
- /**
- * Handles file header. Creates file writer if it doesn't exist yet. Closes file writer if all chunks are received.
- *
- * @param header File header.
- */
- private void handleFileHeader(FileHeader header) {
- File file = createFile(header.name());
- if (header.length() == 0) {
- filesComplete.incrementAndGet();
- } else {
- fileNameToWriter.put(header.name(), createFileWriter(file, header.length()));
- }
- }
-
- /**
- * Handles file chunk. Creates file writer if it doesn't exist yet. Closes file writer if all chunks are received.
- *
- * @param fileChunk File chunk.
- */
- void handleFileChunk(FileChunkMessage fileChunk) {
- if (result.isDone()) {
- throw new IllegalStateException("Received chunked file after result is already done");
- }
-
- try {
- ChunkedFileWriter writer = fileNameToWriter.get(fileChunk.fileName());
-
- assert writer != null : "Received file chunk for unknown file: " + fileChunk.fileName();
-
- if (writer.write(fileChunk)) {
- filesComplete.incrementAndGet();
- completeIfAllFilesFinished();
- }
- } catch (IOException e) {
- handleFileTransferError(e);
- }
- }
-
- /**
- * Handles file transfer error. Closes all file writers and completes result exceptionally.
- *
- * @param error Error.
- */
- void handleFileTransferError(Throwable error) {
- if (result.isDone()) {
- throw new IllegalStateException("Received file transfer error after result is already done");
- }
-
- result.completeExceptionally(error);
- closeAllWriters();
- }
-
- /**
- * Completes result if all files are finished.
- */
- private void completeIfAllFilesFinished() throws IOException {
- if (filesCount == filesComplete.get()) {
- try (Stream<Path> stream = java.nio.file.Files.list(dir)) {
- result.complete(stream.collect(Collectors.toList()));
- }
- }
- }
-
- /**
- * Returns result future.
- *
- * @return Result future.
- */
- @Override
- public CompletableFuture<List<Path>> collectedFiles() {
- return result;
- }
-
- /**
- * Close all writers.
- */
- private void closeAllWriters() {
- try {
- closeAll(fileNameToWriter.values());
- } catch (Exception e) {
- throw new FileTransferException("Failed to close file writers", e);
- }
- }
-
- /**
- * Creates file writer with known file size.
- *
- * @param file File to write to.
- * @param expectedFileSize Expected file size.
- * @return File writer.
- */
- private ChunkedFileWriter createFileWriter(File file, long expectedFileSize) {
- try {
- return ChunkedFileWriter.open(file, expectedFileSize);
- } catch (IOException e) {
- handleFileTransferError(e);
- throw new FileTransferException("Failed to open file writer", e);
- }
- }
-
- private File createFile(String fileName) {
- try {
- return java.nio.file.Files.createFile(dir.resolve(fileName)).toFile();
- } catch (IOException e) {
- handleFileTransferError(e);
- throw new FileTransferException("Failed to create file", e);
- }
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferService.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferService.java
deleted file mode 100644
index bbbcc587a35..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferService.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.manager.IgniteComponent;
-import org.apache.ignite.internal.network.file.messages.Identifier;
-
-/**
- * File transferring service.
- */
-public interface FileTransferService extends IgniteComponent {
- /**
- * Adds a file provider for the given identifier.
- *
- * @param identifier Files identifier.
- * @param provider Provider.
- */
- <M extends Identifier> void addFileProvider(
- Class<M> identifier,
- FileProvider<M> provider
- );
-
- /**
- * Adds a file handler for the given identifier.
- *
- * @param identifier Files identifier.
- * @param consumer Consumer.
- */
- <M extends Identifier> void addFileConsumer(Class<M> identifier, FileConsumer<M> consumer);
-
- /**
- * Downloads files for the given identifier from the given node.
- *
- * @param sourceNodeConsistentId consistent ID of a node.
- * @param identifier Files identifier.
- * @param targetDir Target directory. The directory will be created if it doesn't exist. If the directory exists, it will be
- * cleaned up.
- * @return Future that will be completed when the download is finished. The future will contain a list of paths to the downloaded
- * files.
- */
- CompletableFuture<List<Path>> download(String sourceNodeConsistentId, Identifier identifier, Path targetDir);
-
- /**
- * Uploads files for the given identifier to the given node.
- *
- * @param targetNodeConsistentId consistent ID of a node.
- * @param identifier Files identifier.
- * @return Future that will be completed when the upload is finished.
- */
- CompletableFuture<Void> upload(String targetNodeConsistentId, Identifier identifier);
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferServiceImpl.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferServiceImpl.java
deleted file mode 100644
index be488a976a1..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/FileTransferServiceImpl.java
+++ /dev/null
@@ -1,654 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static java.util.concurrent.CompletableFuture.runAsync;
-import static java.util.concurrent.CompletableFuture.supplyAsync;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.ignite.internal.network.file.Channel.FILE_TRANSFER_CHANNEL;
-import static org.apache.ignite.internal.network.file.messages.FileHeader.fromPaths;
-import static org.apache.ignite.internal.network.file.messages.FileTransferError.fromThrowable;
-import static org.apache.ignite.internal.network.file.messages.FileTransferError.toException;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-
-import java.io.IOException;
-import java.nio.file.Path;
-import java.nio.file.StandardCopyOption;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.manager.ComponentContext;
-import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.network.MessagingService;
-import org.apache.ignite.internal.network.TopologyEventHandler;
-import org.apache.ignite.internal.network.TopologyService;
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.network.configuration.FileTransferConfiguration;
-import org.apache.ignite.internal.network.file.exception.FileTransferException;
-import org.apache.ignite.internal.network.file.messages.FileChunkMessage;
-import org.apache.ignite.internal.network.file.messages.FileChunkResponse;
-import org.apache.ignite.internal.network.file.messages.FileDownloadRequest;
-import org.apache.ignite.internal.network.file.messages.FileDownloadResponse;
-import org.apache.ignite.internal.network.file.messages.FileTransferErrorMessage;
-import org.apache.ignite.internal.network.file.messages.FileTransferFactory;
-import org.apache.ignite.internal.network.file.messages.FileTransferInitMessage;
-import org.apache.ignite.internal.network.file.messages.FileTransferInitResponse;
-import org.apache.ignite.internal.network.file.messages.FileTransferMessageType;
-import org.apache.ignite.internal.network.file.messages.Identifier;
-import org.apache.ignite.internal.thread.IgniteThreadFactory;
-import org.apache.ignite.internal.util.IgniteUtils;
-
-/**
- * Implementation of {@link FileTransferService}.
- */
-public class FileTransferServiceImpl implements FileTransferService {
- private static final IgniteLogger LOG = Loggers.forClass(FileTransferServiceImpl.class);
-
- /**
- * Response timeout.
- */
- private final long responseTimeout;
-
- /**
- * Topology service.
- */
- private final TopologyService topologyService;
-
- /**
- * Cluster service.
- */
- private final MessagingService messagingService;
-
- /**
- * Transfer directory. All files will be saved here before being moved to their final location.
- */
- private final Path transferDirectory;
-
- /**
- * File sender.
- */
- private final FileSender fileSender;
-
- /**
- * File receiver.
- */
- private final FileReceiver fileReceiver;
-
- /**
- * Executor service.
- */
- private final ExecutorService executorService;
-
- /**
- * Map of file providers.
- */
- private final Map<Short, FileProvider<Identifier>> identifierToProvider = new ConcurrentHashMap<>();
-
- /**
- * Map of file consumers.
- */
- private final Map<Short, FileConsumer<Identifier>> identifierToConsumer = new ConcurrentHashMap<>();
-
- /**
- * Map of download requests consumers.
- */
- private final Map<UUID, DownloadRequestConsumer> transferIdToDownloadConsumer = new ConcurrentHashMap<>();
-
- /**
- * File transfer factory.
- */
- private final FileTransferFactory messageFactory = new FileTransferFactory();
-
- /**
- * Constructor.
- *
- * @param nodeName Node name.
- * @param topologyService Topology service.
- * @param messagingService Messaging service.
- * @param configuration File transfer configuration.
- * @param transferDirectory Transfer directory. All files will be saved here before being moved to their final location.
- */
- FileTransferServiceImpl(
- String nodeName,
- TopologyService topologyService,
- MessagingService messagingService,
- FileTransferConfiguration configuration,
- Path transferDirectory
- ) {
- this(
- topologyService,
- messagingService,
- configuration,
- transferDirectory,
- createExecutor(nodeName, configuration)
- );
- }
-
- private static ExecutorService createExecutor(String nodeName, FileTransferConfiguration configuration) {
- ThreadPoolExecutor executor = new ThreadPoolExecutor(
- configuration.value().threadPoolSize(),
- configuration.value().threadPoolSize(),
- 10, SECONDS,
- new LinkedBlockingQueue<>(),
- IgniteThreadFactory.create(nodeName, "file-transfer", LOG)
- );
- executor.allowCoreThreadTimeOut(true);
-
- return executor;
- }
-
- /**
- * Constructor.
- *
- * @param topologyService Topology service.
- * @param messagingService Messaging service.
- * @param configuration File transfer configuration.
- * @param transferDirectory Transfer directory. All files will be saved here before being moved to their final location.
- * @param executorService Executor service.
- */
- private FileTransferServiceImpl(
- TopologyService topologyService,
- MessagingService messagingService,
- FileTransferConfiguration configuration,
- Path transferDirectory,
- ExecutorService executorService
- ) {
- this(
- configuration.value().responseTimeoutMillis(),
- topologyService,
- messagingService,
- transferDirectory,
- new FileSender(
- configuration.value().chunkSizeBytes(),
- new Semaphore(configuration.value().maxConcurrentRequests()),
- configuration.value().responseTimeoutMillis(),
- messagingService,
- executorService
- ),
- new FileReceiver(),
- executorService
- );
- }
-
- /**
- * Constructor.
- *
- * @param responseTimeout Response timeout.
- * @param topologyService Topology service.
- * @param messagingService Messaging service.
- * @param transferDirectory Transfer directory. All files will be saved here before being moved to their final location.
- * @param fileSender File sender.
- * @param fileReceiver File receiver.
- * @param executorService Executor service.
- */
- FileTransferServiceImpl(
- long responseTimeout,
- TopologyService topologyService,
- MessagingService messagingService,
- Path transferDirectory,
- FileSender fileSender,
- FileReceiver fileReceiver,
- ExecutorService executorService
- ) {
- this.responseTimeout = responseTimeout;
- this.topologyService = topologyService;
- this.messagingService = messagingService;
- this.transferDirectory = transferDirectory;
- this.fileSender = fileSender;
- this.fileReceiver = fileReceiver;
- this.executorService = executorService;
- }
-
- @Override
- public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
- topologyService.addEventHandler(new TopologyEventHandler() {
- @Override
- public void onDisappeared(InternalClusterNode member) {
- fileReceiver.cancelTransfersFromSender(member.name());
- }
- });
-
- messagingService.addMessageHandler(FileTransferMessageType.class,
- (message, sender, correlationId) -> {
- String senderConsistentId = sender.name();
-
- if (message instanceof FileDownloadRequest) {
- processDownloadRequest((FileDownloadRequest) message, senderConsistentId, correlationId);
- } else if (message instanceof FileTransferInitMessage) {
- processFileTransferInitMessage((FileTransferInitMessage) message, senderConsistentId, correlationId);
- } else if (message instanceof FileChunkMessage) {
- processFileChunkMessage((FileChunkMessage) message, senderConsistentId, correlationId);
- } else if (message instanceof FileTransferErrorMessage) {
- processFileTransferErrorMessage((FileTransferErrorMessage) message);
- } else {
- LOG.error("Unexpected message received: {}", message);
- }
- });
-
- return nullCompletedFuture();
- }
-
- @Override
- public CompletableFuture<Void> stopAsync(ComponentContext componentContext) {
- IgniteUtils.shutdownAndAwaitTermination(executorService, 10, SECONDS);
-
- return nullCompletedFuture();
- }
-
- /**
- * Processes {@link FileTransferInitMessage} message. Creates transfer directory, registers transfer, sends response to the sender and
- * passes the transferred files to the consumer.
- *
- * @param message File transfer init message.
- * @param senderConsistentId Sender consistent ID.
- * @param correlationId Correlation ID.
- */
- private void processFileTransferInitMessage(FileTransferInitMessage message, String senderConsistentId, long correlationId) {
- UUID transferId = message.transferId();
-
- Identifier identifier = message.identifier();
-
- // Create transfer directory.
- CompletableFuture<Path> directoryFuture = supplyAsync(() -> createTransferDirectory(transferId), executorService)
- .whenComplete((directory, e) -> {
- if (e != null) {
- LOG.error("Failed to create transfer directory [transferId={}, identifier={}]", e, transferId, identifier);
- }
- });
-
- // We have to register transfer before sending response to avoid the case
- // when chunks are received when FileTransferMessagesHandler is not yet registered.
- CompletableFuture<TransferredFilesCollector> collectorFuture = directoryFuture.thenApply(
- directory -> fileReceiver.registerTransfer(senderConsistentId, transferId, message.headers(), directory)
- );
-
- // Send response to the sender.
- collectorFuture.handle((collector, throwable) -> {
- return messageFactory.fileTransferInitResponse()
- .error(throwable != null ? fromThrowable(messageFactory, throwable) : null)
- .build();
- })
- .thenCompose(response -> messagingService.respond(senderConsistentId, FILE_TRANSFER_CHANNEL, response, correlationId))
- .whenComplete((v, e) -> {
- if (e != null) {
- LOG.error("Failed to send file transfer response [transferId={}, identifier={}]", e, transferId,
- identifier);
- fileReceiver.cancelTransfer(transferId, e);
- }
- });
-
- // Pass transferred files to the consumer.
- collectorFuture.thenCompose(TransferredFilesCollector::collectedFiles)
- .whenComplete((files, throwable) -> {
- if (throwable != null) {
- LOG.error("Failed to collect transferred files [transferId={}, identifier={}]",
- throwable,
- transferId,
- identifier
- );
-
- transferIdToDownloadConsumer.computeIfPresent(transferId, (k, v) -> {
- v.onError(throwable);
- return null;
- });
- }
- })
- .thenCompose(files -> {
- return getFileConsumer(transferId, identifier).consume(identifier, files)
- .whenComplete((v, e) -> {
- if (e != null) {
- LOG.error("Failed to process file transfer [transferId={}, identifier={}]",
- e,
- transferId,
- identifier
- );
- }
- });
- })
- .whenComplete((v, e) -> {
- transferIdToDownloadConsumer.remove(transferId);
- directoryFuture.thenAccept(IgniteUtils::deleteIfExists);
- });
- }
-
- /**
- * Processes the download request. Gets the files from the provider and starts the transfer. If there is an error or there are no files
- * to transfer, then sends the response to the sender.
- *
- * @param message Download request.
- * @param senderConsistentId Sender consistent ID.
- * @param correlationId Correlation ID.
- */
- private void processDownloadRequest(FileDownloadRequest message, String senderConsistentId, Long correlationId) {
- supplyAsync(() -> getFileProvider(message.identifier()), executorService)
- .thenCompose(provider -> provider.files(message.identifier()))
- .whenComplete((files, e) -> {
- if (e != null) {
- LOG.error("Failed to get files for download [transferId={}, identifier={}]",
- e,
- message.transferId(),
- message.identifier()
- );
-
- FileDownloadResponse response = messageFactory.fileDownloadResponse()
- .error(fromThrowable(messageFactory, e))
- .build();
-
- messagingService.respond(senderConsistentId, FILE_TRANSFER_CHANNEL, response, correlationId);
- } else if (files.isEmpty()) {
- LOG.warn("No files to download [transferId={}, identifier={}]", message.transferId(), message.identifier());
-
- FileDownloadResponse response = messageFactory.fileDownloadResponse()
- .error(fromThrowable(messageFactory, new FileTransferException("No files to download")))
- .build();
-
- messagingService.respond(senderConsistentId, FILE_TRANSFER_CHANNEL, response, correlationId);
- } else {
- FileDownloadResponse response = messageFactory.fileDownloadResponse()
- .build();
-
- messagingService.respond(senderConsistentId, FILE_TRANSFER_CHANNEL, response, correlationId)
- .thenComposeAsync(v -> {
- return transferFilesToNode(senderConsistentId, message.transferId(), message.identifier(), files);
- }, executorService);
- }
- });
- }
-
- /**
- * Processes {@link FileChunkMessage} message. Passes the chunk to the receiver and sends the response to the sender.
- *
- * @param message File chunk message.
- * @param senderConsistentId Sender consistent ID.
- * @param correlationId Correlation ID.
- */
- private void processFileChunkMessage(FileChunkMessage message, String senderConsistentId, long correlationId) {
- runAsync(() -> fileReceiver.receiveFileChunk(message), executorService)
- .whenComplete((v, e) -> {
- if (e != null) {
- LOG.error("Failed to process file chunk [transferId={}]", e, message.transferId());
- }
-
- FileChunkResponse ack = messageFactory.fileChunkResponse()
- .error(e != null ? fromThrowable(messageFactory, e) : null)
- .build();
-
- messagingService.respond(senderConsistentId, FILE_TRANSFER_CHANNEL, ack, correlationId);
- });
- }
-
- /**
- * Processes {@link FileTransferErrorMessage} message. Cancels the corresponding transfer.
- *
- * @param message File transfer error message.
- */
- private void processFileTransferErrorMessage(FileTransferErrorMessage message) {
- LOG.error("Received file transfer error message. Transfer will be cancelled [transferId={}, error={}",
- message.transferId(),
- message.error()
- );
- runAsync(
- () -> fileReceiver.cancelTransfer(message.transferId(), toException(message.error())),
- executorService
- );
- }
-
- private CompletableFuture<Void> sendFiles(String targetNodeConsistentId, UUID transferId, List<Path> paths) {
- return fileSender.send(targetNodeConsistentId, transferId, paths)
- .whenComplete((v, e) -> {
- if (e != null) {
- LOG.error("Failed to send files to node [nodeConsistentId={}, transferId={}]",
- e,
- targetNodeConsistentId,
- transferId
- );
-
- FileTransferErrorMessage message = messageFactory.fileTransferErrorMessage()
- .transferId(transferId)
- .error(fromThrowable(messageFactory, e))
- .build();
-
- messagingService.send(targetNodeConsistentId, FILE_TRANSFER_CHANNEL, message);
- }
- });
- }
-
- @Override
- public <M extends Identifier> void addFileProvider(
- Class<M> identifier,
- FileProvider<M> provider
- ) {
- identifierToProvider.compute(
- getMessageType(identifier),
- (k, v) -> {
- if (v != null) {
- throw new IllegalArgumentException("File provider for identifier " + identifier.getName() + " already exists");
- } else {
- return (FileProvider<Identifier>) provider;
- }
- }
- );
- }
-
- @Override
- public <M extends Identifier> void addFileConsumer(
- Class<M> identifier,
- FileConsumer<M> consumer
- ) {
- identifierToConsumer.compute(
- getMessageType(identifier),
- (k, v) -> {
- if (v != null) {
- throw new IllegalArgumentException("File handler for identifier " + identifier.getName() + " already exists");
- } else {
- return (FileConsumer<Identifier>) consumer;
- }
- }
- );
- }
-
- @Override
- public CompletableFuture<List<Path>> download(String sourceNodeConsistentId, Identifier identifier, Path targetDir) {
- UUID transferId = UUID.randomUUID();
- FileDownloadRequest downloadRequest = messageFactory.fileDownloadRequest()
- .transferId(transferId)
- .identifier(identifier)
- .build();
-
- CompletableFuture<List<Path>> downloadedFiles = new CompletableFuture<List<Path>>()
- .whenComplete((v, e) -> transferIdToDownloadConsumer.remove(transferId));
-
- transferIdToDownloadConsumer.put(transferId, new DownloadRequestConsumer(downloadedFiles, targetDir));
-
- messagingService.invoke(sourceNodeConsistentId, FILE_TRANSFER_CHANNEL, downloadRequest, responseTimeout)
- .thenApply(FileDownloadResponse.class::cast)
- .whenComplete((response, e) -> {
- if (e != null) {
- downloadedFiles.completeExceptionally(e);
- } else if (response.error() != null) {
- downloadedFiles.completeExceptionally(toException(response.error()));
- }
- });
-
- return downloadedFiles;
- }
-
- @Override
- public CompletableFuture<Void> upload(String targetNodeConsistentId, Identifier identifier) {
- return getFileProvider(identifier).files(identifier)
- .thenCompose(files -> transferFilesToNode(targetNodeConsistentId, UUID.randomUUID(), identifier, files));
- }
-
- /**
- * Transfers files to the node with the given consistent id. Sends a {@link FileTransferInitMessage} to the node and then sends the
- * files. If the node responds with an error, the returned future will be completed exceptionally.
- *
- * @param targetNodeConsistentId The consistent id of the node to transfer the files to.
- * @param transferId The id of the transfer.
- * @param identifier The identifier of the files.
- * @param paths The paths of the files to transfer.
- * @return A future that will be completed when the transfer is complete.
- */
- private CompletableFuture<Void> transferFilesToNode(
- String targetNodeConsistentId,
- UUID transferId,
- Identifier identifier,
- List<Path> paths
- ) {
- if (paths.isEmpty()) {
- return failedFuture(new FileTransferException("No files to upload"));
- }
-
- FileTransferInitMessage message = messageFactory.fileTransferInitMessage()
- .transferId(transferId)
- .identifier(identifier)
- .headers(fromPaths(messageFactory, paths))
- .build();
-
- return messagingService.invoke(targetNodeConsistentId, FILE_TRANSFER_CHANNEL, message, responseTimeout)
- .thenApply(FileTransferInitResponse.class::cast)
- .thenComposeAsync(response -> {
- if (response.error() != null) {
- return failedFuture(
- new FileTransferException("Failed to upload files: " + response.error().message())
- );
- } else {
- return sendFiles(targetNodeConsistentId, transferId, paths);
- }
- }, executorService)
- .whenComplete((v, e) -> {
- if (e != null) {
- LOG.error("Failed to transfer files to node [nodeConsistentId={}, transferId={}]",
- e,
- targetNodeConsistentId,
- transferId
- );
- }
- });
- }
-
- private static short getMessageType(Class<?> identifier) {
- Transferable annotation = identifier.getAnnotation(Transferable.class);
- if (annotation == null) {
- throw new IllegalArgumentException("Class " + identifier.getName() + " is not annotated with @Transferable");
- } else {
- return annotation.value();
- }
- }
-
- private <M extends Identifier> FileProvider<M> getFileProvider(M identifier) {
- FileProvider<Identifier> provider = identifierToProvider.get(identifier.messageType());
- if (provider == null) {
- throw new IllegalArgumentException("File provider for identifier " + identifier.getClass().getName() + " not found");
- } else {
- return (FileProvider<M>) provider;
- }
- }
-
- /**
- * Returns the file consumer for the given transfer ID or identifier. If there is no consumer for the transfer ID, the consumer for the
- * identifier is returned.
- *
- * @param transferId The transfer ID.
- * @param identifier The identifier.
- * @return The file consumer.
- */
- private <M extends Identifier> FileConsumer<M> getFileConsumer(UUID transferId, M identifier) {
- return transferIdToDownloadConsumer.containsKey(transferId)
- ? (FileConsumer<M>) transferIdToDownloadConsumer.get(transferId)
- : getFileConsumer(identifier);
- }
-
- private <M extends Identifier> FileConsumer<M> getFileConsumer(M identifier) {
- FileConsumer<Identifier> consumer = identifierToConsumer.get(identifier.messageType());
- if (consumer == null) {
- throw new IllegalArgumentException("File consumer for identifier " + identifier.getClass().getName() + " not found");
- } else {
- return (FileConsumer<M>) consumer;
- }
- }
-
- private Path createTransferDirectory(UUID transferId) {
- try {
- return java.nio.file.Files.createDirectories(transferDirectory.resolve(transferId.toString()));
- } catch (IOException e) {
- throw new FileTransferException("Failed to create the transfer directory with transferId: " + transferId, e);
- }
- }
-
- /**
- * Consumer for file download request. It moves downloaded files to target directory and completes future with list of downloaded
- * files.
- */
- private static class DownloadRequestConsumer implements FileConsumer<Identifier> {
- private final CompletableFuture<List<Path>> downloadedFiles;
- private final Path targetDir;
-
- /**
- * Constructor.
- *
- * @param downloadedFiles Future to complete with downloaded files.
- * @param targetDir Target directory to move downloaded files.
- */
- private DownloadRequestConsumer(CompletableFuture<List<Path>> downloadedFiles, Path targetDir) {
- this.downloadedFiles = downloadedFiles;
- this.targetDir = targetDir;
- }
-
- @Override
- public CompletableFuture<Void> consume(Identifier identifier, List<Path> uploadedFiles) {
- IgniteUtils.deleteIfExists(targetDir);
-
- if (!uploadedFiles.isEmpty()) {
- Path directory = uploadedFiles.get(0).getParent();
- try {
- java.nio.file.Files.move(directory, targetDir, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
-
- try (Stream<Path> stream = java.nio.file.Files.list(targetDir)) {
- downloadedFiles.complete(stream.collect(Collectors.toList()));
- }
- } catch (IOException e) {
- downloadedFiles.completeExceptionally(e);
- return failedFuture(e);
- }
- }
-
- return nullCompletedFuture();
- }
-
- /**
- * Completes future with error.
- *
- * @param e Error.
- */
- private void onError(Throwable e) {
- downloadedFiles.completeExceptionally(e);
- }
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/TransferredFilesCollector.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/TransferredFilesCollector.java
deleted file mode 100644
index c4c43ad7c6a..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/TransferredFilesCollector.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.nio.file.Path;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-
-/**
- * Collector for transferred files.
- */
-@FunctionalInterface
-public interface TransferredFilesCollector {
- /**
- * Returns future with list of collected files. Future is completed when all files are collected.
- *
- * @return Future with list of collected files.
- */
- CompletableFuture<List<Path>> collectedFiles();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/exception/FileTransferException.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/exception/FileTransferException.java
deleted file mode 100644
index b389311c5c5..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/exception/FileTransferException.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.exception;
-
-import static org.apache.ignite.lang.ErrorGroups.Network.FILE_TRANSFER_ERR;
-
-import org.apache.ignite.lang.IgniteException;
-
-/**
- * Exception is thrown when file transfer fails.
- */
-public class FileTransferException extends IgniteException {
- /**
- * Constructor.
- *
- * @param message Message.
- */
- public FileTransferException(String message) {
- super(FILE_TRANSFER_ERR, message);
- }
-
- /**
- * Constructor.
- *
- * @param message Message.
- * @param cause Cause.
- */
- public FileTransferException(String message, Throwable cause) {
- super(FILE_TRANSFER_ERR, message, cause);
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/exception/FileValidationException.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/exception/FileValidationException.java
deleted file mode 100644
index f8bf6cc6329..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/exception/FileValidationException.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.exception;
-
-import static org.apache.ignite.lang.ErrorGroups.Network.FILE_VALIDATION_ERR;
-
-import org.apache.ignite.lang.IgniteException;
-
-/**
- * Exception is thrown when there is an error during file validation.
- */
-public class FileValidationException extends IgniteException {
- /**
- * Constructor.
- *
- * @param message Message.
- */
- public FileValidationException(String message) {
- super(FILE_VALIDATION_ERR, message);
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileChunkMessage.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileChunkMessage.java
deleted file mode 100644
index 996de35b6e3..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileChunkMessage.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import java.util.UUID;
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-
-/**
- * Chunked file. This message is sent by the sender to the receiver to transfer a file chunk.
- */
-@Transferable(FileTransferMessageType.FILE_CHUNK_MESSAGE)
-public interface FileChunkMessage extends NetworkMessage {
- /**
- * Returns transfer ID.
- *
- * @return Transfer ID.
- */
- UUID transferId();
-
- /**
- * Returns file name.
- *
- * @return File name.
- */
- String fileName();
-
- /**
- * Returns number of the chunk.
- *
- * @return Number of the chunk.
- */
- int number();
-
- /**
- * Returns data.
- *
- * @return Data.
- */
- byte[] data();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileChunkResponse.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileChunkResponse.java
deleted file mode 100644
index 01c44f625da..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileChunkResponse.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * File chunk ack. This message is sent by the receiver to the sender to acknowledge the successful reception of a file chunk.
- */
-@Transferable(FileTransferMessageType.FILE_CHUNK_RESPONSE)
-public interface FileChunkResponse extends NetworkMessage {
- /**
- * Returns the error.
- *
- * @return Error error.
- */
- @Nullable
- FileTransferError error();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileDownloadRequest.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileDownloadRequest.java
deleted file mode 100644
index 056ac03adb5..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileDownloadRequest.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import java.util.UUID;
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-
-/**
- * File download request. This message is sent by the receiver to the sender to request a file download.
- */
-@Transferable(FileTransferMessageType.FILE_DOWNLOAD_REQUEST)
-public interface FileDownloadRequest extends NetworkMessage {
- /**
- * Returns the ID of the transfer.
- *
- * @return Transfer ID.
- */
- UUID transferId();
-
- /**
- * Returns the identifier of the files to download.
- *
- * @return Files identifier.
- */
- Identifier identifier();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileDownloadResponse.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileDownloadResponse.java
deleted file mode 100644
index 44ac673acee..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileDownloadResponse.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * File download response. This message is sent by the sender to the receiver to confirm the file download request or to reject it.
- */
-@Transferable(FileTransferMessageType.FILE_DOWNLOAD_RESPONSE)
-public interface FileDownloadResponse extends NetworkMessage {
- /**
- * Returns the error that occurred during the handling file download request.
- *
- * @return The error. {@code null} if the request was handled successfully and the files are going to be transferred.
- */
- @Nullable
- FileTransferError error();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileHeader.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileHeader.java
deleted file mode 100644
index 5205d76c558..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileHeader.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import static java.util.stream.Collectors.toList;
-
-import java.io.File;
-import java.nio.file.Path;
-import java.util.List;
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-
-/**
- * File header. This message is sent by the sender to the receiver to provide information about the file to be transferred.
- */
-@Transferable(FileTransferMessageType.FILE_HEADER)
-public interface FileHeader extends NetworkMessage {
- /**
- * Returns the name of the file.
- *
- * @return File name.
- */
- String name();
-
- /**
- * Returns the size of the file in bytes.
- *
- * @return File size.
- */
- long length();
-
- /**
- * Creates a new {@link FileHeader} instance from the given {@link File}.
- */
- static FileHeader fromPath(FileTransferFactory factory, Path path) {
- return factory.fileHeader()
- .name(path.getFileName().toString())
- .length(path.toFile().length())
- .build();
- }
-
- /**
- * Creates a list of {@link FileHeader} instances from the given list of {@link Path}s.
- *
- * @param factory File transfer factory.
- * @param paths List of paths.
- * @return List of file headers.
- */
- static List<FileHeader> fromPaths(FileTransferFactory factory, List<Path> paths) {
- return paths.stream().map(path -> fromPath(factory, path))
- .collect(toList());
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferError.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferError.java
deleted file mode 100644
index 3e41c463b90..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferError.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.network.file.exception.FileTransferException;
-import org.apache.ignite.internal.util.ExceptionUtils;
-
-/**
- * File transfer error.
- */
-@Transferable(FileTransferMessageType.FILE_TRANSFER_ERROR)
-public interface FileTransferError extends NetworkMessage {
- /**
- * Returns the error code.
- *
- * @return Error code.
- */
- int code();
-
- /**
- * Returns the error message.
- *
- * @return Error message.
- */
- String message();
-
- /**
- * Creates a new {@link FileTransferError} instance from the given {@link Throwable}.
- *
- * @param factory File transfer factory.
- * @param throwable Throwable.
- * @return File transfer error.
- */
- static FileTransferError fromThrowable(FileTransferFactory factory, Throwable throwable) {
- return factory.fileTransferError()
- .message(ExceptionUtils.unwrapCause(throwable).getMessage())
- .build();
- }
-
- /**
- * Converts the given {@link FileTransferError} to an {@link Exception}.
- *
- * @param error File transfer error.
- * @return Exception.
- */
- static Exception toException(FileTransferError error) {
- return new FileTransferException(error.message());
- }
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferErrorMessage.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferErrorMessage.java
deleted file mode 100644
index f84f57e151b..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferErrorMessage.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import java.util.UUID;
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-
-/**
- * File transfer error message.
- */
-@Transferable(FileTransferMessageType.FILE_TRANSFER_ERROR_MESSAGE)
-public interface FileTransferErrorMessage extends NetworkMessage {
- /**
- * Returns transfer ID.
- *
- * @return Transfer ID.
- */
- UUID transferId();
-
- /**
- * Returns error message.
- *
- * @return Error message.
- */
- FileTransferError error();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferInitMessage.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferInitMessage.java
deleted file mode 100644
index 14691cf4769..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferInitMessage.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import java.util.List;
-import java.util.UUID;
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-
-/**
- * File transfer init message. This message is sent by the sender to the receiver to initiate a file transfer.
- */
-@Transferable(FileTransferMessageType.FILE_TRANSFER_INIT_MESSAGE)
-public interface FileTransferInitMessage extends NetworkMessage {
- /**
- * Returns the identifier of the file transfer.
- *
- * @return Identifier of the file transfer.
- */
- UUID transferId();
-
- /**
- * Returns the identifier of the files that are going to be transferred.
- *
- * @return Identifier of the files.
- */
- Identifier identifier();
-
- /**
- * Returns the headers of the files that are going to be transferred.
- *
- * @return Headers of the files.
- */
- List<FileHeader> headers();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferInitResponse.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferInitResponse.java
deleted file mode 100644
index 1a128de797d..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferInitResponse.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * File transfer init response message. This message is sent by the receiver to the sender to confirm the file transfer or to reject it.
- */
-@Transferable(FileTransferMessageType.FILE_TRANSFER_INIT_RESPONSE)
-public interface FileTransferInitResponse extends NetworkMessage {
- /**
- * Returns the error.
- *
- * @return Error.
- */
- @Nullable
- FileTransferError error();
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferMessageType.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferMessageType.java
deleted file mode 100644
index ab66537a4ca..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/FileTransferMessageType.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import org.apache.ignite.internal.network.annotations.MessageGroup;
-
-/**
- * File transfer message types.
- */
-@MessageGroup(groupType = 200, groupName = "FileTransfer")
-public final class FileTransferMessageType {
- /**
- * Type for {@link Identifier}.
- */
- public static final short FILE_IDENTIFIER = 0;
-
- /**
- * Type for {@link FileHeader}.
- */
- public static final short FILE_HEADER = 1;
-
- /**
- * Type for {@link FileChunkMessage}.
- */
- public static final short FILE_CHUNK_MESSAGE = 2;
-
- /**
- * Type for {@link FileChunkResponse}.
- */
- public static final short FILE_CHUNK_RESPONSE = 3;
-
- /**
- * Type for {@link FileTransferError}.
- */
- public static final short FILE_TRANSFER_ERROR = 4;
-
- /**
- * Type for {@link FileDownloadRequest}.
- */
- public static final short FILE_DOWNLOAD_REQUEST = 5;
-
- /**
- * Type for {@link FileDownloadResponse}.
- */
- public static final short FILE_DOWNLOAD_RESPONSE = 6;
-
- /**
- * Type for {@link FileTransferInitMessage}.
- */
- public static final short FILE_TRANSFER_INIT_MESSAGE = 7;
-
- /**
- * Type for {@link FileTransferInitResponse}.
- */
- public static final short FILE_TRANSFER_INIT_RESPONSE = 8;
-
- /**
- * Type for {@link FileTransferErrorMessage}.
- */
- public static final short FILE_TRANSFER_ERROR_MESSAGE = 9;
-}
diff --git a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/Identifier.java b/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/Identifier.java
deleted file mode 100644
index 90a4832fad2..00000000000
--- a/modules/file-transfer/src/main/java/org/apache/ignite/internal/network/file/messages/Identifier.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file.messages;
-
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.network.file.FileConsumer;
-import org.apache.ignite.internal.network.file.FileProvider;
-
-/**
- * Files identifier. This interface is used to mark all identifier messages. Identifier messages are used to retrieve files from
- * {@link FileProvider} and handle them on the receiving side by {@link FileConsumer}.
- */
-@Transferable(FileTransferMessageType.FILE_IDENTIFIER)
-public interface Identifier extends NetworkMessage {
-}
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/ChunkedFileReaderTest.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/ChunkedFileReaderTest.java
deleted file mode 100644
index 7dd194f2811..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/ChunkedFileReaderTest.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-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.io.IOException;
-import java.nio.file.Path;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-@ExtendWith(WorkDirectoryExtension.class)
-class ChunkedFileReaderTest {
- private static final int CHUNK_SIZE = 1024;
-
- @WorkDirectory
- private Path workDir;
-
- @Test
- void readEmptyFile() throws IOException {
- Path empty = FileGenerator.randomFile(workDir, 0);
- try (ChunkedFileReader reader = ChunkedFileReader.open(empty.toFile(), CHUNK_SIZE)) {
- assertFalse(reader.hasNextChunk());
- IOException exception = assertThrows(IOException.class, reader::readNextChunk);
- assertEquals("No more chunks to read", exception.getMessage());
- }
- }
-
- @Test
- void readFileWithLengthLessThanChunkSize() throws IOException {
- Path file = FileGenerator.randomFile(workDir, CHUNK_SIZE - 1);
- try (ChunkedFileReader reader = ChunkedFileReader.open(file.toFile(), CHUNK_SIZE)) {
- assertTrue(reader.hasNextChunk());
- assertEquals(CHUNK_SIZE - 1, reader.readNextChunk().length);
- assertFalse(reader.hasNextChunk());
- IOException exception = assertThrows(IOException.class, reader::readNextChunk);
- assertEquals("No more chunks to read", exception.getMessage());
- }
- }
-
- @Test
- void readFileWithLengthEqualToChunkSize() throws IOException {
- Path file = FileGenerator.randomFile(workDir, CHUNK_SIZE);
- try (ChunkedFileReader reader = ChunkedFileReader.open(file.toFile(), CHUNK_SIZE)) {
- assertTrue(reader.hasNextChunk());
- assertEquals(CHUNK_SIZE, reader.readNextChunk().length);
- assertFalse(reader.hasNextChunk());
- IOException exception = assertThrows(IOException.class, reader::readNextChunk);
- assertEquals("No more chunks to read", exception.getMessage());
- }
- }
-
- @Test
- void readFileWithLengthGreaterThanChunkSize() throws IOException {
- Path file = FileGenerator.randomFile(workDir, CHUNK_SIZE + 1);
- try (ChunkedFileReader reader = ChunkedFileReader.open(file.toFile(), CHUNK_SIZE)) {
- assertTrue(reader.hasNextChunk());
- assertEquals(CHUNK_SIZE, reader.readNextChunk().length);
- assertTrue(reader.hasNextChunk());
- assertEquals(1, reader.readNextChunk().length);
- assertFalse(reader.hasNextChunk());
- IOException exception = assertThrows(IOException.class, reader::readNextChunk);
- assertEquals("No more chunks to read", exception.getMessage());
- }
- }
-}
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/ChunkedFileWriterTest.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/ChunkedFileWriterTest.java
deleted file mode 100644
index ed06b72d52f..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/ChunkedFileWriterTest.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Stream;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.apache.ignite.internal.testframework.matchers.PathMatcher;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
-
-@ExtendWith(WorkDirectoryExtension.class)
-class ChunkedFileWriterTest {
- private static final int CHUNK_SIZE = 1024;
-
- @WorkDirectory
- private Path workDir;
-
- private Path writerDir;
-
- private static Stream<Arguments> fileLengths() {
- return Stream.of(
- Arguments.of(0),
- Arguments.of(CHUNK_SIZE - 1),
- Arguments.of(CHUNK_SIZE),
- Arguments.of(CHUNK_SIZE + 1)
- );
- }
-
- @BeforeEach
- void setUp() throws IOException {
- writerDir = Files.createDirectories(workDir.resolve("writer"));
- }
-
- @ParameterizedTest
- @MethodSource("fileLengths")
- void write(int length) throws IOException {
- Path pathToRead = FileGenerator.randomFile(workDir, length);
- Path pathToWrite = writerDir.resolve(pathToRead.getFileName());
- try (FileChunkMessagesStream stream = FileChunkMessagesStream.fromPath(CHUNK_SIZE, UUID.randomUUID(), pathToRead);
- ChunkedFileWriter writer = ChunkedFileWriter.open(pathToWrite.toFile(), length)) {
-
- AtomicBoolean isComplete = new AtomicBoolean(false);
- while (stream.hasNextMessage()) {
- if (writer.write(stream.nextMessage())) {
- assertTrue(isComplete.compareAndSet(false, true));
- }
- }
-
- // The file should be written.
- assertThat(pathToWrite, PathMatcher.hasSameContentAndName(pathToRead));
- }
- }
-}
-
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileChunkMessagesStreamTest.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileChunkMessagesStreamTest.java
deleted file mode 100644
index b1b7236d370..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileChunkMessagesStreamTest.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static org.apache.ignite.internal.network.file.FileGenerator.randomFile;
-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.io.IOException;
-import java.nio.file.Path;
-import java.util.UUID;
-import org.apache.ignite.internal.network.file.messages.FileChunkMessage;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-@ExtendWith(WorkDirectoryExtension.class)
-class FileChunkMessagesStreamTest {
- private static final int CHUNK_SIZE = 1024;
-
- @WorkDirectory
- private Path workDir;
-
- @Test
- void negativeChunkSize() {
- UUID transferId = UUID.randomUUID();
- Path file = randomFile(workDir, 1024);
- int chunkSize = -1;
-
- assertThrows(IllegalArgumentException.class, () -> FileChunkMessagesStream.fromPath(chunkSize, transferId, file));
- }
-
- @Test
- void emptyFile() throws IOException {
- UUID transferId = UUID.randomUUID();
- Path file = randomFile(workDir, 0);
-
- try (FileChunkMessagesStream stream = FileChunkMessagesStream.fromPath(CHUNK_SIZE, transferId, file)) {
- assertFalse(stream.hasNextMessage());
- assertThrows(IllegalStateException.class, stream::nextMessage);
- }
- }
-
- @Test
- void fileWithSizeLessThanChunkSize() throws IOException {
- UUID transferId = UUID.randomUUID();
- int fileSize = CHUNK_SIZE - 1;
- Path file = randomFile(workDir, fileSize);
-
- try (FileChunkMessagesStream stream = FileChunkMessagesStream.fromPath(CHUNK_SIZE, transferId, file)) {
-
- assertTrue(stream.hasNextMessage());
- FileChunkMessage fileChunkMessage = stream.nextMessage();
- assertEquals(transferId, fileChunkMessage.transferId());
- assertEquals(0, fileChunkMessage.number());
- assertEquals(fileSize, fileChunkMessage.data().length);
-
- assertFalse(stream.hasNextMessage());
- }
- }
-
- @Test
- void fileWithSizeEqualToChunkSize() throws IOException {
- UUID transferId = UUID.randomUUID();
- int fileSize = CHUNK_SIZE;
- Path file = randomFile(workDir, fileSize);
-
- try (FileChunkMessagesStream stream = FileChunkMessagesStream.fromPath(CHUNK_SIZE, transferId, file)) {
-
- assertTrue(stream.hasNextMessage());
- FileChunkMessage fileChunkMessage = stream.nextMessage();
- assertEquals(transferId, fileChunkMessage.transferId());
- assertEquals(0, fileChunkMessage.number());
- assertEquals(fileSize, fileChunkMessage.data().length);
-
- assertFalse(stream.hasNextMessage());
- }
- }
-
- @Test
- void fileWithSizeGreaterThanChunkSize() throws IOException {
- UUID transferId = UUID.randomUUID();
- int fileSize = CHUNK_SIZE + 1;
- Path file = randomFile(workDir, fileSize);
-
- try (FileChunkMessagesStream stream = FileChunkMessagesStream.fromPath(CHUNK_SIZE, transferId, file)) {
-
- assertTrue(stream.hasNextMessage());
- FileChunkMessage fileChunkMessage = stream.nextMessage();
- assertEquals(transferId, fileChunkMessage.transferId());
- assertEquals(0, fileChunkMessage.number());
- assertEquals(CHUNK_SIZE, fileChunkMessage.data().length);
-
- assertTrue(stream.hasNextMessage());
- fileChunkMessage = stream.nextMessage();
- assertEquals(transferId, fileChunkMessage.transferId());
- assertEquals(1, fileChunkMessage.number());
- assertEquals(1, fileChunkMessage.data().length);
-
- assertFalse(stream.hasNextMessage());
- }
- }
-}
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileReceiverTest.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileReceiverTest.java
deleted file mode 100644
index debba800f26..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileReceiverTest.java
+++ /dev/null
@@ -1,174 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static org.apache.ignite.internal.network.file.FileGenerator.randomFile;
-import static org.apache.ignite.internal.network.file.PathAssertions.namesAndContentEquals;
-import static org.apache.ignite.internal.network.file.messages.FileHeader.fromPaths;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
-import static org.hamcrest.MatcherAssert.assertThat;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.UUID;
-import org.apache.ignite.internal.network.file.exception.FileTransferException;
-import org.apache.ignite.internal.network.file.messages.FileTransferFactory;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-@ExtendWith(WorkDirectoryExtension.class)
-class FileReceiverTest {
- private static final int CHUNK_SIZE = 1024;
-
- @WorkDirectory
- private Path workDir;
-
- private final FileTransferFactory messageFactory = new FileTransferFactory();
-
- @Test
- void receiveSingleFile() throws IOException {
- // When a single file is sent.
- List<Path> filesToSend = List.of(randomFile(workDir, CHUNK_SIZE));
- FileReceiver receiver = new FileReceiver();
- UUID transferId = UUID.randomUUID();
-
- Path path = Files.createDirectory(workDir.resolve(transferId.toString()));
-
- TransferredFilesCollector collector = receiver.registerTransfer(
- "node2",
- transferId,
- fromPaths(messageFactory, filesToSend),
- path
- );
-
- sendFilesToReceiver(receiver, transferId, filesToSend);
-
- // Then the file is received.
- assertThat(
- collector.collectedFiles(),
- willBe(namesAndContentEquals(filesToSend))
- );
- }
-
- @Test
- void receiveMultipleFiles() throws IOException {
- // When multiple files are sent.
- List<Path> filesToSend = List.of(
- randomFile(workDir, CHUNK_SIZE),
- randomFile(workDir, CHUNK_SIZE * 2),
- randomFile(workDir, CHUNK_SIZE * 3)
- );
- FileReceiver receiver = new FileReceiver();
- UUID transferId = UUID.randomUUID();
-
- Path path = Files.createDirectory(workDir.resolve(transferId.toString()));
-
- TransferredFilesCollector collector = receiver.registerTransfer(
- "node2",
- transferId,
- fromPaths(messageFactory, filesToSend),
- path
- );
-
- sendFilesToReceiver(receiver, transferId, filesToSend);
-
- // Then the files are received.
- assertThat(
- collector.collectedFiles(),
- willBe(namesAndContentEquals(filesToSend))
- );
- }
-
- @Test
- void transfersCanceled() throws IOException {
- // When.
- FileReceiver receiver = new FileReceiver();
-
- // And the first file transfer is started.
- UUID transferId1 = UUID.randomUUID();
- List<Path> filesToSend1 = List.of(randomFile(workDir, CHUNK_SIZE * 2));
-
- Path path1 = Files.createDirectory(workDir.resolve(transferId1.toString()));
-
- TransferredFilesCollector collector1 = receiver.registerTransfer(
- "node2",
- transferId1,
- fromPaths(messageFactory, filesToSend1),
- path1
- );
-
- // And the second file transfer is registered.
- UUID transferId2 = UUID.randomUUID();
-
- Path path2 = Files.createDirectory(workDir.resolve(transferId2.toString()));
-
- TransferredFilesCollector collector2 = receiver.registerTransfer("node2", transferId2, List.of(), path2);
-
- // And the third file transfer from another node is started.
- UUID transferId3 = UUID.randomUUID();
- List<Path> filesToSend3 = List.of(randomFile(workDir, CHUNK_SIZE));
-
- Path path3 = Files.createDirectory(workDir.resolve(transferId3.toString()));
-
- TransferredFilesCollector collector3 = receiver.registerTransfer(
- "node3",
- transferId3,
- fromPaths(messageFactory, filesToSend3),
- path3
- );
-
- sendFilesToReceiver(receiver, transferId3, filesToSend3);
-
- // All transfers from node2 are canceled.
- receiver.cancelTransfersFromSender("node2");
-
- // Then the file transfer is canceled.
- assertThat(
- collector1.collectedFiles(),
- willThrow(FileTransferException.class)
- );
-
- // And the second file transfer is canceled.
- assertThat(
- collector2.collectedFiles(),
- willThrow(FileTransferException.class)
- );
-
- // And the third file transfer is not canceled.
- assertThat(
- collector3.collectedFiles(),
- willCompleteSuccessfully()
- );
- }
-
- private static void sendFilesToReceiver(FileReceiver receiver, UUID transferId, List<Path> files) {
- files.forEach(file -> {
- try (FileChunkMessagesStream stream = FileChunkMessagesStream.fromPath(CHUNK_SIZE, transferId, file)) {
- stream.forEach(receiver::receiveFileChunk);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- });
- }
-}
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileSenderTest.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileSenderTest.java
deleted file mode 100644
index 3c0dbad129b..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileSenderTest.java
+++ /dev/null
@@ -1,296 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.network.file.messages.FileTransferError.fromThrowable;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowWithCauseOrSuppressed;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.greaterThan;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.lenient;
-import static org.mockito.Mockito.spy;
-
-import java.nio.file.Path;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.ignite.internal.network.MessagingService;
-import org.apache.ignite.internal.network.file.messages.FileChunkMessage;
-import org.apache.ignite.internal.network.file.messages.FileChunkResponse;
-import org.apache.ignite.internal.network.file.messages.FileTransferFactory;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.apache.ignite.internal.util.IgniteUtils;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-@ExtendWith(MockitoExtension.class)
-@ExtendWith(WorkDirectoryExtension.class)
-class FileSenderTest extends BaseIgniteAbstractTest {
- private static final long RESPONSE_TIMEOUT = 1000;
- private static final int CHUNK_SIZE = 1024;
-
- @WorkDirectory
- private Path workDir;
-
- @Mock
- private MessagingService messagingService;
-
- private final ExecutorService executorService = Executors.newSingleThreadExecutor();
-
- private final FileTransferFactory messageFactory = new FileTransferFactory();
-
- @BeforeEach
- void setUp() {
- lenient().doReturn(completedFuture(messageFactory.fileChunkResponse().build()))
- .when(messagingService)
- .invoke(anyString(), eq(Channel.FILE_TRANSFER_CHANNEL), any(FileChunkMessage.class), eq(RESPONSE_TIMEOUT));
- }
-
- @AfterEach
- void tearDown() {
- IgniteUtils.shutdownAndAwaitTermination(executorService, 10, TimeUnit.SECONDS);
- }
-
- @Test
- void sendSingleFile() {
- // When.
- Path randomFile = FileGenerator.randomFile(workDir, CHUNK_SIZE);
- UUID transferId = UUID.randomUUID();
- FileSender sender = new FileSender(
- CHUNK_SIZE,
- new Semaphore(4),
- RESPONSE_TIMEOUT,
- messagingService,
- executorService
- );
-
- // Then - no exception is thrown.
- assertThat(
- sender.send("node2", transferId, List.of(randomFile)),
- willCompleteSuccessfully()
- );
- }
-
- @Test
- void sendMultipleFiles() {
- // When.
- List<Path> randomFiles = List.of(
- FileGenerator.randomFile(workDir, CHUNK_SIZE),
- FileGenerator.randomFile(workDir, CHUNK_SIZE),
- FileGenerator.randomFile(workDir, CHUNK_SIZE)
- );
- UUID transferId = UUID.randomUUID();
- FileSender sender = new FileSender(
- CHUNK_SIZE,
- new Semaphore(4),
- RESPONSE_TIMEOUT,
- messagingService,
- executorService
- );
-
- // Then - no exception is thrown.
- assertThat(
- sender.send("node2", transferId, randomFiles),
- willCompleteSuccessfully()
- );
- }
-
- @Test
- void exceptionIsThrownWhenInvokeReturnException() {
- // Setup messaging service to fail on second file transfer.
- AtomicInteger count = new AtomicInteger();
- doAnswer(invocation -> {
- if (count.incrementAndGet() == 2) {
- return failedFuture(new RuntimeException("Test exception"));
- } else {
- return completedFuture(messageFactory.fileChunkResponse().build());
- }
- })
- .when(messagingService)
- .invoke(anyString(), eq(Channel.FILE_TRANSFER_CHANNEL), any(FileChunkMessage.class), eq(RESPONSE_TIMEOUT));
-
- // When.
- Path randomFile = FileGenerator.randomFile(workDir, CHUNK_SIZE * 5);
- UUID transferId = UUID.randomUUID();
- FileSender sender = new FileSender(
- CHUNK_SIZE,
- new Semaphore(4),
- RESPONSE_TIMEOUT,
- messagingService,
- executorService
- );
-
- // Then - exception is thrown.
- assertThat(
- sender.send("node2", transferId, List.of(randomFile)),
- willThrowWithCauseOrSuppressed(RuntimeException.class)
- );
- }
-
- @Test
- void exceptionIsThrownWhenInvokeReturnAckWithError() {
- // Setup messaging service to fail on second file transfer.
- AtomicInteger count = new AtomicInteger();
- doAnswer(invocation -> {
- if (count.incrementAndGet() == 2) {
- FileChunkResponse ackWithError = messageFactory.fileChunkResponse()
- .error(fromThrowable(messageFactory, new RuntimeException("Test exception")))
- .build();
-
- return completedFuture(ackWithError);
- } else {
- return completedFuture(messageFactory.fileChunkResponse().build());
- }
- })
- .when(messagingService)
- .invoke(anyString(), eq(Channel.FILE_TRANSFER_CHANNEL), any(FileChunkMessage.class), eq(RESPONSE_TIMEOUT));
-
- // When.
- Path randomFile = FileGenerator.randomFile(workDir, CHUNK_SIZE * 5);
- UUID transferId = UUID.randomUUID();
- FileSender sender = new FileSender(
- CHUNK_SIZE,
- new Semaphore(4),
- RESPONSE_TIMEOUT,
- messagingService,
- executorService
- );
-
- // Then - exception is thrown.
- assertThat(
- sender.send("node2", transferId, List.of(randomFile)),
- willThrowWithCauseOrSuppressed(RuntimeException.class)
- );
- }
-
- @Test
- void maxConcurrentRequestsLimitIsNotExceeded() {
- // Setup mock messaging service to emulate long processing and count concurrent requests.
- // Max concurrent requests limit is 5. If it is exceeded, exception is thrown.
- int maxConcurrentRequests = 5;
- AtomicInteger concurrentRequests = new AtomicInteger();
-
- doAnswer(invocation -> {
- int currentConcurrentRequests = concurrentRequests.incrementAndGet();
- if (currentConcurrentRequests > maxConcurrentRequests) {
- throw new RuntimeException("Max concurrent requests limit exceeded");
- }
-
- try {
- // Emulate long processing.
- TimeUnit.MILLISECONDS.sleep(5);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- } finally {
- concurrentRequests.decrementAndGet();
- }
- return completedFuture(messageFactory.fileChunkResponse().build());
- })
- .when(messagingService)
- .invoke(anyString(), eq(Channel.FILE_TRANSFER_CHANNEL), any(FileChunkMessage.class), eq(RESPONSE_TIMEOUT));
-
- // When.
- List<Path> randomFiles = List.of(
- FileGenerator.randomFile(workDir, CHUNK_SIZE),
- FileGenerator.randomFile(workDir, CHUNK_SIZE * 10),
- FileGenerator.randomFile(workDir, CHUNK_SIZE * 20)
- );
- UUID transferId = UUID.randomUUID();
-
- FileSender sender = new FileSender(
- CHUNK_SIZE,
- new Semaphore(maxConcurrentRequests),
- RESPONSE_TIMEOUT,
- messagingService,
- executorService
- );
-
- // Then - no exception is thrown.
- assertThat(
- sender.send("node2", transferId, randomFiles),
- willCompleteSuccessfully()
- );
- }
-
- @Test
- void rateLimiterIsReleasedIfSendThrowsException() {
- // Setup messaging service to fail on second file transfer.
- AtomicInteger count = new AtomicInteger();
- doAnswer(invocation -> {
- if (count.incrementAndGet() == 2) {
- return failedFuture(new RuntimeException("Test exception"));
- } else {
- return completedFuture(messageFactory.fileChunkResponse().build());
- }
- })
- .when(messagingService)
- .invoke(anyString(), eq(Channel.FILE_TRANSFER_CHANNEL), any(FileChunkMessage.class), eq(RESPONSE_TIMEOUT));
-
- // Setup rate limiter to count tryAcquire and release calls.
- Semaphore rateLimiter = spy(new Semaphore(4));
- AtomicInteger tryAcquireCount = new AtomicInteger();
- doAnswer(invocation -> {
- tryAcquireCount.incrementAndGet();
- return true;
- }).when(rateLimiter).tryAcquire();
-
- AtomicInteger releaseCount = new AtomicInteger();
- doAnswer(invocation -> {
- releaseCount.incrementAndGet();
- return null;
- }).when(rateLimiter).release();
-
- // When.
- Path randomFile = FileGenerator.randomFile(workDir, CHUNK_SIZE * 5);
- UUID transferId = UUID.randomUUID();
- FileSender sender = new FileSender(
- CHUNK_SIZE,
- rateLimiter,
- RESPONSE_TIMEOUT,
- messagingService,
- executorService
- );
-
- // Then - exception is thrown.
- assertThat(
- sender.send("node2", transferId, List.of(randomFile)),
- willThrowWithCauseOrSuppressed(RuntimeException.class)
- );
-
- // And - rate limiter is released the same number of times as it was acquired.
- assertThat(tryAcquireCount.get(), greaterThan(0));
- assertThat(releaseCount.get(), equalTo(tryAcquireCount.get()));
- }
-}
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileTransferServiceImplTest.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileTransferServiceImplTest.java
deleted file mode 100644
index 007b30a523f..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/FileTransferServiceImplTest.java
+++ /dev/null
@@ -1,311 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static java.util.UUID.randomUUID;
-import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.network.file.Channel.FILE_TRANSFER_CHANNEL;
-import static org.apache.ignite.internal.network.file.messages.FileTransferError.fromThrowable;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-import static org.awaitility.Awaitility.await;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.emptyArray;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.nullValue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.atLeastOnce;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.verify;
-
-import java.nio.file.Path;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.ignite.internal.manager.ComponentContext;
-import org.apache.ignite.internal.network.ClusterNodeImpl;
-import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.network.NetworkMessageHandler;
-import org.apache.ignite.internal.network.file.messages.FileDownloadRequest;
-import org.apache.ignite.internal.network.file.messages.FileDownloadResponse;
-import org.apache.ignite.internal.network.file.messages.FileHeader;
-import org.apache.ignite.internal.network.file.messages.FileTransferErrorMessageImpl;
-import org.apache.ignite.internal.network.file.messages.FileTransferFactory;
-import org.apache.ignite.internal.network.file.messages.FileTransferInitMessage;
-import org.apache.ignite.internal.network.file.messages.FileTransferInitResponse;
-import org.apache.ignite.internal.network.file.messages.FileTransferMessageType;
-import org.apache.ignite.internal.network.file.messages.Identifier;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.apache.ignite.internal.util.CompletableFutures;
-import org.apache.ignite.network.NetworkAddress;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.Spy;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-@ExtendWith(MockitoExtension.class)
-@ExtendWith(WorkDirectoryExtension.class)
-class FileTransferServiceImplTest extends BaseIgniteAbstractTest {
- private static final String SOURCE_CONSISTENT_ID = "source";
-
- private static final String TARGET_CONSISTENT_ID = "target";
-
- private static final InternalClusterNode TARGET_NODE = new ClusterNodeImpl(
- randomUUID(),
- TARGET_CONSISTENT_ID,
- new NetworkAddress("target", 1234)
- );
-
- @WorkDirectory
- private Path workDir;
-
- private Path transferDir;
-
- @Mock
- private FileSender fileSender;
-
- @Mock
- private FileReceiver fileReceiver;
-
- @Spy
- private TestMessagingService messagingService = new TestMessagingService();
-
- @Spy
- private TestTopologyService topologyService = new TestTopologyService();
-
- private FileTransferServiceImpl fileTransferService;
-
- private final FileTransferFactory messageFactory = new FileTransferFactory();
-
- @BeforeEach
- void setUp() {
- transferDir = workDir.resolve("transfer");
- fileTransferService = new FileTransferServiceImpl(
- 10_000,
- topologyService,
- messagingService,
- transferDir,
- fileSender,
- fileReceiver,
- Executors.newSingleThreadExecutor()
- );
-
- assertThat(fileTransferService.startAsync(new ComponentContext()), willCompleteSuccessfully());
- }
-
- @AfterEach
- void tearDown() {
- assertThat(fileTransferService.stopAsync(new ComponentContext()), willCompleteSuccessfully());
- }
-
- @Test
- void handlersAreAddedOnStart() {
- verify(messagingService).addMessageHandler(eq(FileTransferMessageType.class), any(NetworkMessageHandler.class));
- verify(topologyService).addEventHandler(any());
- }
-
- @Test
- void fileTransfersCanceledWhenSenderLeft() {
- topologyService.fireDisappearedEvent(new ClusterNodeImpl(randomUUID(), "sender", new NetworkAddress("localhost", 1234)));
-
- verify(fileReceiver).cancelTransfersFromSender("sender");
- }
-
- @Test
- void uploadFailsWhenInvokeReturnsResponseWithErrorOnFileTransferInitMessage() {
- // Set messaging service to fail to invoke.
- FileTransferInitResponse response = messageFactory.fileTransferInitResponse()
- .error(fromThrowable(messageFactory, new RuntimeException("Test exception")))
- .build();
- doReturn(completedFuture(response))
- .when(messagingService)
- .invoke(eq(TARGET_CONSISTENT_ID), eq(FILE_TRANSFER_CHANNEL), any(FileTransferInitMessage.class), anyLong());
-
- // Set file provider to provide a file.
- Path path1 = FileGenerator.randomFile(workDir, 100);
- fileTransferService.addFileProvider(Identifier.class, id -> completedFuture(List.of(path1)));
-
- // Upload file to target.
- CompletableFuture<Void> uploaded = fileTransferService.upload(TARGET_CONSISTENT_ID, messageFactory.identifier().build());
-
- // Check that upload failed.
- assertThat(uploaded, willThrow(RuntimeException.class, "Test exception"));
- }
-
- @Test
- void uploadFailsWhenSenderReturnsException() {
- // Set messaging service to fail to invoke.
- FileTransferInitResponse fileTransferInitResponse = messageFactory.fileTransferInitResponse().build();
- doReturn(completedFuture(fileTransferInitResponse))
- .when(messagingService)
- .invoke(eq(TARGET_CONSISTENT_ID), eq(FILE_TRANSFER_CHANNEL), any(FileTransferInitMessage.class), anyLong());
-
- doReturn(failedFuture(new RuntimeException("Test exception")))
- .when(fileSender)
- .send(eq(TARGET_CONSISTENT_ID), any(UUID.class), anyList());
-
- // Set file provider to provide a file.
- Path path1 = FileGenerator.randomFile(workDir, 100);
- fileTransferService.addFileProvider(Identifier.class, id -> completedFuture(List.of(path1)));
-
- // Upload file to target.
- CompletableFuture<Void> uploaded = fileTransferService.upload(TARGET_CONSISTENT_ID, messageFactory.identifier().build());
-
- // Check that upload failed.
- assertThat(uploaded, willThrow(RuntimeException.class, "Test exception"));
-
- // Check that error message was sent.
- await().untilAsserted(() -> {
- verify(messagingService, atLeastOnce())
- .send(eq(TARGET_CONSISTENT_ID), eq(FILE_TRANSFER_CHANNEL), any(FileTransferErrorMessageImpl.class));
- });
- }
-
- @Test
- void uploadFailsWhenInvokeReturnsExceptionOnFileTransferInitResponse() {
- long correlationId = 1L;
-
- // Set messaging service to fail to send upload response.
- RuntimeException testException = new RuntimeException("Test exception");
- doReturn(failedFuture(testException))
- .when(messagingService)
- .respond(eq(TARGET_CONSISTENT_ID), eq(FILE_TRANSFER_CHANNEL), any(FileTransferInitResponse.class),
- eq(correlationId));
-
- // Set file receiver to complete transfer registration.
- CompletableFuture<UUID> registeredTransferIdFuture = new CompletableFuture<>();
- CompletableFuture<List<Path>> transferredFilesFuture = new CompletableFuture<>();
- doAnswer(invocation -> {
- registeredTransferIdFuture.complete(invocation.getArgument(1, UUID.class));
- return transferredFilesFuture;
- }).when(fileReceiver).registerTransfer(eq(TARGET_CONSISTENT_ID), any(UUID.class), anyList(), any(Path.class));
-
- // Set file receiver to complete transfer cancellation.
- CompletableFuture<UUID> candeledTransferIdFuture = new CompletableFuture<>();
- doAnswer(invocation -> {
- candeledTransferIdFuture.complete(invocation.getArgument(0, UUID.class));
- transferredFilesFuture.completeExceptionally(invocation.getArgument(1, RuntimeException.class));
- return null;
- }).when(fileReceiver).cancelTransfer(any(UUID.class), any(RuntimeException.class));
-
- Path path1 = FileGenerator.randomFile(workDir, 0);
- List<Path> paths = List.of(path1);
-
- FileTransferInitMessage uploadRequest = messageFactory.fileTransferInitMessage()
- .transferId(randomUUID())
- .identifier(messageFactory.identifier().build())
- .headers(FileHeader.fromPaths(messageFactory, paths))
- .build();
-
- messagingService.fireMessage(uploadRequest, TARGET_NODE, correlationId);
-
- // Check that transfer was registered and canceled.
- assertThat(registeredTransferIdFuture, willCompleteSuccessfully());
- assertThat(candeledTransferIdFuture, willBe(registeredTransferIdFuture.join()));
-
- // Check that transfer directory is empty.
- await().untilAsserted(() -> assertThat(transferDir.toFile().listFiles(), emptyArray()));
- }
-
- @Test
- void downloadFailsWhenInvokeReturnsExceptionOnDownloadRequest() {
- // Set messaging service to fail to send download request.
- RuntimeException testException = new RuntimeException("Test exception");
- doReturn(failedFuture(testException))
- .when(messagingService)
- .invoke(eq(SOURCE_CONSISTENT_ID), eq(FILE_TRANSFER_CHANNEL), any(FileDownloadRequest.class), any(Long.class));
-
- CompletableFuture<List<Path>> downloaded = fileTransferService.download(SOURCE_CONSISTENT_ID, messageFactory.identifier().build(),
- workDir.resolve("download"));
-
- // Check that download future is completed exceptionally.
- assertThat(downloaded, willThrow(RuntimeException.class, "Test exception"));
-
- // Check that transfer directory is empty.
- await().untilAsserted(() -> assertThat(transferDir.toFile().listFiles(), nullValue()));
- }
-
- @Test
- void downloadFailsWhenInvokeReturnsResponseWithErrorOnDownloadRequest() {
- // Set messaging service to fail to send download request.
- FileDownloadResponse response = messageFactory.fileDownloadResponse()
- .error(fromThrowable(messageFactory, new RuntimeException("Test exception")))
- .build();
- doReturn(completedFuture(response))
- .when(messagingService)
- .invoke(eq(SOURCE_CONSISTENT_ID), eq(FILE_TRANSFER_CHANNEL), any(FileDownloadRequest.class), any(Long.class));
-
- CompletableFuture<List<Path>> downloaded = fileTransferService.download(SOURCE_CONSISTENT_ID, messageFactory.identifier().build(),
- workDir.resolve("download"));
-
- // Check that download future is completed exceptionally.
- assertThat(downloaded, willThrow(RuntimeException.class, "Test exception"));
-
- // Check that transfer directory is empty.
- await().untilAsserted(() -> assertThat(transferDir.toFile().listFiles(), nullValue()));
- }
-
- @Test
- void transferIsRegisteredBeforeResponseIsSent() {
- // Set file receiver to complete transfer registration.
- AtomicInteger transferLifecycleState = new AtomicInteger(0);
- doAnswer(invocation -> {
- // Update transfer lifecycle state - transfer is registered.
- transferLifecycleState.compareAndSet(0, 1);
- return (TransferredFilesCollector) CompletableFutures::emptyListCompletedFuture;
- }).when(fileReceiver).registerTransfer(anyString(), any(UUID.class), anyList(), any(Path.class));
-
- // Set messaging service to fail to send upload response if transfer is not registered.
- doAnswer(invocation -> {
- // Check lifecycle state - transfer is registered.
- transferLifecycleState.compareAndSet(1, 2);
- return nullCompletedFuture();
- }).when(messagingService).respond(anyString(), eq(FILE_TRANSFER_CHANNEL), any(FileTransferInitResponse.class), anyLong());
-
- // Create file transfer request.
- Path path1 = FileGenerator.randomFile(workDir, 0);
- List<Path> paths = List.of(path1);
-
- FileTransferInitMessage fileTransferInitMessage = messageFactory.fileTransferInitMessage()
- .transferId(randomUUID())
- .identifier(messageFactory.identifier().build())
- .headers(FileHeader.fromPaths(messageFactory, paths))
- .build();
-
- // Send file transfer request.
- messagingService.fireMessage(fileTransferInitMessage, TARGET_NODE, 1L);
-
- // Check that transfer was registered before response was sent.
- await().untilAtomic(transferLifecycleState, equalTo(2));
- }
-}
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/TestMessagingService.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/TestMessagingService.java
deleted file mode 100644
index ddee740e2d3..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/TestMessagingService.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static java.util.concurrent.CompletableFuture.failedFuture;
-
-import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.network.AbstractMessagingService;
-import org.apache.ignite.internal.network.ChannelType;
-import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.network.NetworkMessageHandler;
-import org.apache.ignite.network.NetworkAddress;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Test messaging service. It does not provide any messaging functionality, but allows to trigger events.
- */
-public class TestMessagingService extends AbstractMessagingService {
-
- @Override
- public void weakSend(InternalClusterNode recipient, NetworkMessage msg) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public CompletableFuture<Void> send(InternalClusterNode recipient, ChannelType channelType, NetworkMessage msg) {
- return failedFuture(new UnsupportedOperationException());
- }
-
- @Override
- public CompletableFuture<Void> send(String recipientConsistentId, ChannelType channelType, NetworkMessage msg) {
- return failedFuture(new UnsupportedOperationException());
- }
-
- @Override
- public CompletableFuture<Void> send(NetworkAddress recipientNetworkAddress, ChannelType channelType, NetworkMessage msg) {
- return failedFuture(new UnsupportedOperationException());
- }
-
- @Override
- public CompletableFuture<Void> respond(InternalClusterNode recipient, ChannelType channelType, NetworkMessage msg, long correlationId) {
- return failedFuture(new UnsupportedOperationException());
- }
-
- @Override
- public CompletableFuture<Void> respond(String recipientConsistentId, ChannelType channelType, NetworkMessage msg, long correlationId) {
- return failedFuture(new UnsupportedOperationException());
- }
-
- @Override
- public CompletableFuture<NetworkMessage> invoke(
- InternalClusterNode recipient,
- ChannelType channelType,
- NetworkMessage msg,
- long timeout
- ) {
- return failedFuture(new UnsupportedOperationException());
- }
-
- @Override
- public CompletableFuture<NetworkMessage> invoke(String recipientConsistentId, ChannelType channelType, NetworkMessage msg,
- long timeout) {
- return failedFuture(new UnsupportedOperationException());
- }
-
- /**
- * Calls {@link NetworkMessageHandler#onReceived(NetworkMessage, InternalClusterNode, Long)} on all registered.
- *
- * @param msg Message.
- * @param sender Sender node.
- * @param correlationId Correlation ID.
- */
- public void fireMessage(NetworkMessage msg, InternalClusterNode sender, @Nullable Long correlationId) {
- getMessageHandlers(msg.groupType()).forEach(h -> h.onReceived(msg, sender, correlationId));
- }
-}
diff --git a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/TestTopologyService.java b/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/TestTopologyService.java
deleted file mode 100644
index d1275b92cfc..00000000000
--- a/modules/file-transfer/src/test/java/org/apache/ignite/internal/network/file/TestTopologyService.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.util.Collection;
-import java.util.UUID;
-import org.apache.ignite.internal.network.AbstractTopologyService;
-import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.network.TopologyEventHandler;
-import org.apache.ignite.network.NetworkAddress;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Test topology service. It does not provide any topology information, but allows to trigger events.
- */
-public class TestTopologyService extends AbstractTopologyService {
- @Override
- public InternalClusterNode localMember() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Collection<InternalClusterNode> allMembers() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Collection<InternalClusterNode> logicalTopologyMembers() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public long logicalTopologyVersion() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public @Nullable InternalClusterNode getByAddress(NetworkAddress addr) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public @Nullable InternalClusterNode getByConsistentId(String consistentId) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public @Nullable InternalClusterNode getById(UUID id) {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Calls {@link TopologyEventHandler#onAppeared(InternalClusterNode)} on all registered event handlers.
- *
- * @param member Member.
- */
- public void fireAppearedEvent(InternalClusterNode member) {
- getEventHandlers().forEach(it -> it.onAppeared(member));
- }
-
- /**
- * Calls {@link TopologyEventHandler#onDisappeared(InternalClusterNode)} on all registered event handlers.
- *
- * @param member Member.
- */
- public void fireDisappearedEvent(InternalClusterNode member) {
- getEventHandlers().forEach(it -> it.onDisappeared(member));
- }
-
- @Override
- public void onJoined(InternalClusterNode node, long topologyVersion) {
- }
-
- @Override
- public void onLeft(InternalClusterNode node, long topologyVersion) {
- }
-}
diff --git a/modules/file-transfer/src/testFixtures/java/org/apache/ignite/internal/network/file/FileGenerator.java b/modules/file-transfer/src/testFixtures/java/org/apache/ignite/internal/network/file/FileGenerator.java
deleted file mode 100644
index 8a39794c6b4..00000000000
--- a/modules/file-transfer/src/testFixtures/java/org/apache/ignite/internal/network/file/FileGenerator.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.Random;
-
-/**
- * File generator.
- */
-public class FileGenerator {
- /**
- * Generates a random file with the given content size.
- *
- * @param dir Directory to create the file in.
- * @param contentSize Size of the file content.
- * @return Path to the generated file.
- */
- public static Path randomFile(Path dir, int contentSize) {
- try {
- Path tempFile = Files.createTempFile(dir, null, null);
- byte[] bytes = new byte[contentSize];
- new Random().nextBytes(bytes);
- return Files.write(tempFile, bytes);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-}
diff --git a/modules/file-transfer/src/testFixtures/java/org/apache/ignite/internal/network/file/PathAssertions.java b/modules/file-transfer/src/testFixtures/java/org/apache/ignite/internal/network/file/PathAssertions.java
deleted file mode 100644
index fce9d2d9240..00000000000
--- a/modules/file-transfer/src/testFixtures/java/org/apache/ignite/internal/network/file/PathAssertions.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.file;
-
-import static org.hamcrest.Matchers.containsInAnyOrder;
-
-import java.nio.file.Path;
-import java.util.List;
-import org.apache.ignite.internal.testframework.matchers.PathMatcher;
-import org.hamcrest.Matcher;
-
-/**
- * Path assertions.
- */
-public class PathAssertions {
-
- /**
- * Asserts that the given list of paths contains the same files as the expected list.
- *
- * @param expectedFiles The expected list of files.
- * @return A matcher that will match if the given list of paths contains the same files as the expected list.
- */
- public static Matcher<Iterable<? extends Path>> namesAndContentEquals(List<Path> expectedFiles) {
- Matcher<Path>[] matchers = expectedFiles.stream()
- .map(PathMatcher::hasSameContentAndName)
- .toArray(Matcher[]::new);
- return containsInAnyOrder(matchers);
- }
-}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/configuration/FileTransferConfigurationSchema.java b/modules/network/src/main/java/org/apache/ignite/internal/network/configuration/FileTransferConfigurationSchema.java
deleted file mode 100644
index 1c7bd4fc472..00000000000
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/configuration/FileTransferConfigurationSchema.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.network.configuration;
-
-import org.apache.ignite.configuration.annotation.Config;
-import org.apache.ignite.configuration.annotation.PublicName;
-import org.apache.ignite.configuration.annotation.Value;
-import org.apache.ignite.configuration.validation.Range;
-
-/**
- * File transferring configuration.
- */
-@Config
-public class FileTransferConfigurationSchema {
- @Range(min = 0)
- @Value(hasDefault = true)
- @PublicName(legacyNames = "responseTimeout")
- public final long responseTimeoutMillis = 10_000;
-
- /** Chunk size in bytes. */
- @Range(min = 1)
- @Value(hasDefault = true)
- @PublicName(legacyNames = "chunkSize")
- public final int chunkSizeBytes = 1024 * 1024;
-
- /** File sender thread pool size. */
- @Range(min = 1)
- @Value(hasDefault = true)
- public final int threadPoolSize = 8;
-
- /** Max concurrent requests. */
- @Range(min = 1)
- @Value(hasDefault = true)
- public final int maxConcurrentRequests = 4;
-}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/configuration/NetworkConfigurationSchema.java b/modules/network/src/main/java/org/apache/ignite/internal/network/configuration/NetworkConfigurationSchema.java
index 97ec469c5d6..c6b9685fdc3 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/configuration/NetworkConfigurationSchema.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/configuration/NetworkConfigurationSchema.java
@@ -80,7 +80,4 @@ public class NetworkConfigurationSchema {
@SslConfigurationValidator
public SslConfigurationSchema ssl;
- /** File transferring configuration. */
- @ConfigValue
- public FileTransferConfigurationSchema fileTransfer;
}
diff --git a/settings.gradle b/settings.gradle
index 24af08b7697..ac9ca0377ac 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -91,7 +91,6 @@ include(':ignite-security')
include(':ignite-catalog')
include(':ignite-catalog-compaction')
include(":ignite-arch-test")
-include(":ignite-file-transfer")
include(":ignite-system-view")
include(":ignite-system-view-api")
include(":ignite-error-code-annotation-processor")
@@ -175,7 +174,6 @@ project(":ignite-security-api").projectDir = file('modules/security-api')
project(":ignite-catalog").projectDir = file('modules/catalog')
project(":ignite-catalog-compaction").projectDir = file('modules/catalog-compaction')
project(":ignite-arch-test").projectDir = file('modules/arch-test')
-project(":ignite-file-transfer").projectDir = file('modules/file-transfer')
project(":ignite-system-view").projectDir = file('modules/system-view')
project(":ignite-system-view-api").projectDir = file('modules/system-view-api')
project(":ignite-error-code-annotation-processor").projectDir = file('modules/error-code-annotation-processor')