Merge pull request #158 from hasadna/fix-gtfs-files-leak

Fix gtfs files leak

.travis.yml

@@ -26,7 +26,7 @@ matrix:
       script:
         - docker login -u ${DOCKER_HUB_USERNAME} -p ${DOCKER_HUB_PASSWORD}
         - cd ./siri/gtfs_reader
-        - mvn install -DskipTests=true
+        - mvn install
         - docker build -t gtfs-collector .
         - docker tag gtfs-collector openbus1/gtfs-collector:${TRAVIS_BRANCH}
         - docker push openbus1/gtfs-collector:${TRAVIS_BRANCH}

siri/docker-compose.yml

@@ -3,7 +3,7 @@ version: "3"
 services:
   siri-collector:
     container_name: siri-collector
-    image: openbus1/siri-collector:containerize-siri-collector
+    image: openbus1/siri-collector:${TAG:-master}
     environment:
       - spring.profiles.active=production
       - scheduler.data.file=/opt/hasadna/data/
@@ -15,7 +15,7 @@ services:
     restart: always
   gtfs-collector:
     container_name: gtfs-collector
-    image: openbus1/gtfs-collector:containerize-siri-collector
+    image: openbus1/gtfs-collector:${TAG:-master}
     environment:
       - gtfs.agencies=[3,4,5,7,8,14,15,16,18,19,25,30,31,32]
       - gtfs.dateOfLastDownload=2018-11-17

siri/gtfs_reader/pom.xml

@@ -7,7 +7,7 @@
     <version>0.0.1-SNAPSHOT</version>
     <packaging>jar</packaging>
 
-    <name>siri-client</name>
+    <name>gtfs-collector</name>
 
   <description>OpenBus Java Data Retriever</description>
 
@@ -84,7 +84,7 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.9.6</version>
+            <version>2.9.8</version>
         </dependency>
 
 

siri/gtfs_reader/src/main/java/il/org/hasadna/siri_client/gtfs/crud/DownloadFailedException.java

@@ -3,15 +3,12 @@
 import java.io.IOException;
 
 public class DownloadFailedException extends IOException {
-    public DownloadFailedException() {
-        super();
-    }
-    public DownloadFailedException(String message) {
-        super(message);
-    }
-    public DownloadFailedException(String message, Throwable cause) {
-        super(message, cause);
-    }
 
+  public DownloadFailedException(String message) {
+    super(message);
+  }
 
+  public DownloadFailedException(Throwable cause) {
+    super(cause);
+  }
 }

siri/gtfs_reader/src/main/java/il/org/hasadna/siri_client/gtfs/crud/FtpClientService.java

@@ -0,0 +1,89 @@
+package il.org.hasadna.siri_client.gtfs.crud;
+
+import il.org.hasadna.siri_client.gtfs.main.GtfsCollectorConfiguration;
+import java.io.BufferedOutputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.RetryPolicy;
+import org.apache.commons.net.ftp.FTP;
+import org.apache.commons.net.ftp.FTPClient;
+import org.apache.commons.net.ftp.FTPReply;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+@Service
+public class FtpClientService {
+
+  private static Logger logger = LoggerFactory.getLogger(FtpClientService.class);
+
+  public Path downloadFileWithRetry(String ftpServer, String remoteFileName, Path localFilePath) throws DownloadFailedException {
+
+    try {
+      RetryPolicy retryPolicy = new RetryPolicy()
+          .retryOn(DownloadFailedException.class)
+          .withBackoff(1, 30, TimeUnit.MINUTES)
+          .withMaxRetries(5);
+
+      return Failsafe.with(retryPolicy).get(() -> downloadFile(ftpServer, remoteFileName, localFilePath));
+
+    } catch (Exception e) {
+      throw new DownloadFailedException(e);
+    }
+  }
+
+  private Path downloadFile(String ftpServer, String remoteFileName, Path localFilePath) throws DownloadFailedException {
+    OutputStream localFileOutputStream = null;
+
+    try {
+      logger.info("connect to ftp...");
+      FTPClient ftpClient = connect(ftpServer);
+      localFileOutputStream = new BufferedOutputStream(new FileOutputStream(localFilePath.toFile()));
+
+      logger.info("start downloading from ftp...");
+
+      if (!ftpClient.retrieveFile(remoteFileName, localFileOutputStream)) {
+        logger.error("retrieveFile returned false, fileName={}", remoteFileName);
+        throw new DownloadFailedException("Failed to download the file: " + remoteFileName);
+      }
+
+      return localFilePath;
+
+    } catch (IOException ex) {
+      logger.error("failed to retrieve file from ftp", ex);
+      throw new DownloadFailedException("Failed to download the file: " + remoteFileName);
+
+    } finally {
+      logger.info("                          ... done");
+
+      if (localFileOutputStream != null) {
+        try {
+          localFileOutputStream.close();
+        } catch (IOException e) {
+          logger.error("failed to close local file output stream", e);
+        }
+      }
+    }
+
+  }
+
+  private FTPClient connect(String host) throws IOException {
+    FTPClient ftpClient = new FTPClient();
+    ftpClient.setConnectTimeout((int)GtfsCollectorConfiguration.getGtfsConnectTimeoutMilliseconds());
+    ftpClient.connect(host);
+    ftpClient.login("anonymous", "");
+    ftpClient.enterLocalPassiveMode();
+    ftpClient.setFileType(FTP.BINARY_FILE_TYPE);
+
+    if (!FTPReply.isPositiveCompletion(ftpClient.getReplyCode())) {
+      throw new IOException("Faild to connect to: " + host);
+    }
+
+    return ftpClient;
+  }
+
+}

siri/gtfs_reader/src/main/java/il/org/hasadna/siri_client/gtfs/crud/GtfsCrud.java

@@ -14,7 +14,6 @@
  */
 public class GtfsCrud {
 
-	GtfsZipFile gtfsZipFile;
 	private Crud<Trip> tripCrud;
 	private Crud<Calendar> calendarCrud;
 	private Crud<StopTime> stopTimesCrud;