Removed PosgresQueryQueue and refactored PosgresProfileStreamer to us…

…e shared PostgresProfileQueryHandler

merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/postgres/PostgresProfileQueryHandler.java

@@ -25,29 +25,22 @@
 import static gov.nasa.jpl.aerie.merlin.server.remotes.postgres.PostgresParsers.realProfileTypeP;
 
 /**
- * Utility class to handle upload of resource profiles to the database. This class allows for each upload to have its
- * own Connection object.
+ * Utility class to handle upload of resource profiles to the database.
  * */
 public class PostgresProfileQueryHandler implements AutoCloseable {
   private final Connection connection;
   private final HashMap<String, Integer> profileIds;
   private final HashMap<String, Duration> profileDurations;
 
-  private PreparedStatement postProfileStatement;
-  private PreparedStatement postSegmentsStatement;
-  private PreparedStatement updateDurationStatement;
-
-  private long datasetId;
+  private final PreparedStatement postProfileStatement;
+  private final PreparedStatement postSegmentsStatement;
+  private final PreparedStatement updateDurationStatement;
 
   public PostgresProfileQueryHandler(DataSource dataSource, long datasetId) throws SQLException {
-    this.connection = dataSource.getConnection();
+    connection = dataSource.getConnection();
     profileIds = new HashMap<>();
     profileDurations = new HashMap<>();
-    this.datasetId = datasetId;
-    prepareStatements();
-  }
 
-  public void prepareStatements() throws SQLException {
     final String postProfilesSql =
         //language=sql
         """

merlin-worker/src/main/java/gov/nasa/jpl/aerie/merlin/worker/postgres/PostgresProfileStreamer.java

@@ -5,33 +5,34 @@
 
 import javax.sql.DataSource;
 import java.sql.SQLException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.function.Consumer;
 
 public class PostgresProfileStreamer implements Consumer<ResourceProfiles>, AutoCloseable {
-  private final DataSource dataSource;
-  private long datasetId;
-  private PostgresQueryQueue queryQueue;
+  private final ExecutorService queryQueue;
+  private final PostgresProfileQueryHandler queryHandler;
 
   public PostgresProfileStreamer(DataSource dataSource, long datasetId) throws SQLException {
-    this.dataSource = dataSource;
-    this.datasetId = datasetId;
-    this.queryQueue = new PostgresQueryQueue();
+    this.queryQueue = Executors.newSingleThreadExecutor();
+    this.queryHandler = new PostgresProfileQueryHandler(dataSource, datasetId);
   }
 
   @Override
   public void accept(final ResourceProfiles resourceProfiles) {
-    queryQueue.addToQueue(() -> {
-      try (var transaction = new PostgresProfileQueryHandler(dataSource, datasetId)) {
-          transaction.uploadResourceProfiles(resourceProfiles);
-      } catch (SQLException e) {
-        throw new DatabaseException("Exception occurred while posting profiles.", e);
-      }
+    queryQueue.submit(() -> {
+      queryHandler.uploadResourceProfiles(resourceProfiles);
     });
   }
 
   @Override
   public void close() {
     queryQueue.shutdown();
+      try {
+          queryHandler.close();
+      } catch (SQLException e) {
+          throw new DatabaseException("Error occurred while attempting to close PostgresProfileQueryHandler", e);
+      }
   }
 
 }