You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@olingo.apache.org by mi...@apache.org on 2015/06/23 15:49:37 UTC
[1/2] olingo-odata4 git commit: [OLINGO-708] Add
'isStatusMonitorResource'
Repository: olingo-odata4
Updated Branches:
refs/heads/OLINGO-708_AsyncSupport a61ddd83c -> b3f6f0602
[OLINGO-708] Add 'isStatusMonitorResource'
Project: http://git-wip-us.apache.org/repos/asf/olingo-odata4/repo
Commit: http://git-wip-us.apache.org/repos/asf/olingo-odata4/commit/0557f115
Tree: http://git-wip-us.apache.org/repos/asf/olingo-odata4/tree/0557f115
Diff: http://git-wip-us.apache.org/repos/asf/olingo-odata4/diff/0557f115
Branch: refs/heads/OLINGO-708_AsyncSupport
Commit: 0557f1157043364b49014e9bdb89208787ee45cc
Parents: a61ddd8
Author: Michael Bolz <mi...@sap.com>
Authored: Tue Jun 23 09:56:52 2015 +0200
Committer: Michael Bolz <mi...@sap.com>
Committed: Tue Jun 23 09:56:52 2015 +0200
----------------------------------------------------------------------
.../server/api/processor/AsynchronousProcessor.java | 4 +++-
.../org/apache/olingo/server/core/ODataDispatcher.java | 10 ++++------
.../server/tecsvc/processor/TechnicalAsyncProcessor.java | 5 +++++
3 files changed, 12 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/0557f115/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsynchronousProcessor.java
----------------------------------------------------------------------
diff --git a/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsynchronousProcessor.java b/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsynchronousProcessor.java
index 2bc147c..67eb8f6 100644
--- a/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsynchronousProcessor.java
+++ b/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsynchronousProcessor.java
@@ -43,7 +43,9 @@ public interface AsynchronousProcessor extends Processor {
void cancel(ODataRequest request, ODataResponse response) throws ODataApplicationException, ODataLibraryException;
- void result(ODataRequest request, ODataResponse response);
+ void result(ODataRequest request, ODataResponse response) throws ODataApplicationException, ODataLibraryException;
+
+ boolean isStatusMonitorResource(ODataRequest request) throws ODataApplicationException, ODataLibraryException;
// /**
// * Processes an asynchronous request.
// *
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/0557f115/lib/server-core/src/main/java/org/apache/olingo/server/core/ODataDispatcher.java
----------------------------------------------------------------------
diff --git a/lib/server-core/src/main/java/org/apache/olingo/server/core/ODataDispatcher.java b/lib/server-core/src/main/java/org/apache/olingo/server/core/ODataDispatcher.java
index 525fa87..4a1c2b2 100644
--- a/lib/server-core/src/main/java/org/apache/olingo/server/core/ODataDispatcher.java
+++ b/lib/server-core/src/main/java/org/apache/olingo/server/core/ODataDispatcher.java
@@ -86,12 +86,10 @@ public class ODataDispatcher {
public void dispatch(ODataRequest request, ODataResponse response) throws ODataApplicationException,
ODataLibraryException {
// XXX: only for POC reason
- if(request.getRawQueryPath() != null && request.getRawQueryPath().contains("async")) {
- AsynchronousProcessor asp = handler.getAsynchronousProcessor();
- if(asp != null) {
- asp.result(request, response);
- return;
- }
+ AsynchronousProcessor asp = handler.getAsynchronousProcessor();
+ if(asp != null && asp.isStatusMonitorResource(request)) {
+ asp.result(request, response);
+ return;
}
// XXX END
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/0557f115/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
----------------------------------------------------------------------
diff --git a/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
index a63449d..040f5f4 100644
--- a/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
+++ b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
@@ -41,6 +41,11 @@ public class TechnicalAsyncProcessor implements AsynchronousProcessor {
private static AtomicInteger ai = new AtomicInteger();
@Override
+ public boolean isStatusMonitorResource(ODataRequest request) throws ODataApplicationException, ODataLibraryException {
+ return request.getRawQueryPath() != null && request.getRawQueryPath().contains("async");
+ }
+
+ @Override
public void processAsynchronous(ODataRequest request, ODataResponse response, AsyncWrap dispatchedProcessor)
throws ODataApplicationException, ODataLibraryException {
String id = String.valueOf(ai.incrementAndGet());
[2/2] olingo-odata4 git commit: [OLINGO-708] Adapted test case
Posted by mi...@apache.org.
[OLINGO-708] Adapted test case
Project: http://git-wip-us.apache.org/repos/asf/olingo-odata4/repo
Commit: http://git-wip-us.apache.org/repos/asf/olingo-odata4/commit/b3f6f060
Tree: http://git-wip-us.apache.org/repos/asf/olingo-odata4/tree/b3f6f060
Diff: http://git-wip-us.apache.org/repos/asf/olingo-odata4/diff/b3f6f060
Branch: refs/heads/OLINGO-708_AsyncSupport
Commit: b3f6f06029f7f5b3582f99178be5ee827a310c8c
Parents: 0557f11
Author: Michael Bolz <mi...@sap.com>
Authored: Tue Jun 23 15:23:14 2015 +0200
Committer: Michael Bolz <mi...@sap.com>
Committed: Tue Jun 23 15:49:21 2015 +0200
----------------------------------------------------------------------
.../olingo/fit/AbstractBaseTestITCase.java | 7 +-
.../fit/tecsvc/client/AsyncSupportITCase.java | 24 ++--
.../olingo/server/api/processor/AsyncWrap.java | 2 +-
.../olingo/server/tecsvc/TechnicalServlet.java | 2 +-
.../tecsvc/TechnicalServletContextListener.java | 37 +++++
.../processor/TechnicalAsyncProcessor.java | 134 +++++++++++++++----
.../src/main/webapp/WEB-INF/web.xml | 4 +-
7 files changed, 167 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/b3f6f060/fit/src/test/java/org/apache/olingo/fit/AbstractBaseTestITCase.java
----------------------------------------------------------------------
diff --git a/fit/src/test/java/org/apache/olingo/fit/AbstractBaseTestITCase.java b/fit/src/test/java/org/apache/olingo/fit/AbstractBaseTestITCase.java
index 58dd2a3..b327eaf 100644
--- a/fit/src/test/java/org/apache/olingo/fit/AbstractBaseTestITCase.java
+++ b/fit/src/test/java/org/apache/olingo/fit/AbstractBaseTestITCase.java
@@ -39,6 +39,8 @@ import org.apache.olingo.commons.api.data.EntityCollection;
import org.apache.olingo.commons.api.format.ODataFormat;
import org.apache.olingo.fit.server.TomcatTestServer;
import org.apache.olingo.server.tecsvc.TechnicalServlet;
+import org.apache.olingo.server.tecsvc.processor.TechnicalAsyncProcessor;
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,11 +65,6 @@ public abstract class AbstractBaseTestITCase {
.start();
}
- // @AfterClass
- // public static void cleanUp() throws LifecycleException {
- // server.stop();
- // }
-
protected void debugEntity(final Entity entity, final String message) {
if (LOG.isDebugEnabled()) {
final StringWriter writer = new StringWriter();
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/b3f6f060/fit/src/test/java/org/apache/olingo/fit/tecsvc/client/AsyncSupportITCase.java
----------------------------------------------------------------------
diff --git a/fit/src/test/java/org/apache/olingo/fit/tecsvc/client/AsyncSupportITCase.java b/fit/src/test/java/org/apache/olingo/fit/tecsvc/client/AsyncSupportITCase.java
index 5ed4464..a116f19 100644
--- a/fit/src/test/java/org/apache/olingo/fit/tecsvc/client/AsyncSupportITCase.java
+++ b/fit/src/test/java/org/apache/olingo/fit/tecsvc/client/AsyncSupportITCase.java
@@ -30,6 +30,7 @@ import org.junit.Test;
import java.io.InputStream;
import java.net.URI;
+import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
@@ -54,11 +55,11 @@ public final class AsyncSupportITCase extends AbstractBaseTestITCase {
// first async request
InputStream rawResponse = client.getRetrieveRequestFactory()
.getEntityRequest(uri)
- .addCustomHeader("Prefer", "respond-async")
+ .addCustomHeader("Prefer", "respond-async, wait=1")
.rawExecute();
- String resp = IOUtils.toString(rawResponse);
- assertEquals("1", resp);
+ String respLocationFirst = IOUtils.toString(rawResponse);
+ assertEquals("http://localhost:9080/odata-server-tecsvc/odata.svc/ESAllPrim(32767)?async=1", respLocationFirst);
// assertEquals(3, response.getStatusCode());
// second async request
@@ -67,16 +68,23 @@ public final class AsyncSupportITCase extends AbstractBaseTestITCase {
.addCustomHeader("Prefer", "respond-async")
.rawExecute();
- resp = IOUtils.toString(rawResponse);
- assertEquals("2", resp);
+ String respLocationSecond = IOUtils.toString(rawResponse);
+ assertEquals("http://localhost:9080/odata-server-tecsvc/odata.svc/ESAllPrim(32767)?async=2", respLocationSecond);
// get result of first async request
- URI asyncUri = new URI(uri.toASCIIString() + "/?async=1");
+ URI respLocationFirstUri = new URI(respLocationFirst);
+ rawResponse = client.getRetrieveRequestFactory()
+ .getEntityRequest(respLocationFirstUri)
+ .rawExecute();
+
+ String respInProgressFirst = IOUtils.toString(rawResponse);
+ assertEquals("In progress for async location = " + respLocationFirst, respInProgressFirst);
+
+ TimeUnit.SECONDS.sleep(1);
final ODataRetrieveResponse<ClientEntity> firstAsyncResult = getClient().getRetrieveRequestFactory()
- .getEntityRequest(asyncUri).execute();
+ .getEntityRequest(respLocationFirstUri).execute();
assertEquals(32767, firstAsyncResult.getBody().getProperty("PropertyInt16").getPrimitiveValue().toValue());
assertEquals(200, firstAsyncResult.getStatusCode());
-
}
@Override
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/b3f6f060/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsyncWrap.java
----------------------------------------------------------------------
diff --git a/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsyncWrap.java b/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsyncWrap.java
index 7540d95..56b09c3 100644
--- a/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsyncWrap.java
+++ b/lib/server-api/src/main/java/org/apache/olingo/server/api/processor/AsyncWrap.java
@@ -29,7 +29,7 @@ public interface AsyncWrap {
ODataResponse getResult();
boolean isFinished();
- enum WrappedCall { READ_ENTITY };
+ enum WrappedCall { READ_ENTITY }
void forward(CallInfo info) throws ODataApplicationException, ODataLibraryException;
void forward() throws ODataApplicationException, ODataLibraryException;
Processor getWrappedProcessor();
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/b3f6f060/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServlet.java
----------------------------------------------------------------------
diff --git a/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServlet.java b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServlet.java
index 1faa567..d2aac9f 100644
--- a/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServlet.java
+++ b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServlet.java
@@ -85,7 +85,7 @@ public class TechnicalServlet extends HttpServlet {
handler.register(new TechnicalActionProcessor(dataProvider, serviceMetadata));
handler.register(new TechnicalBatchProcessor(dataProvider));
handler.register(new ETagSupport());
- handler.register(new TechnicalAsyncProcessor());
+ handler.register(TechnicalAsyncProcessor.getInstance());
handler.process(request, response);
} catch (final RuntimeException e) {
LOG.error("Server Error", e);
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/b3f6f060/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServletContextListener.java
----------------------------------------------------------------------
diff --git a/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServletContextListener.java b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServletContextListener.java
new file mode 100644
index 0000000..a09abff
--- /dev/null
+++ b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/TechnicalServletContextListener.java
@@ -0,0 +1,37 @@
+/*
+ * 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.tecsvc;
+
+import org.apache.olingo.server.tecsvc.processor.TechnicalAsyncProcessor;
+
+import javax.servlet.ServletContextEvent;
+import javax.servlet.ServletContextListener;
+
+/**
+ */
+public class TechnicalServletContextListener implements ServletContextListener {
+ @Override
+ public void contextInitialized(ServletContextEvent servletContextEvent) {
+ }
+
+ @Override
+ public void contextDestroyed(ServletContextEvent servletContextEvent) {
+ TechnicalAsyncProcessor.getInstance().shutdownThreadPool();
+ }
+}
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/b3f6f060/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
----------------------------------------------------------------------
diff --git a/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
index 040f5f4..e22a668 100644
--- a/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
+++ b/lib/server-tecsvc/src/main/java/org/apache/olingo/server/tecsvc/processor/TechnicalAsyncProcessor.java
@@ -33,31 +33,51 @@ import java.io.ByteArrayInputStream;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
public class TechnicalAsyncProcessor implements AsynchronousProcessor {
- private static Map<String, AsyncWrap> id2AsyncWrap = Collections.synchronizedMap(new HashMap<String, AsyncWrap>());
- private static AtomicInteger ai = new AtomicInteger();
+ private static final Map<String, AsyncRunner> LOCATION_2_ASYNC_RUNNER =
+ Collections.synchronizedMap(new HashMap<String, AsyncRunner>());
+ private static final ExecutorService ASYNC_REQUEST_EXECUTOR = Executors.newFixedThreadPool(10);
+ private static final AtomicInteger ai = new AtomicInteger();
+
+ private static final class AsyncProcessorHolder {
+ private static final TechnicalAsyncProcessor INSTANCE = new TechnicalAsyncProcessor();
+ }
+
+ public static TechnicalAsyncProcessor getInstance() {
+ return AsyncProcessorHolder.INSTANCE;
+ }
+
+ public void shutdownThreadPool() {
+ ASYNC_REQUEST_EXECUTOR.shutdown();
+ }
@Override
public boolean isStatusMonitorResource(ODataRequest request) throws ODataApplicationException, ODataLibraryException {
- return request.getRawQueryPath() != null && request.getRawQueryPath().contains("async");
+ return request.getRawRequestUri() != null && request.getRawRequestUri().contains("async");
+// return LOCATION_2_ASYNC_RUNNER.containsKey(getAsyncLocation(request));
}
@Override
public void processAsynchronous(ODataRequest request, ODataResponse response, AsyncWrap dispatchedProcessor)
throws ODataApplicationException, ODataLibraryException {
- String id = String.valueOf(ai.incrementAndGet());
- id2AsyncWrap.put(id, dispatchedProcessor);
-
- // thread?
- dispatchedProcessor.forward();
+ // use executor thread pool
+ String location = createNewAsyncLocation(request);
+ AsyncRunner run = new AsyncRunner(dispatchedProcessor);
+ LOCATION_2_ASYNC_RUNNER.put(location, run);
+ ASYNC_REQUEST_EXECUTOR.execute(run);
//
response.setStatusCode(HttpStatusCode.ACCEPTED.getStatusCode());
- response.setHeader(HttpHeader.LOCATION, request.getRawRequestUri() + "/?async=" + id);
- response.setContent(new ByteArrayInputStream(id.getBytes()));
+ response.setHeader(HttpHeader.LOCATION, location);
+ response.setContent(new ByteArrayInputStream(location.getBytes()));
}
@Override
@@ -74,42 +94,102 @@ public class TechnicalAsyncProcessor implements AsynchronousProcessor {
@Override
public void result(ODataRequest request, ODataResponse response) {
- String id = getAsyncId(request);
- AsyncWrap wrap = id2AsyncWrap.get(id);
+ String location = getAsyncLocation(request);
+ AsyncRunner runner = LOCATION_2_ASYNC_RUNNER.get(location);
- if(wrap == null) {
+ if(runner == null) {
response.setStatusCode(HttpStatusCode.NOT_FOUND.getStatusCode());
} else {
- if(wrap.isFinished()) {
- ODataResponse wrapResult = wrap.getResult();
+ if(runner.isFinished()) {
+ ODataResponse wrapResult = runner.getDispatched().getResult();
response.setContent(wrapResult.getContent());
response.setStatusCode(wrapResult.getStatusCode());
for (Map.Entry<String,String> header : wrapResult.getHeaders().entrySet()) {
response.setHeader(header.getKey(), header.getValue());
}
- id2AsyncWrap.remove(id);
+ LOCATION_2_ASYNC_RUNNER.remove(location);
} else {
response.setStatusCode(HttpStatusCode.ACCEPTED.getStatusCode());
- response.setHeader(HttpHeader.LOCATION, request.getRawBaseUri() + "async=" + id);
- String content = "In progress for async id = " + id;
+ response.setHeader(HttpHeader.LOCATION, location);
+ String content = "In progress for async location = " + location;
response.setContent(new ByteArrayInputStream(content.getBytes()));
}
}
}
+ @Override
+ public void init(OData odata, ServiceMetadata serviceMetadata) {
+ }
- private String getAsyncId(ODataRequest request) {
- return request.getRawQueryPath().substring(6);
+ private String createNewAsyncLocation(ODataRequest request) {
+ return request.getRawBaseUri() + request.getRawODataPath() + "?async=" + ai.incrementAndGet();
}
-// @Override
-// public void processAsynchronous(AsyncWrap dispatchedProcessor) throws ODataApplicationException,
-// ODataLibraryException {
-// dispatchedProcessor.forward();
-// }
+ private String getAsyncLocation(ODataRequest request) {
+ return request.getRawBaseUri() + request.getRawODataPath() + "?" + getAsyncQueryPart(request);
+ }
+
+ private String getAsyncQueryPart(ODataRequest request) {
+ String rawQueryPath = request.getRawQueryPath();
+ if(rawQueryPath != null) {
+ Matcher m = Pattern.compile("(async=\\d*)").matcher(rawQueryPath);
+ if(m.find()) {
+ return m.group();
+ }
+ }
+ return "";
+ }
- @Override
- public void init(OData odata, ServiceMetadata serviceMetadata) {
+ private static class AsyncRunner implements Runnable {
+ private final AsyncWrap wrap;
+ private int defaultSleepTimeInSeconds = 0;
+ private Exception exception;
+ boolean finished = false;
+
+ public AsyncRunner(AsyncWrap wrap) {
+ this(wrap, 0);
+ }
+
+ public AsyncRunner(AsyncWrap wrap, int defaultSleepTimeInSeconds) {
+ this.wrap = wrap;
+ if(defaultSleepTimeInSeconds > 0) {
+ this.defaultSleepTimeInSeconds = defaultSleepTimeInSeconds;
+ }
+ }
+
+ @Override
+ public void run() {
+ try {
+ int sleep = getSleepTime(wrap);
+ TimeUnit.SECONDS.sleep(sleep);
+ wrap.forward();
+ } catch (Exception e) {
+ exception = e;
+ }
+ finished = true;
+ }
+
+ private int getSleepTime(AsyncWrap wrap) {
+ String preferHeader = wrap.getCallInfo().request.getHeader(HttpHeader.PREFER);
+ Matcher matcher = Pattern.compile("(wait=)(\\d*)").matcher(preferHeader);
+ if(matcher.find()) {
+ String waitTimeAsString = matcher.group(2);
+ return Integer.parseInt(waitTimeAsString);
+ }
+ return defaultSleepTimeInSeconds;
+ }
+
+ public Exception getException() {
+ return exception;
+ }
+
+ public boolean isFinished() {
+ return finished && wrap.isFinished();
+ }
+
+ public AsyncWrap getDispatched() {
+ return wrap;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/olingo-odata4/blob/b3f6f060/lib/server-tecsvc/src/main/webapp/WEB-INF/web.xml
----------------------------------------------------------------------
diff --git a/lib/server-tecsvc/src/main/webapp/WEB-INF/web.xml b/lib/server-tecsvc/src/main/webapp/WEB-INF/web.xml
index 3c68d16..3c587c0 100644
--- a/lib/server-tecsvc/src/main/webapp/WEB-INF/web.xml
+++ b/lib/server-tecsvc/src/main/webapp/WEB-INF/web.xml
@@ -38,5 +38,7 @@
<servlet-name>ODataServlet</servlet-name>
<url-pattern>/odata.svc/*</url-pattern>
</servlet-mapping>
-
+ <listener>
+ <listener-class>org.apache.olingo.server.tecsvc.TechnicalServletContextListener</listener-class>
+ </listener>
</web-app>