This is an automated email from the ASF dual-hosted git repository.
shuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new ffd5a48 UNOMI-328-handle Elasticsearch exceptions (#154)
ffd5a48 is described below
commit ffd5a488111a950a581c89c55313a4c12eabb2fe
Author: nlevitsky <[email protected]>
AuthorDate: Thu Apr 30 16:46:23 2020 +0300
UNOMI-328-handle Elasticsearch exceptions (#154)
* improve(ContextServlet) - change parameters from
ServletRequest/ServletResponse to HttpServletRequest/HttpServletResponse
* feat(ï¿¿EventServiceImpl) - add "ERROR" as a possible event-creation
outcome after inserting a new event
* feat(ServletCommon) - trace and return the number of successfully
processed events
* feat(ContextServlet) - return HTTP status 500 if an error occurred in
adding the events
* feat(EventsCollectorServlet) - return HTTP status 500 if an error
occurred in adding the events
Co-authored-by: nlevitsky <Pitzek1305>
---
.../java/org/apache/unomi/api/ContextResponse.java | 11 +++++
.../apache/unomi/api/services/EventService.java | 8 +++-
.../services/impl/events/EventServiceImpl.java | 52 ++++++++++++----------
.../main/java/org/apache/unomi/web/Changes.java | 10 +++++
.../java/org/apache/unomi/web/ContextServlet.java | 21 +++++----
.../apache/unomi/web/EventsCollectorServlet.java | 5 +++
.../java/org/apache/unomi/web/ServletCommon.java | 11 ++++-
7 files changed, 82 insertions(+), 36 deletions(-)
diff --git a/api/src/main/java/org/apache/unomi/api/ContextResponse.java
b/api/src/main/java/org/apache/unomi/api/ContextResponse.java
index 0969cb4..632d1b7 100644
--- a/api/src/main/java/org/apache/unomi/api/ContextResponse.java
+++ b/api/src/main/java/org/apache/unomi/api/ContextResponse.java
@@ -48,6 +48,8 @@ public class ContextResponse implements Serializable {
private Map<String, Boolean> filteringResults;
+ private int processedEvents;
+
private Map<String, List<String>> personalizations;
private Set<Condition> trackedConditions;
@@ -169,6 +171,15 @@ public class ContextResponse implements Serializable {
this.filteringResults = filteringResults;
}
+
+ public int getProcessedEvents() {
+ return processedEvents;
+ }
+
+ public void setProcessedEvents(int processedEvents) {
+ this.processedEvents = processedEvents;
+ }
+
public Map<String, List<String>> getPersonalizations() {
return personalizations;
}
diff --git a/api/src/main/java/org/apache/unomi/api/services/EventService.java
b/api/src/main/java/org/apache/unomi/api/services/EventService.java
index 4dc53f7..d27d60c 100644
--- a/api/src/main/java/org/apache/unomi/api/services/EventService.java
+++ b/api/src/main/java/org/apache/unomi/api/services/EventService.java
@@ -38,13 +38,17 @@ public interface EventService {
*/
int NO_CHANGE = 0;
/**
+ * An error occurred while processing the event.
+ */
+ int ERROR = 1;
+ /**
* The associated session was updated following an event being handled.
*/
- int SESSION_UPDATED = 1;
+ int SESSION_UPDATED = 2;
/**
* The associated profile was updated following an event being handled.
*/
- int PROFILE_UPDATED = 2;
+ int PROFILE_UPDATED = 4;
/**
* Propagates the specified event in the context server, notifying
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/events/EventServiceImpl.java
b/services/src/main/java/org/apache/unomi/services/impl/events/EventServiceImpl.java
index b1846dc..a6e6bbc 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/events/EventServiceImpl.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/events/EventServiceImpl.java
@@ -150,38 +150,44 @@ public class EventServiceImpl implements EventService {
return NO_CHANGE;
}
+ boolean saveSucceeded = true;
if (event.isPersistent()) {
- persistenceService.save(event);
+ saveSucceeded = persistenceService.save(event);
}
- int changes = NO_CHANGE;
+ int changes;
- final Session session = event.getSession();
- if (event.isPersistent() && session != null) {
- session.setLastEventDate(event.getTimeStamp());
- }
+ if (saveSucceeded) {
+ changes = NO_CHANGE;
+ final Session session = event.getSession();
+ if (event.isPersistent() && session != null) {
+ session.setLastEventDate(event.getTimeStamp());
+ }
- if (event.getProfile() != null) {
- for (EventListenerService eventListenerService : eventListeners) {
- if (eventListenerService.canHandle(event)) {
- changes |= eventListenerService.onEvent(event);
+ if (event.getProfile() != null) {
+ for (EventListenerService eventListenerService :
eventListeners) {
+ if (eventListenerService.canHandle(event)) {
+ changes |= eventListenerService.onEvent(event);
+ }
+ }
+ // At the end of the processing event execute the post
executor actions
+ for (ActionPostExecutor actionPostExecutor :
event.getActionPostExecutors()) {
+ changes |= actionPostExecutor.execute() ? changes :
NO_CHANGE;
}
- }
- // At the end of the processing event execute the post executor
actions
- for (ActionPostExecutor actionPostExecutor :
event.getActionPostExecutors()) {
- changes |= actionPostExecutor.execute() ? changes : NO_CHANGE;
- }
- if ((changes & PROFILE_UPDATED) == PROFILE_UPDATED) {
- Event profileUpdated = new Event("profileUpdated", session,
event.getProfile(), event.getScope(), event.getSource(), event.getProfile(),
event.getTimeStamp());
- profileUpdated.setPersistent(false);
- profileUpdated.getAttributes().putAll(event.getAttributes());
- changes |= send(profileUpdated, depth + 1);
- if (session != null && session.getProfileId() != null) {
- changes |= SESSION_UPDATED;
- session.setProfile(event.getProfile());
+ if ((changes & PROFILE_UPDATED) == PROFILE_UPDATED) {
+ Event profileUpdated = new Event("profileUpdated",
session, event.getProfile(), event.getScope(), event.getSource(),
event.getProfile(), event.getTimeStamp());
+ profileUpdated.setPersistent(false);
+
profileUpdated.getAttributes().putAll(event.getAttributes());
+ changes |= send(profileUpdated, depth + 1);
+ if (session != null && session.getProfileId() != null) {
+ changes |= SESSION_UPDATED;
+ session.setProfile(event.getProfile());
+ }
}
}
+ } else {
+ changes = ERROR;
}
return changes;
}
diff --git a/wab/src/main/java/org/apache/unomi/web/Changes.java
b/wab/src/main/java/org/apache/unomi/web/Changes.java
index 280bbb4..1f1043c 100644
--- a/wab/src/main/java/org/apache/unomi/web/Changes.java
+++ b/wab/src/main/java/org/apache/unomi/web/Changes.java
@@ -26,10 +26,16 @@ import org.apache.unomi.api.Profile;
*/
public class Changes {
private int changeType;
+ private int processedItems;
private Profile profile;
public Changes(int changeType, Profile profile) {
+ this(changeType,0,profile);
+ }
+
+ public Changes(int changeType, int processedItems, Profile profile) {
this.changeType = changeType;
+ this.processedItems = processedItems;
this.profile = profile;
}
@@ -37,6 +43,10 @@ public class Changes {
return changeType;
}
+ public int getProcessedItems() {
+ return processedItems;
+ }
+
public Profile getProfile() {
return profile;
}
diff --git a/wab/src/main/java/org/apache/unomi/web/ContextServlet.java
b/wab/src/main/java/org/apache/unomi/web/ContextServlet.java
index 961276b..693c23e 100644
--- a/wab/src/main/java/org/apache/unomi/web/ContextServlet.java
+++ b/wab/src/main/java/org/apache/unomi/web/ContextServlet.java
@@ -69,19 +69,17 @@ public class ContextServlet extends HttpServlet {
}
@Override
- public void service(ServletRequest request, ServletResponse response)
throws IOException {
+ public void service(HttpServletRequest request, HttpServletResponse
response) throws IOException {
final Date timestamp = new Date();
if (request.getParameter("timestamp") != null) {
timestamp.setTime(Long.parseLong(request.getParameter("timestamp")));
}
- HttpServletRequest httpServletRequest = (HttpServletRequest) request;
-
// set up CORS headers as soon as possible so that errors are not
misconstrued on the client for CORS errors
- HttpUtils.setupCORSHeaders(httpServletRequest, response);
+ HttpUtils.setupCORSHeaders(request, response);
// Handle OPTIONS request
- String httpMethod = httpServletRequest.getMethod();
+ String httpMethod = request.getMethod();
if ("options".equals(httpMethod.toLowerCase())) {
response.flushBuffer();
if (logger.isDebugEnabled()) {
@@ -109,7 +107,7 @@ public class ContextServlet extends HttpServlet {
ContextRequest contextRequest = null;
String scope = null;
String sessionId = null;
- String stringPayload = HttpUtils.getPayload(httpServletRequest);
+ String stringPayload = HttpUtils.getPayload(request);
if (stringPayload != null) {
ObjectMapper mapper = CustomObjectMapper.getObjectMapper();
JsonFactory factory = mapper.getFactory();
@@ -131,13 +129,13 @@ public class ContextServlet extends HttpServlet {
}
// Get profile id from the cookie
- String cookieProfileId =
ServletCommon.getProfileIdCookieValue(httpServletRequest, profileIdCookieName);
+ String cookieProfileId =
ServletCommon.getProfileIdCookieValue(request, profileIdCookieName);
if (cookieProfileId == null && sessionId == null && personaId == null)
{
((HttpServletResponse)response).sendError(HttpServletResponse.SC_BAD_REQUEST,
"Check logs for more details");
logger.error("Couldn't find cookieProfileId, sessionId or
personaId in incoming request! Stopped processing request. See debug level for
more information");
if (logger.isDebugEnabled()) {
- logger.debug("Request dump: {}",
HttpUtils.dumpRequestInfo(httpServletRequest));
+ logger.debug("Request dump: {}",
HttpUtils.dumpRequestInfo(request));
}
return;
}
@@ -270,7 +268,11 @@ public class ContextServlet extends HttpServlet {
contextResponse.setSessionId(session.getItemId());
}
- String extension =
httpServletRequest.getRequestURI().substring(httpServletRequest.getRequestURI().lastIndexOf(".")
+ 1);
+ if ((changes & EventService.ERROR) == EventService.ERROR) {
+ response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+ }
+
+ String extension =
request.getRequestURI().substring(request.getRequestURI().lastIndexOf(".") + 1);
boolean noScript = "json".equals(extension);
String contextAsJSONString =
CustomObjectMapper.getObjectMapper().writeValueAsString(contextResponse);
Writer responseWriter;
@@ -321,6 +323,7 @@ public class ContextServlet extends HttpServlet {
ServletRequest request, ServletResponse
response, Date timestamp) {
Changes changes =
ServletCommon.handleEvents(contextRequest.getEvents(), session, profile,
request, response, timestamp,
privacyService, eventService);
+ data.setProcessedEvents(changes.getProcessedItems());
profile = changes.getProfile();
diff --git a/wab/src/main/java/org/apache/unomi/web/EventsCollectorServlet.java
b/wab/src/main/java/org/apache/unomi/web/EventsCollectorServlet.java
index 2f27836..2c34659 100644
--- a/wab/src/main/java/org/apache/unomi/web/EventsCollectorServlet.java
+++ b/wab/src/main/java/org/apache/unomi/web/EventsCollectorServlet.java
@@ -187,6 +187,11 @@ public class EventsCollectorServlet extends HttpServlet {
profileService.saveSession(session);
}
}
+ if ((changes & EventService.ERROR) == EventService.ERROR) {
+ String errorMessage = "Error processing events. Total number of
processed events: " + changesObject.getProcessedItems() + "/" +
eventsCollectorRequest.getEvents().size();
+ response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
errorMessage);
+ return;
+ }
response.setContentType("application/json");
PrintWriter responseWriter = response.getWriter();
diff --git a/wab/src/main/java/org/apache/unomi/web/ServletCommon.java
b/wab/src/main/java/org/apache/unomi/web/ServletCommon.java
index 87917e1..597295f 100644
--- a/wab/src/main/java/org/apache/unomi/web/ServletCommon.java
+++ b/wab/src/main/java/org/apache/unomi/web/ServletCommon.java
@@ -64,8 +64,10 @@ public class ServletCommon {
int changes = EventService.NO_CHANGE;
// execute provided events if any
+ int processedEventsCnt = 0;
if (events != null && !(profile instanceof Persona)) {
for (Event event : events) {
+ processedEventsCnt++;
if (event.getEventType() != null) {
Event eventToSend = new Event(event.getEventType(),
session, profile, event.getScope(), event.getSource(),
event.getTarget(), event.getProperties(),
timestamp, event.isPersistent());
@@ -94,10 +96,15 @@ public class ServletCommon {
if ((changes & EventService.PROFILE_UPDATED) ==
EventService.PROFILE_UPDATED) {
profile = eventToSend.getProfile();
}
+ if ((changes & EventService.ERROR) == EventService.ERROR) {
+ //Don't count the event that failed
+ processedEventsCnt--;
+ logger.error("Error processing events. Total number of
processed events: {}/{}", processedEventsCnt,events.size());
+ break;
+ }
}
}
}
-
- return new Changes(changes, profile);
+ return new Changes(changes, processedEventsCnt, profile);
}
}
