Commit before actually start processing the TIPLOCs

src/main/java/org/leolo/nrdatad/cron/ScheduleImportJob.java

@@ -77,6 +77,8 @@ public class ScheduleImportJob implements Job {
                         log.atInfo().log("New batch for tiploc.");
                         pt.flush();
                         pt.close();
+                        final int OLD_BATCH = batchT;
+                        threadPool.execute(()->{processTiploc("T_"+OLD_BATCH);});
                         pt = new PrintWriter(new GZIPOutputStream(new FileOutputStream(new File(baseTempDir, "T_"+(++batchT)))));
                     }
                 } else if (TYPE_SCHEDULE.equals(objectType)) {
@@ -101,6 +103,8 @@ public class ScheduleImportJob implements Job {
             ps.close();
             final int OLD_BATCH_A = batchA;
             threadPool.execute(()->{processAssociation("A_"+OLD_BATCH_A);});
+            final int OLD_BATCH_T = batchT;
+            threadPool.execute(()->{processTiploc("T_"+OLD_BATCH_T);});
             log.atInfo().log("There are {} records on total. Including {} headers, {} association, {} TIPLOC and {} schedules", count, countH, countA, countT, countS);
         }catch (IOException | URISyntaxException e){
             log.atError().withThrowable(e).log("Unable to read SCHEDULE feed");
@@ -118,4 +122,9 @@ public class ScheduleImportJob implements Job {
         log.atDebug().log("Processing {}",file);
 
     }
+
+    private void processTiploc(String file){
+        log.atDebug().log("Processing {}",file);
+
+    }
 }

src/main/java/org/leolo/nrdatad/db/TiplocDao.java

@@ -19,5 +19,23 @@ public abstract class TiplocDao extends BaseDao{
         }
     }
 
+    public abstract boolean hasTiploc(String tiplocCode) throws SQLException;
+
+    public void replace(Tiploc tiploc) throws SQLException{
+        if(hasTiploc(tiploc.getTiplocCode())){
+            update(tiploc);
+        } else {
+            insert(tiploc);
+        }
+    }
+
+    public abstract void update(Tiploc tiploc) throws SQLException;
+
+    public void replaceAll(Collection<Tiploc> tiplocs) throws SQLException{
+        for(Tiploc tiploc:tiplocs){
+            replace(tiploc);
+        }
+    }
+
     public abstract void truncateTable() throws SQLException;
 }

src/main/java/org/leolo/nrdatad/db/mariadb/TiplocDaoImpl.java

@@ -6,10 +6,8 @@ import org.leolo.nrdatad.db.DatabaseManager;
 import org.leolo.nrdatad.db.TiplocDao;
 import org.leolo.nrdatad.model.Tiploc;
 
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
-import java.sql.Statement;
+import java.sql.*;
+import java.util.Collection;
 
 public class TiplocDaoImpl extends TiplocDao {
 
@@ -21,10 +19,10 @@ public class TiplocDaoImpl extends TiplocDao {
 
     @Override
     public void insert(Tiploc tiploc) throws SQLException {
-        try(
+        try (
                 Connection conn = getConnection();
                 PreparedStatement pstmt = conn.prepareStatement("INSERT INTO tiploc (tiploc_code, nalco, stanox, crs_code, description, tps_description) VALUES (?,?,?,?,?,?)")
-        ){
+        ) {
             setString(pstmt, 1, tiploc.getTiplocCode());
             setString(pstmt, 2, tiploc.getNalco());
             setString(pstmt, 3, tiploc.getStanox());
@@ -36,6 +34,90 @@ public class TiplocDaoImpl extends TiplocDao {
         }
     }
 
+
+    @Override
+    public void insertAll(Collection<Tiploc> tiplocs) throws SQLException {
+        try (
+                Connection conn = getConnection();
+                PreparedStatement pstmt = conn.prepareStatement("INSERT INTO tiploc (tiploc_code, nalco, stanox, crs_code, description, tps_description) VALUES (?,?,?,?,?,?)")
+        ) {
+            for(Tiploc tiploc:tiplocs) {
+                setString(pstmt, 1, tiploc.getTiplocCode());
+                setString(pstmt, 2, tiploc.getNalco());
+                setString(pstmt, 3, tiploc.getStanox());
+                setString(pstmt, 4, tiploc.getCrsCode());
+                setString(pstmt, 5, tiploc.getDescription());
+                setString(pstmt, 6, tiploc.getTpsDescription());
+                pstmt.addBatch();
+            }
+            pstmt.executeBatch();
+            conn.commit();
+        }
+    }
+
+    @Override
+    public void replaceAll(Collection<Tiploc> tiplocs) throws SQLException {
+        try (
+                Connection conn = getConnection();
+                PreparedStatement psIns = conn.prepareStatement("INSERT INTO tiploc (tiploc_code, nalco, stanox, crs_code, description, tps_description) VALUES (?,?,?,?,?,?)");
+                PreparedStatement psUpd = conn.prepareStatement("UPDATE tiploc SET nalco=?, stanox=?, crs_code=?, description=?, tps_description=? WHERE tiploc_code=?")
+        ) {
+            for(Tiploc tiploc:tiplocs) {
+                if(hasTiploc(tiploc.getTiplocCode())){
+                    setString(psUpd, 1, tiploc.getNalco());
+                    setString(psUpd, 2, tiploc.getStanox());
+                    setString(psUpd, 3, tiploc.getCrsCode());
+                    setString(psUpd, 4, tiploc.getDescription());
+                    setString(psUpd, 5, tiploc.getTpsDescription());
+                    setString(psUpd, 6, tiploc.getTiplocCode());
+                    psUpd.addBatch();
+
+                } else {
+                    setString(psIns, 1, tiploc.getTiplocCode());
+                    setString(psIns, 2, tiploc.getNalco());
+                    setString(psIns, 3, tiploc.getStanox());
+                    setString(psIns, 4, tiploc.getCrsCode());
+                    setString(psIns, 5, tiploc.getDescription());
+                    setString(psIns, 6, tiploc.getTpsDescription());
+                    psIns.addBatch();
+                }
+            }
+            psIns.executeBatch();
+            psUpd.executeBatch();
+            conn.commit();
+        }
+    }
+
+    @Override
+    public boolean hasTiploc(String tiplocCode) throws SQLException {
+        try(
+                Connection conn = getConnection();
+                PreparedStatement pstmt = conn.prepareStatement("SELECT 1 FROM tiploc WHERE tiploc_code = ?")
+        ){
+            pstmt.setString(1, tiplocCode);
+            try(ResultSet rs = pstmt.executeQuery()){
+                return rs.next();
+            }
+        }
+    }
+
+    @Override
+    public void update(Tiploc tiploc) throws SQLException {
+        try (
+                Connection conn = getConnection();
+                PreparedStatement pstmt = conn.prepareStatement("UPDATE tiploc SET nalco=?, stanox=?, crs_code=?, description=?, tps_description=? WHERE tiploc_code=?")
+        ) {
+            setString(pstmt, 1, tiploc.getNalco());
+            setString(pstmt, 2, tiploc.getStanox());
+            setString(pstmt, 3, tiploc.getCrsCode());
+            setString(pstmt, 4, tiploc.getDescription());
+            setString(pstmt, 5, tiploc.getTpsDescription());
+            setString(pstmt, 6, tiploc.getTiplocCode());
+            pstmt.executeUpdate();
+            conn.commit();
+        }
+    }
+
     @Override
     public void truncateTable() throws SQLException{
         try(