Cleaning up potential resource handling errors. (#1258)

Explicitly calling blob and clob free. Putting streams from blob and
clob in try-with-resources blocks. Calling jOOQ fetchStream in
try-with-resources.

I don't think any of these is actually causing connection leaks but they
seem like easy changes for better safety

(cherry picked from commit a660ea5)

Author: rma-rripken

cwms-data-api/src/main/java/cwms/cda/api/BinaryTimeSeriesValueController.java

@@ -39,7 +39,6 @@
 
 import javax.servlet.http.HttpServletResponse;
 import java.io.InputStream;
-import java.util.logging.Logger;
 
 import static com.codahale.metrics.MetricRegistry.name;
 import static cwms.cda.api.Controllers.*;
@@ -100,8 +99,9 @@ public void handle(Context ctx) {
                 } else {
                     long size = blob.length();
                     requestResultSize.update(size);
-                    InputStream is = blob.getBinaryStream();
-                    ctx.seekableStream(is, mediaType, size);
+                    try (InputStream is = blob.getBinaryStream()) {
+                        RangeRequestUtil.seekableStream(ctx, is, mediaType, size);
+                    }
                 }
             });
         }

cwms-data-api/src/main/java/cwms/cda/api/BlobController.java

@@ -1,6 +1,7 @@
 package cwms.cda.api;
 
 import static com.codahale.metrics.MetricRegistry.name;
+
 import static cwms.cda.api.Controllers.*;
 
 import com.codahale.metrics.Histogram;
@@ -28,12 +29,13 @@
 import java.io.InputStream;
 import java.util.List;
 import java.util.Optional;
+
 import javax.servlet.http.HttpServletResponse;
+
 import org.jetbrains.annotations.NotNull;
 import org.jooq.DSLContext;
 
 
-
 /**
  *
  */
@@ -62,32 +64,32 @@ protected DSLContext getDslContext(Context ctx) {
     }
 
     @OpenApi(
-        queryParams = {
-            @OpenApiParam(name = OFFICE,
-                    description = "Specifies the owning office. If this field is not "
-                            + "specified, matching information from all offices shall be "
-                            + "returned."),
-            @OpenApiParam(name = PAGE,
-                    description = "This end point can return a lot of data, this "
-                            + "identifies where in the request you are. This is an opaque"
-                            + " value, and can be obtained from the 'next-page' value in "
-                            + "the response."),
-            @OpenApiParam(name = PAGE_SIZE,
-                    type = Integer.class,
-                    description = "How many entries per page returned. Default "
-                            + DEFAULT_PAGE_SIZE + "."),
-            @OpenApiParam(name = LIKE,
-                    description = "Posix <a href=\"regexp.html\">regular expression</a> "
-                            + "describing the blob id's you want")
-        },
-        responses = {@OpenApiResponse(status = STATUS_200,
-                description = "A list of blobs.",
-                content = {
-                    @OpenApiContent(type = Formats.JSON, from = Blobs.class),
-                    @OpenApiContent(type = Formats.JSONV2, from = Blobs.class),
-                })
-        },
-        tags = {TAG}
+            queryParams = {
+                    @OpenApiParam(name = OFFICE,
+                            description = "Specifies the owning office. If this field is not "
+                                    + "specified, matching information from all offices shall be "
+                                    + "returned."),
+                    @OpenApiParam(name = PAGE,
+                            description = "This end point can return a lot of data, this "
+                                    + "identifies where in the request you are. This is an opaque"
+                                    + " value, and can be obtained from the 'next-page' value in "
+                                    + "the response."),
+                    @OpenApiParam(name = PAGE_SIZE,
+                            type = Integer.class,
+                            description = "How many entries per page returned. Default "
+                                    + DEFAULT_PAGE_SIZE + "."),
+                    @OpenApiParam(name = LIKE,
+                            description = "Posix <a href=\"regexp.html\">regular expression</a> "
+                                    + "describing the blob id's you want")
+            },
+            responses = {@OpenApiResponse(status = STATUS_200,
+                    description = "A list of blobs.",
+                    content = {
+                            @OpenApiContent(type = Formats.JSON, from = Blobs.class),
+                            @OpenApiContent(type = Formats.JSONV2, from = Blobs.class),
+                    })
+            },
+            tags = {TAG}
     )
     @Override
     public void getAll(@NotNull Context ctx) {
@@ -130,7 +132,7 @@ public void getAll(@NotNull Context ctx) {
             description = "Returns the binary value of the requested blob as a seekable stream with the "
                     + "appropriate media type.",
             queryParams = {
-                @OpenApiParam(name = OFFICE, description = "Specifies the owning office."),
+                    @OpenApiParam(name = OFFICE, description = "Specifies the owning office."),
             },
             tags = {TAG}
     )
@@ -151,8 +153,9 @@ public void getOne(@NotNull Context ctx, @NotNull String blobId) {
                 } else {
                     long size = blob.length();
                     requestResultSize.update(size);
-                    InputStream is = blob.getBinaryStream();
-                    ctx.seekableStream(is, mediaType, size);
+                    try (InputStream is = blob.getBinaryStream()) { // is  OracleBlobInputStream
+                        RangeRequestUtil.seekableStream(ctx, is, mediaType, size);
+                    }
                 }
             };
             if (office.isPresent()) {
@@ -168,12 +171,12 @@ public void getOne(@NotNull Context ctx, @NotNull String blobId) {
             description = "Create new Blob",
             requestBody = @OpenApiRequestBody(
                     content = {
-                        @OpenApiContent(from = Blob.class, type = Formats.JSONV2)
+                            @OpenApiContent(from = Blob.class, type = Formats.JSONV2)
                     },
                     required = true),
             queryParams = {
-                @OpenApiParam(name = FAIL_IF_EXISTS, type = Boolean.class,
-                        description = "Create will fail if provided ID already exists. Default: true")
+                    @OpenApiParam(name = FAIL_IF_EXISTS, type = Boolean.class,
+                            description = "Create will fail if provided ID already exists. Default: true")
             },
             method = HttpMethod.POST,
             tags = {TAG}
@@ -199,8 +202,8 @@ public void create(@NotNull Context ctx) {
             },
             requestBody = @OpenApiRequestBody(
                     content = {
-                        @OpenApiContent(from = Blob.class, type = Formats.JSONV2),
-                        @OpenApiContent(from = Blob.class, type = Formats.JSON)
+                            @OpenApiContent(from = Blob.class, type = Formats.JSONV2),
+                            @OpenApiContent(from = Blob.class, type = Formats.JSON)
                     },
                     required = true),
             method = HttpMethod.PATCH,
@@ -239,10 +242,10 @@ public void update(@NotNull Context ctx, @NotNull String blobId) {
     @OpenApi(
             description = "Deletes requested blob",
             pathParams = {
-                @OpenApiParam(name = BLOB_ID, description = "The blob identifier to be deleted"),
+                    @OpenApiParam(name = BLOB_ID, description = "The blob identifier to be deleted"),
             },
             queryParams = {
-                @OpenApiParam(name = OFFICE, required = true, description = "Specifies the "
+                    @OpenApiParam(name = OFFICE, required = true, description = "Specifies the "
                             + "owning office of the blob to be deleted"),
             },
             method = HttpMethod.DELETE,

cwms-data-api/src/main/java/cwms/cda/api/ClobController.java

@@ -26,6 +26,8 @@
 import org.jooq.DSLContext;
 
 import javax.servlet.http.HttpServletResponse;
+
+import java.io.InputStream;
 import java.util.Objects;
 import java.util.Optional;
 
@@ -175,7 +177,9 @@ public void getOne(@NotNull Context ctx, @NotNull String clobId) {
                         ctx.status(HttpServletResponse.SC_NOT_FOUND).json(new CdaError("Unable to find "
                                 + "clob based on given parameters"));
                     } else {
-                        ctx.seekableStream(c.getAsciiStream(), TEXT_PLAIN, c.length());
+                        try (InputStream is = c.getAsciiStream()) {
+                            RangeRequestUtil.seekableStream(ctx, is, TEXT_PLAIN, c.length());
+                        }
                     }
                 });
             } else {

cwms-data-api/src/main/java/cwms/cda/api/ForecastFileController.java

@@ -109,8 +109,9 @@ public void handle(Context ctx) {
                 } else {
                     long size = blob.length();
                     requestResultSize.update(size);
-                    InputStream is = blob.getBinaryStream();
-                    ctx.seekableStream(is, mediaType, size);
+                    try (InputStream is = blob.getBinaryStream()) {
+                        RangeRequestUtil.seekableStream(ctx, is, mediaType, size);
+                    }
                 }
             });
         }

cwms-data-api/src/main/java/cwms/cda/api/RangeRequestUtil.java

@@ -0,0 +1,96 @@
+package cwms.cda.api;
+
+import io.javalin.core.util.Header;
+import io.javalin.http.Context;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.List;
+
+public class RangeRequestUtil {
+
+    private RangeRequestUtil() {
+        // utility class
+    }
+
+    /**
+     * Javalin has a method very similar to this in its Context class.  The issue is that Javalin decided to
+     * take the InputStream, wrap it in a CompletedFuture and then process the request asynchronously.  This
+     * causes problems when the InputStream is tied to a database connection that gets closed before the
+     * async processing happens.  This method doesn't do the async thing but tries to support the rest.
+     * @param ctx
+     * @param is
+     * @param mediaType
+     * @param totalBytes
+     * @throws IOException
+     */
+    public static void seekableStream(Context ctx, InputStream is, String mediaType, long totalBytes) throws IOException {
+        long from = 0;
+        long to = totalBytes - 1;
+        if (ctx.header(Header.RANGE) == null) {
+            ctx.res.setContentType(mediaType);
+            // Javalin's version of this method doesn't set the content-length
+            // Not setting the content-length makes the servlet container use Transfer-Encoding=chunked.
+            // Chunked is a worse experience overall, seems like we should just set the length if we know it.
+            writeRange(ctx.res.getOutputStream(), is, from, Math.min(to, totalBytes - 1));
+        } else {
+            int chunkSize = 128000;
+            String rangeHeader = ctx.header(Header.RANGE);
+            String[] eqSplit = rangeHeader.split("=", 2);
+            String[] dashSplit = eqSplit[1].split("-", -1); // keep empty trailing part
+
+            List<String> requestedRange = Arrays.stream(dashSplit)
+                    .filter(s -> !s.isEmpty())
+                    .collect(java.util.stream.Collectors.toList());
+
+            from = Long.parseLong(requestedRange.get(0));
+
+            if (from + chunkSize > totalBytes) {
+                // chunk bigger than file, write all
+                to = totalBytes - 1;
+            } else if (requestedRange.size() == 2) {
+                // chunk smaller than file, to/from specified
+                to = Long.parseLong(requestedRange.get(1));
+            } else {
+                // chunk smaller than file, to/from not specified
+                to = from + chunkSize - 1;
+            }
+
+            ctx.status(206);
+
+            ctx.header(Header.ACCEPT_RANGES, "bytes");
+            ctx.header(Header.CONTENT_RANGE, "bytes " + from + "-" + to + "/" + totalBytes);
+
+            ctx.res.setContentType(mediaType);
+            ctx.header(Header.CONTENT_LENGTH, String.valueOf(Math.min(to - from + 1, totalBytes)));
+            writeRange(ctx.res.getOutputStream(), is, from, Math.min(to, totalBytes - 1));
+        }
+    }
+
+
+    public static void writeRange(OutputStream out, InputStream in, long from, long to) throws IOException {
+        writeRange(out, in, from, to, new byte[8192]);
+    }
+
+    public static void writeRange(OutputStream out, InputStream is, long from, long to, byte[] buffer) throws IOException {
+        long toSkip = from;
+        while (toSkip > 0) {
+            long skipped = is.skip(toSkip);
+            toSkip -= skipped;
+        }
+
+        long bytesLeft = to - from + 1;
+        while (bytesLeft != 0L) {
+            int maxRead = (int) Math.min(buffer.length, bytesLeft);
+            int read = is.read(buffer, 0, maxRead);
+            if (read == -1) {
+                break;
+            }
+            out.write(buffer, 0, read);
+            bytesLeft -= read;
+        }
+
+    }
+
+}

cwms-data-api/src/main/java/cwms/cda/api/TextTimeSeriesValueController.java

@@ -99,8 +99,9 @@ public void handle(Context ctx) {
                 } else {
                     long size = clob.length();
                     requestResultSize.update(size);
-                    InputStream is = clob.getAsciiStream();
-                    ctx.seekableStream(is, TEXT_PLAIN, size);
+                    try(InputStream is = clob.getAsciiStream()){
+                        RangeRequestUtil.seekableStream(ctx, is, TEXT_PLAIN, size);
+                    }
                 }
             });
         }