working FTP

SymphonyManager/pom.xml

@@ -49,6 +49,11 @@
             <artifactId>logback-classic</artifactId>
             <version>1.4.12</version>
         </dependency>
+        <dependency>
+            <groupId>commons-net</groupId>
+            <artifactId>commons-net</artifactId>
+            <version>3.11.1</version>
+        </dependency>
 
         <dependency>
             <groupId>ch.qos.logback</groupId>

SymphonyManager/src/main/java/io/github/lumijiez/FTPFetcher.java

@@ -0,0 +1,80 @@
+package io.github.lumijiez;
+
+import org.apache.commons.net.ftp.FTPClient;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Timer;
+import java.util.TimerTask;
+
+public class FTPFetcher {
+    String ftpServer = "ftp_server";
+    int ftpPort = 21;
+    String ftpUser = "symphony";
+    String ftpPass = "symphony";
+    String ftpDir = "/uploads";
+
+    public FTPFetcher() {
+        Timer timer = new Timer();
+        timer.scheduleAtFixedRate(new TimerTask() {
+            @Override
+            public void run() {
+                fetchAndDeleteFilesFromFtp(ftpServer, ftpPort, ftpUser, ftpPass, ftpDir);
+            }
+        }, 0, 5000);
+    }
+
+    public static void fetchAndDeleteFilesFromFtp(String ftpServer, int ftpPort, String ftpUser, String ftpPass, String ftpDir) {
+        FTPClient ftpClient = new FTPClient();
+
+        try {
+            ftpClient.connect(ftpServer, ftpPort);
+            ftpClient.setControlKeepAliveTimeout(300);
+            boolean login = ftpClient.login(ftpUser, ftpPass);
+            if (!login) {
+                System.out.println("FTP login failed.");
+                return;
+            }
+
+            ftpClient.enterLocalPassiveMode();
+            ftpClient.changeWorkingDirectory(ftpDir);
+
+            String[] fileNames = ftpClient.listNames();
+            if (fileNames == null || fileNames.length == 0) {
+                System.out.println("No files found in the directory.");
+                return;
+            }
+
+            for (String fileName : fileNames) {
+                File localFile = new File("downloaded_" + fileName);
+                try (FileOutputStream fos = new FileOutputStream(localFile)) {
+                    boolean success = ftpClient.retrieveFile(fileName, fos);
+                    if (success) {
+                        System.out.println("File downloaded: " + localFile.getName());
+                    } else {
+                        System.out.println("Failed to download the file: " + fileName);
+                    }
+                }
+
+                boolean deleted = ftpClient.deleteFile(fileName);
+                if (deleted) {
+                    System.out.println("File deleted from FTP server: " + fileName);
+                } else {
+                    System.out.println("Failed to delete the file from FTP server: " + fileName);
+                }
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        } finally {
+            try {
+                if (ftpClient.isConnected()) {
+                    ftpClient.logout();
+                    ftpClient.disconnect();
+                }
+            } catch (IOException ex) {
+                ex.printStackTrace();
+            }
+        }
+    }
+}

SymphonyManager/src/main/java/io/github/lumijiez/Main.java

@@ -12,6 +12,8 @@ public class Main {
     public static void main(String[] args) {
         new Thread(BrokerConnector::connect, "RabbitMQ-Connection").start();
 
+        new Thread(FTPFetcher::new, "FTP-Fetcher").start();
+
         Javalin app = Javalin.create(config -> {
             config.jsonMapper(new JavalinGson());
             config.jetty.modifyWebSocketServletFactory(wsFactoryConfig -> {

SymphonyProducer/pom.xml

@@ -15,6 +15,11 @@
             <artifactId>amqp-client</artifactId>
             <version>5.23.0</version>
         </dependency>
+        <dependency>
+            <groupId>commons-net</groupId>
+            <artifactId>commons-net</artifactId>
+            <version>3.11.1</version>
+        </dependency>
 
         <dependency>
             <groupId>com.google.code.gson</groupId>

SymphonyProducer/src/main/java/io/github/lumijiez/FTPClientConnector.java

@@ -0,0 +1,98 @@
+package io.github.lumijiez;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.apache.commons.net.ftp.FTPClient;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Random;
+import java.util.Timer;
+import java.util.TimerTask;
+
+public class FTPClientConnector {
+    String ftpServer = "ftp_server";
+    int ftpPort = 21;
+    String ftpUser = "symphony";
+    String ftpPass = "symphony";
+    String ftpDir = "/";
+
+    FTPClientConnector() {
+        Timer timer = new Timer();
+        timer.scheduleAtFixedRate(new TimerTask() {
+            @Override
+            public void run() {
+                uploadRandomJsonToFtp(ftpServer, ftpPort, ftpUser, ftpPass, ftpDir);
+            }
+        }, 0, 30000);
+    }
+
+    public static void uploadRandomJsonToFtp(String ftpServer, int ftpPort, String ftpUser, String ftpPass, String ftpDir) {
+        JsonObject json = generateRandomJson();
+
+        Gson gson = new Gson();
+        String jsonString = gson.toJson(json);
+        InputStream inputStream = new ByteArrayInputStream(jsonString.getBytes());
+
+        FTPClient ftpClient = new FTPClient();
+        try {
+            ftpClient.connect(ftpServer, ftpPort);
+            boolean login = ftpClient.login(ftpUser, ftpPass);
+            if (!login) {
+                System.out.println("FTP login failed.");
+                return;
+            }
+
+            ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE);
+            ftpClient.enterLocalPassiveMode();
+
+            String subDir = "uploads";
+            if (!ftpClient.changeWorkingDirectory(subDir)) {
+                System.out.println("Subdirectory does not exist. Creating: " + subDir);
+                boolean dirCreated = ftpClient.makeDirectory(subDir);
+                if (!dirCreated) {
+                    System.out.println("Failed to create subdirectory: " + subDir);
+                    System.out.println("Server reply: " + ftpClient.getReplyString());
+                    return;
+                }
+                ftpClient.changeWorkingDirectory(subDir);
+            }
+
+            String filename = "file_" + System.currentTimeMillis() + ".json";
+            boolean uploaded = ftpClient.storeFile(filename, inputStream);
+            if (uploaded) {
+                System.out.println("Successfully uploaded: " + filename);
+            } else {
+                System.out.println("Failed to upload: " + filename);
+                System.out.println("Server reply: " + ftpClient.getReplyString());
+            }
+
+            ftpClient.logout();
+        } catch (IOException e) {
+            System.out.println("Error during FTP operation: " + e.getMessage());
+            e.printStackTrace();
+        } finally {
+            try {
+                inputStream.close();
+                if (ftpClient.isConnected()) {
+                    ftpClient.disconnect();
+                }
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    public static JsonObject generateRandomJson() {
+        Random rand = new Random();
+        JsonObject json = new JsonObject();
+
+        json.addProperty("id", rand.nextInt(1000));
+        json.addProperty("name", "RandomUser" + rand.nextInt(1000));
+        json.addProperty("email", "user" + rand.nextInt(1000) + "@example.com");
+        json.addProperty("active", rand.nextBoolean());
+
+        return json;
+    }
+}

SymphonyProducer/src/main/java/io/github/lumijiez/Main.java

@@ -3,6 +3,17 @@ package io.github.lumijiez;
 public class Main {
 
     public static void main(String[] args) {
-        BrokerConnector.connect();
+        Thread brokerThread = new Thread(BrokerConnector::connect);
+        brokerThread.start();
+
+        Thread ftpThread = new Thread(FTPClientConnector::new);
+        ftpThread.start();
+
+        try {
+            brokerThread.join();
+            ftpThread.join();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
     }
 }

docker-compose.yml

@@ -207,8 +207,24 @@ services:
       - symphony-network
     restart: always
 
+  ftp_server:
+    image: stilliard/pure-ftpd:hardened
+    container_name: ftp_server
+    ports:
+      - "21:21"
+      - "30000-30009:30000-30009"
+    environment:
+      FTP_USER_NAME: symphony
+      FTP_USER_PASS: symphony
+      FTP_USER_HOME: /home/symphony
+    networks:
+      - symphony-network
+    volumes:
+      - ftp_data:/home/symphony
+
 volumes:
   grafana-storage:
+  ftp_data:
 
 networks:
   symphony-network: