Reworked to use servlet 3.1 async read support.

Adam Heath
Showing 1 changed file with 202 additions and 49 deletions
src/com/brainfood/ofbiz/DirectControlServlet.java
--- a/src/com/brainfood/ofbiz/DirectControlServlet.java
View file @95b259c
+++ b/src/com/brainfood/ofbiz/DirectControlServlet.java
View file @95b259c
 package com.brainfood.ofbiz;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.PrintWriter;
+import java.io.BufferedOutputStream;
 import java.io.BufferedReader;
 import java.io.StringReader;
 import java.io.InputStreamReader;
 import java.io.InputStream;
 import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.OutputStream;
 import java.util.Collections;
 import java.util.Collection;
 import java.util.Map;
 import java.net.URL;
 import java.net.MalformedURLException;
+import java.security.DigestOutputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Set;
 import java.util.Date;
@@ -26,10 +35,17 @@ import java.util.Iterator;
 import java.sql.Timestamp;
 import javax.script.ScriptContext;
+import javax.servlet.AsyncContext;
+import javax.servlet.AsyncEvent;
+import javax.servlet.AsyncListener;
+import javax.servlet.ReadListener;
 import javax.servlet.RequestDispatcher;
 import javax.servlet.ServletConfig;
 import javax.servlet.ServletContext;
 import javax.servlet.ServletException;
+import javax.servlet.ServletInputStream;
+import javax.servlet.ServletOutputStream;
+import javax.servlet.WriteListener;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -38,6 +54,7 @@ import javax.servlet.http.Cookie;
 import javax.servlet.http.Part;
 import org.ofbiz.base.lang.JSON;
+import org.ofbiz.base.util.Base64;
 import org.ofbiz.base.util.Debug;
 import org.ofbiz.base.util.GroovyUtil;
 import org.ofbiz.base.util.ScriptHelper;
@@ -62,11 +79,11 @@ import org.ofbiz.service.ServiceContainer;
 import org.apache.commons.csv.CSVFormat;
 import org.apache.commons.csv.CSVRecord;
+import org.apache.commons.fileupload.MultipartStream;
-import org.apache.commons.fileupload.*;
+import org.apache.commons.fileupload.ParameterParser;
-import org.apache.commons.fileupload.disk.DiskFileItemFactory;
+import org.apache.http.HeaderElement;
-import org.apache.commons.fileupload.disk.DiskFileItem;
+import org.apache.http.NameValuePair;
-import org.apache.commons.fileupload.servlet.ServletFileUpload;
+import org.apache.http.message.BasicHeaderValueParser;
 import groovy.lang.GroovyClassLoader;
 import groovy.lang.Script;
@@ -157,9 +174,98 @@ public class DirectControlServlet extends HttpServlet {
                    return;
                }
            }
-            Debug.logInfo("Service name " + serviceName, module);
+            Debug.logInfo("Service name %s" +serviceName + " mapped for " + pathInfo + "#" + method, module, serviceName, pathInfo);
+            AsyncContext context = request.startAsync();
+            AsyncHandler asyncHandler = new AsyncHandler(serviceName, request, response, context);
+            context.addListener(asyncHandler);
+            asyncHandler.sin.setReadListener(asyncHandler);
+        } catch (Throwable t) {
+            response.setStatus(500);
+            PrintWriter writer = response.getWriter();
+            Debug.logInfo("Exception processing request", module);
+            Debug.logInfo(t, module);
+            while (t != null) {
+                t.printStackTrace(writer);
+                t = t.getCause();
+            }
+            writer.flush();
+            writer.close();
+        }
+    }
+    protected interface Cleanup {
+        void cleanup();
+    }
+    protected class AsyncHandler implements ReadListener, WriteListener, AsyncListener {
+        protected final String serviceName;
+        protected final String outputHandler;
+        protected final File directory;
+        protected final HttpServletRequest request;
+        protected final HttpServletResponse response;
+        protected final ServletInputStream sin;
+        protected final AsyncContext context;
+        protected final File tmpFile;
+        protected final OutputStream tmpOut;
+        protected final List<Cleanup> cleanups = new ArrayList<Cleanup>();
+        protected AsyncHandler(String serviceName, HttpServletRequest request, HttpServletResponse response, AsyncContext context) throws IOException, NoSuchAlgorithmException {
+            this.directory = (File) request.getServletContext().getAttribute("javax.servlet.context.tempdir");
+            this.request = request;
+            this.response = response;
+            this.sin = request.getInputStream();
+            this.context = context;
+            this.tmpFile = deleteFileWhenDone(File.createTempFile("dcs", ".body", directory));
+            this.tmpOut = new BufferedOutputStream(new FileOutputStream(tmpFile));
+            String outputHandler = "JSON";
+            if (serviceName.indexOf("|") != -1) {
+                String[] parts = serviceName.split("\\|");
+                serviceName = parts[0];
+                outputHandler = parts[1];
+            }
+            this.serviceName = serviceName;
+            this.outputHandler = outputHandler;
+        }
+        public void onAllDataRead() throws IOException {
+            //System.err.println("onAllDataRead");
+            sin.close();
+            tmpOut.flush();
+            tmpOut.close();
+            try {
+                processRequest();
+            } catch (IOException e) {
+                throw e;
+            } catch (Exception e) {
+                e.printStackTrace();
+                throw (IOException) new IOException(e.getMessage()).initCause(e);
+            } finally {
+                context.complete();
+            }
+        }
-            // Load context
+        protected File deleteFileWhenDone(final File file) {
+            file.deleteOnExit();
+            cleanups.add(new Cleanup() {
+                public void cleanup() {
+                    file.delete();
+                }
+            });
+            return file;
+        }
+        protected void processRequest() throws Exception {
            Map<String, Object> context = new HashMap<String, Object>();
            // Directly copy request parameters into context.
@@ -172,21 +278,23 @@ public class DirectControlServlet extends HttpServlet {
            // Determine type of request and load the context from JSON content or
            // parameter values accordingly
-            if (contentType != null) {
+            String rawContentType = request.getContentType();
-                int semi = contentType.indexOf(";");
+            String contentType = rawContentType;
+            if (rawContentType != null) {
+                int semi = rawContentType.indexOf(";");
                if (semi != -1) {
-                    contentType = contentType.substring(0, semi);
+                    contentType = rawContentType.substring(0, semi);
                }
            }
            if ("application/json".equals(contentType)) {
                // Read request body as JSON and insert into the context
-                Map<String, Object> items = UtilGenerics.cast(JSON.from(UtilIO.readString(request.getReader())).toObject(Map.class));
+                Map<String, Object> items = UtilGenerics.cast(JSON.from(UtilIO.readString(new FileReader(tmpFile))).toObject(Map.class));
                for (Object key : items.keySet()) {
                    context.put((String) key, items.get(key));
                }
            } else if ("text/csv".equals(contentType)) {
-                Iterable<CSVRecord> records = CSVFormat.EXCEL.parse(request.getReader());
+                Iterable<CSVRecord> records = CSVFormat.EXCEL.parse(new FileReader(tmpFile));
                List<List<String>> data = new ArrayList<List<String>>();
                for (CSVRecord record : records) {
                    List<String> row = new ArrayList<String>();
@@ -198,30 +306,55 @@ public class DirectControlServlet extends HttpServlet {
                    data.add(row);
                }
                context.put("data", data);
-            } else if (contentType != null && contentType.indexOf("multipart/form-data") != -1) {
+            } else if (rawContentType != null && rawContentType.indexOf("multipart/form-data") != -1) {
-                // Create a factory for disk-based file items
+                //System.err.println("contentType[" + rawContentType + "]");
-                DiskFileItemFactory factory = new DiskFileItemFactory();
+                String boundary = rawContentType.substring("multipart/form-data; boundary=".length());
+                MessageDigest digest = MessageDigest.getInstance("SHA-256");
-                // Configure a repository (to ensure a secure temp location is used)
+                MultipartStream multipartStream = new MultipartStream(new FileInputStream(tmpFile), boundary.getBytes(), 65536, null);
-                ServletContext servletContext = config.getServletContext();
+                ParameterParser parameterParser = new ParameterParser();
-                File repository = (File) servletContext
-                        .getAttribute("javax.servlet.context.tempdir");
+                boolean hasNext = multipartStream.skipPreamble();
-                factory.setRepository(repository);
+                while (hasNext) {
+                    String[] headerLines = multipartStream.readHeaders().split("[\r\n]+(?! )");
-                ServletFileUpload upload = new ServletFileUpload(factory);
+                    Map<String, String> headerData = new HashMap<String, String>();
+                    for (String headerLine: headerLines) {
-                //Map<String, FileItem> itemMap = new HashMap<String, FileItem>();
+                        //System.err.println("headerLine[" + headerLine + "]");
+                        String[] headerNameAndRawValue = headerLine.split(": ", 2);
-                List<FileItem> items = (List<FileItem>)upload.parseRequest(request);
+                        headerData.put(headerNameAndRawValue[0].toLowerCase(), headerNameAndRawValue[1]);
-                for (FileItem item : items) {
+                        HeaderElement[] headerElements = BasicHeaderValueParser.INSTANCE.parseElements(headerNameAndRawValue[1], null);
-                    Debug.logInfo("PART: " + item.getFieldName(), module);
+                        for (HeaderElement headerElement: headerElements) {
-                    context.put(item.getFieldName(), item.getString());
+                            String headerName = headerElement.getName();
+                            String prefix = (headerNameAndRawValue[0] + "." + headerName).toLowerCase();
+                            headerData.put(prefix, headerElement.getValue());
+                            for (NameValuePair nameValuePair: headerElement.getParameters()) {
+                                headerData.put(prefix + "." + nameValuePair.getName(), nameValuePair.getValue());
+                            }
+                        }
+                    }
+                    //System.err.println("header=" + JSON.from(headerData).toString());
+                    String fieldName = headerData.get("content-disposition.form-data.name");
+                    String fileName = headerData.get("content-disposition.form-data.filename");
+                    if (fileName == null) {
+                        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                        multipartStream.readBodyData(baos);
+                        context.put(fieldName, baos.toString());
+                    } else {
+                        File tmpFilePart = deleteFileWhenDone(File.createTempFile("dcs", ".part", directory));
+                        digest.reset();
+                        DigestOutputStream digestOut = new DigestOutputStream(new FileOutputStream(tmpFilePart), digest);
+                        multipartStream.readBodyData(digestOut);
+                        context.put(fieldName, tmpFilePart);
+                        String sha256 = new String(Base64.base64Encode(digest.digest()));
+                        context.put(fieldName + ".contentType", headerData.get("content-type"));
+                        context.put(fieldName + ".sha256", sha256);
+                    }
+                    hasNext = multipartStream.readBoundary();
                }
            } else {
                // Check if the request is a backbone style "emulateJSON" request
                if (contentType != null && contentType.indexOf("x-www-form-urlencoded") != -1 && request.getParameter("model") != null) {
                    Debug.logInfo("MODEL: " + request.getParameter("model"), module);
-                    Map<String, Object> items = UtilGenerics.cast(JSON.from(UtilIO.readString(request.getReader())).toObject(Map.class));
+                    Map<String, Object> items = UtilGenerics.cast(JSON.from(UtilIO.readString(new FileReader(tmpFile))).toObject(Map.class));
                    for (Object key : items.keySet()) {
                        if (!"sessionId".equals(key)) {
                            context.put((String) key, items.get(key));
@@ -234,14 +367,6 @@ public class DirectControlServlet extends HttpServlet {
            LocalDispatcher dispatcher = getDispatcher(request.getServletContext());
            // Check if there is an output handler
-            String outputHandler = "JSON";
-            if (serviceName.indexOf("|") != -1) {
-                String[] parts = serviceName.split("\\|");
-                serviceName = parts[0];
-                outputHandler = parts[1];
-            }
-            Debug.logInfo("Service name" +serviceName + " mapped for " + pathInfo + "#" + method, module);
            // If the sessionId parameter is set, attempt to look up the corresponding
            // UserLogin and apply it to the service context
@@ -373,17 +498,45 @@ public class DirectControlServlet extends HttpServlet {
            if ("PDF".equals(outputHandler)) {
                LibreOfficeRenderer.service(request, response, result);
            }
-        } catch (Throwable t) {
-            response.setStatus(500);
+            //sout.setWriteListener(this);
-            PrintWriter writer = response.getWriter();
-            Debug.logInfo("Exception processing request", module);
-            Debug.logInfo(t, module);
-            while (t != null) {
-                t.printStackTrace(writer);
-                t = t.getCause();
        }
-            writer.flush();
-            writer.close();
+        public void onDataAvailable() throws IOException {
+            byte[] buffer = new byte[65536];
+            int numRead;
+            while (sin.isReady() && (numRead = sin.read(buffer)) > 0) {
+                tmpOut.write(buffer, 0, numRead);
+            }
+        }
+        public void onWritePossible() throws IOException {
+            context.complete();
+        }
+        public void onError(Throwable t) {
+            t.printStackTrace();
+            context.complete();
+        }
+        public void onComplete(AsyncEvent event) throws IOException {
+            //System.err.println("onComplete");
+            for (Cleanup cleanup: cleanups) {
+                cleanup.cleanup();
+            }
+        }
+        public void onError(AsyncEvent event) {
+            event.getThrowable().printStackTrace();
+        }
+        public void onStartAsync(AsyncEvent event) {
+            //System.err.println("onStartAsync");
+        }
+        public void onTimeout(AsyncEvent event) {
+            ////System.err.println("onTimeout");
        }
    }