[OLINGO-708] Added AsyncResponseSerializer

lib/server-api/src/main/java/org/apache/olingo/server/api/serializer/BatchSerializerException.java

@@ -23,7 +23,7 @@ public class BatchSerializerException extends SerializerException {
 
   private static final long serialVersionUID = 2634433974342796905L;
 
-  public static enum MessageKeys implements MessageKey {
+  public enum MessageKeys implements MessageKey {
     MISSING_CONTENT_ID;
 
     @Override

lib/server-api/src/main/java/org/apache/olingo/server/api/serializer/FixedFormatSerializer.java

@@ -22,6 +22,7 @@ import java.io.InputStream;
 import java.util.List;
 
 import org.apache.olingo.commons.api.edm.EdmPrimitiveType;
+import org.apache.olingo.server.api.ODataResponse;
 import org.apache.olingo.server.api.deserializer.batch.ODataResponsePart;
 
 /** OData serializer for fixed output formats. */
@@ -55,4 +56,11 @@ public interface FixedFormatSerializer {
    * @return response as an input stream
    */
   InputStream batchResponse(List<ODataResponsePart> batchResponses, String boundary) throws BatchSerializerException;
+
+  /**
+   * Serializes a ODataResponse into an async response.
+   * @param odataResponse the response parts
+   * @return response as an input stream
+   */
+  InputStream asyncResponse(ODataResponse odataResponse) throws SerializerException;
 }

lib/server-api/src/main/java/org/apache/olingo/server/api/serializer/SerializerException.java

@@ -26,7 +26,7 @@ public class SerializerException extends ODataLibraryException {
   private static final long serialVersionUID = 5358683245923127425L;
 
   /** Keys for exception texts in the resource bundle. */
-  public static enum MessageKeys implements MessageKey {
+  public enum MessageKeys implements MessageKey {
     NOT_IMPLEMENTED,
     /** parameter: format */
     UNSUPPORTED_FORMAT,

lib/server-core/src/main/java/org/apache/olingo/server/core/serializer/AsyncResponseSerializer.java

@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.olingo.server.core.serializer;
+
+import org.apache.olingo.commons.api.http.HttpStatusCode;
+import org.apache.olingo.server.api.ODataResponse;
+import org.apache.olingo.server.api.serializer.SerializerException;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.Map;
+
+public class AsyncResponseSerializer {
+  private static final int BUFFER_SIZE = 4096;
+  private static final String COLON = ":";
+  private static final String SP = " ";
+  private static final String CRLF = "\r\n";
+  public static final String HEADER_CHARSET_NAME = "ISO-8859-1";
+  public static final String HTTP_VERSION = "HTTP/1.1";
+
+  public InputStream serialize(final ODataResponse response) throws SerializerException {
+    try {
+      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+      appendStatusLine(response, buffer);
+      appendResponseHeader(response, buffer);
+      append(CRLF, buffer);
+      appendBody(response, buffer);
+
+      buffer.flush();
+      return new ByteArrayInputStream(buffer.toByteArray(), 0, buffer.size());
+    } catch (IOException e) {
+      throw new SerializerException("Exception occurred during serialization of asynchronous response.",
+          e, SerializerException.MessageKeys.IO_EXCEPTION);
+    }
+  }
+
+  private void appendResponseHeader(final ODataResponse response,
+                                    final ByteArrayOutputStream buffer) throws IOException {
+    final Map<String, String> header = response.getHeaders();
+
+    for (final String key: header.keySet()) {
+      appendHeader(key, header.get(key), buffer);
+    }
+  }
+
+  private void appendHeader(final String name, final String value, final ByteArrayOutputStream buffer)
+      throws IOException {
+    append(name + COLON + SP + value + CRLF, buffer);
+  }
+
+  private void appendStatusLine(final ODataResponse response, final ByteArrayOutputStream buffer)
+      throws IOException {
+    HttpStatusCode status = HttpStatusCode.fromStatusCode(response.getStatusCode());
+    append(HTTP_VERSION + SP + response.getStatusCode() + SP + status + CRLF, buffer);
+  }
+
+  private void appendBody(ODataResponse response, ByteArrayOutputStream buffer) throws IOException {
+    InputStream input = response.getContent();
+    if (input != null) {
+      ByteBuffer inBuffer = ByteBuffer.allocate(BUFFER_SIZE);
+      ReadableByteChannel ic = Channels.newChannel(input);
+      WritableByteChannel oc = Channels.newChannel(buffer);
+      while (ic.read(inBuffer) > 0) {
+        inBuffer.flip();
+        oc.write(inBuffer);
+        inBuffer.rewind();
+      }
+    }
+  }
+
+  private void append(final String value, final ByteArrayOutputStream buffer) throws IOException {
+    try {
+      buffer.write(value.getBytes(HEADER_CHARSET_NAME));
+    } catch (UnsupportedEncodingException e) {
+      throw new IOException("Default header charset with name '" + HEADER_CHARSET_NAME +
+          "' is not available.", e);
+    }
+  }
+}

lib/server-core/src/main/java/org/apache/olingo/server/core/serializer/FixedFormatSerializerImpl.java

@@ -25,6 +25,7 @@ import java.util.List;
 
 import org.apache.olingo.commons.api.edm.EdmPrimitiveType;
 import org.apache.olingo.commons.api.edm.EdmPrimitiveTypeException;
+import org.apache.olingo.server.api.ODataResponse;
 import org.apache.olingo.server.api.deserializer.batch.ODataResponsePart;
 import org.apache.olingo.server.api.serializer.BatchSerializerException;
 import org.apache.olingo.server.api.serializer.FixedFormatSerializer;
@@ -60,6 +61,12 @@ public class FixedFormatSerializerImpl implements FixedFormatSerializer {
     }
   }
 
+  @Override
+  public InputStream asyncResponse(ODataResponse odataResponse) throws SerializerException {
+    AsyncResponseSerializer serializer = new AsyncResponseSerializer();
+    return serializer.serialize(odataResponse);
+  }
+
   // TODO: Signature refactoring for writeBatchResponse
   @Override
   public InputStream batchResponse(final List<ODataResponsePart> batchResponses, final String boundary)

lib/server-core/src/test/java/org/apache/olingo/server/core/serializer/AsyncResponseSerializerTest.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.olingo.server.core.serializer;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.olingo.commons.api.format.ContentType;
+import org.apache.olingo.commons.api.http.HttpHeader;
+import org.apache.olingo.commons.api.http.HttpStatusCode;
+import org.apache.olingo.server.api.ODataResponse;
+import org.junit.Test;
+
+import java.io.InputStream;
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+
+public class AsyncResponseSerializerTest {
+  private static final String CRLF = "\r\n";
+
+  @Test
+  public void testSimpleResponse() throws Exception {
+    ODataResponse response = new ODataResponse();
+    response.setStatusCode(HttpStatusCode.OK.getStatusCode());
+    response.setHeader(HttpHeader.CONTENT_TYPE, ContentType.APPLICATION_JSON.toContentTypeString());
+
+    response.setContent(IOUtils.toInputStream("Walter Winter" + CRLF));
+
+    AsyncResponseSerializer serializer = new AsyncResponseSerializer();
+    InputStream in = serializer.serialize(response);
+    String result = IOUtils.toString(in);
+    assertEquals("HTTP/1.1 200 OK" + CRLF +
+        "Content-Type: application/json" + CRLF + CRLF +
+        "Walter Winter" + CRLF, result);
+  }
+
+  @Test
+  public void testBiggerResponse() throws Exception {
+    ODataResponse response = new ODataResponse();
+    response.setStatusCode(HttpStatusCode.ACCEPTED.getStatusCode());
+    response.setHeader(HttpHeader.CONTENT_TYPE, ContentType.APPLICATION_JSON.toContentTypeString());
+
+    String testData = testData(20000);
+    response.setContent(IOUtils.toInputStream(testData));
+
+    AsyncResponseSerializer serializer = new AsyncResponseSerializer();
+    InputStream in = serializer.serialize(response);
+    String result = IOUtils.toString(in);
+    assertEquals("HTTP/1.1 202 Accepted" + CRLF +
+        "Content-Type: application/json" + CRLF + CRLF +
+        testData, result);
+  }
+
+  private String testData(int amount) {
+    StringBuilder result = new StringBuilder();
+    Random r = new Random();
+    for (int i = 0; i < amount; i++) {
+      result.append((char)(r.nextInt(26) + 'a'));
+    }
+
+    return result.toString();
+  }
+}

lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServlet.java

@@ -64,12 +64,6 @@ public class TechnicalServlet extends HttpServlet {
   protected void service(final HttpServletRequest request, final HttpServletResponse response)
       throws ServletException, IOException {
     try {
-      if(TechnicalAsyncService.getInstance().isStatusMonitorResource(request)) {
-        TechnicalAsyncService asyncService = TechnicalAsyncService.getInstance();
-        asyncService.handle(request, response);
-        return;
-      }
-
       OData odata = OData.newInstance();
       EdmxReference reference = new EdmxReference(URI.create("../v4.0/cs02/vocabularies/Org.OData.Core.V1.xml"));
       reference.addInclude(new EdmxReferenceInclude("Org.OData.Core.V1", "Core"));

lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/async/TechnicalAsyncService.java

@@ -33,13 +33,16 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import org.apache.olingo.commons.api.ODataRuntimeException;
+import org.apache.olingo.commons.api.format.ContentType;
 import org.apache.olingo.commons.api.http.HttpHeader;
 import org.apache.olingo.commons.api.http.HttpStatusCode;
+import org.apache.olingo.server.api.OData;
 import org.apache.olingo.server.api.ODataApplicationException;
 import org.apache.olingo.server.api.ODataLibraryException;
 import org.apache.olingo.server.api.ODataRequest;
 import org.apache.olingo.server.api.ODataResponse;
 import org.apache.olingo.server.api.processor.Processor;
+import org.apache.olingo.server.api.serializer.SerializerException;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -90,7 +93,7 @@ public class TechnicalAsyncService {
     return location;
   }
 
-  public void handle(HttpServletRequest request, HttpServletResponse response) {
+  public void handle(HttpServletRequest request, HttpServletResponse response) throws SerializerException, IOException {
     String location = getAsyncLocation(request);
     AsyncRunner runner = LOCATION_2_ASYNC_RUNNER.get(location);
 
@@ -99,7 +102,7 @@ public class TechnicalAsyncService {
     } else {
       if(runner.isFinished()) {
         ODataResponse wrapResult = runner.getDispatched().getProcessResponse();
-        convertToHttp(response, wrapResult);
+        wrapToAsyncHttpResponse(wrapResult, response);
         LOCATION_2_ASYNC_RUNNER.remove(location);
       } else {
         response.setStatus(HttpStatusCode.ACCEPTED.getStatusCode());
@@ -110,11 +113,19 @@ public class TechnicalAsyncService {
     }
   }
 
+  private static void writeToResponse(HttpServletResponse response, InputStream input) throws IOException {
+    copy(input, response.getOutputStream());
+  }
+
   private void writeToResponse(HttpServletResponse response, String content) {
+    writeToResponse(response, content.getBytes());
+  }
+
+  private static void writeToResponse(HttpServletResponse response, byte[] content) {
     OutputStream output = null;
     try {
       output = response.getOutputStream();
-      output.write(content.getBytes());
+      output.write(content);
     } catch (IOException e) {
       throw new ODataRuntimeException(e);
     } finally {
@@ -122,32 +133,48 @@ public class TechnicalAsyncService {
     }
   }
 
-  static void convertToHttp(final HttpServletResponse response, final ODataResponse odResponse) {
+  static void wrapToAsyncHttpResponse(final ODataResponse odResponse, final HttpServletResponse response)
+      throws SerializerException, IOException {
+    OData odata = OData.newInstance();
+    InputStream odResponseStream = odata.createFixedFormatSerializer().asyncResponse(odResponse);
+
+    response.setHeader(HttpHeader.CONTENT_TYPE, ContentType.APPLICATION_HTTP.toContentTypeString());
+    response.setHeader(HttpHeader.CONTENT_ENCODING, "binary");
+    response.setStatus(HttpStatusCode.OK.getStatusCode());
+
+    writeToResponse(response, odResponseStream);
+  }
+
+  static void writeHttpResponse(final ODataResponse odResponse, final HttpServletResponse response) throws IOException {
     response.setStatus(odResponse.getStatusCode());
 
     for (Map.Entry<String, String> entry : odResponse.getHeaders().entrySet()) {
       response.setHeader(entry.getKey(), entry.getValue());
     }
 
-    InputStream input = odResponse.getContent();
+    copy(odResponse.getContent(), response.getOutputStream());
-    if (input != null) {
+  }
-      OutputStream output = null;
+
-      try {
+  static void copy(final InputStream input, final OutputStream output) {
-        output = response.getOutputStream();
+    if(output == null || input == null) {
-        byte[] buffer = new byte[1024];
+      return;
-        int n;
+    }
-        while (-1 != (n = input.read(buffer))) {
+
-          output.write(buffer, 0, n);
+    try {
-        }
+      byte[] buffer = new byte[1024];
-      } catch (IOException e) {
+      int n;
-        throw new ODataRuntimeException(e);
+      while (-1 != (n = input.read(buffer))) {
-      } finally {
+        output.write(buffer, 0, n);
-        closeStream(output);
-        closeStream(input);
       }
+    } catch (IOException e) {
+      throw new ODataRuntimeException(e);
+    } finally {
+      closeStream(output);
+      closeStream(input);
     }
   }
 
+
   private static void closeStream(final Closeable closeable) {
     if (closeable != null) {
       try {

lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/async/TechnicalStatusMonitorServlet.java

@@ -59,7 +59,7 @@ public class TechnicalStatusMonitorServlet extends HttpServlet {
         TechnicalAsyncService asyncService = TechnicalAsyncService.getInstance();
         asyncService.handle(request, response);
       }
-    } catch (final RuntimeException e) {
+    } catch (final Exception e) {
       LOG.error("Server Error", e);
       throw new ServletException(e);
     }