Skip to content

Result streaming #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Result streaming #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
602842c
Adding option to stream execute query results.
truvenganong Jan 6, 2015
d10fc24
Check for null when closing query results.
truvenganong Feb 9, 2015
da0ab9a
Fixed parseInt typo.
truvenganong Jun 22, 2015
09d245b
Fixing compilation isssues.
truvenganong Jun 23, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 98 additions & 17 deletions src/main/java/mondrian/xmla/XmlaHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,8 @@ public class XmlaHandler {

final ConnectionFactory connectionFactory;
private final String prefix;
private Map<XmlaRequest, QueryResult> queryResultLookup;
private boolean streamExecuteResults;

/**
* Returns a new OlapConnection opened with the credentials specified in the
Expand Down Expand Up @@ -610,6 +612,29 @@ public XmlaHandler(ConnectionFactory connectionFactory, String prefix)
assert prefix != null;
this.connectionFactory = connectionFactory;
this.prefix = prefix;
this.streamExecuteResults = false;
}

public void setStreamExecuteResults(final int maxResultsToDefer) {
if (!this.streamExecuteResults) {
this.streamExecuteResults = true;
this.queryResultLookup = new LinkedHashMap<XmlaRequest, QueryResult>() {
@Override
protected boolean removeEldestEntry(
java.util.Map.Entry<XmlaRequest, QueryResult> eldest)
{
if (size() > maxResultsToDefer) {
try {
eldest.getValue().close();
} catch (Exception e) {
//ignore
}
return true;
}
return false;
}
};
}
}

/**
Expand Down Expand Up @@ -693,22 +718,6 @@ private void execute(
XmlaResponse response)
throws XmlaException
{
final Map<String, String> properties = request.getProperties();

// Default responseMimeType is SOAP.
Enumeration.ResponseMimeType responseMimeType =
getResponseMimeType(request);

// Default value is SchemaData, or Data for JSON responses.
final String contentName =
properties.get(PropertyDefinition.Content.name());
Content content = Util.lookup(
Content.class,
contentName,
responseMimeType == Enumeration.ResponseMimeType.JSON
? Content.Data
: Content.DEFAULT);

// Handle execute
QueryResult result = null;
try {
Expand All @@ -717,8 +726,80 @@ private void execute(
} else {
result = executeQuery(request);
}
} catch (XmlaException xe) {
if (result != null) {
try {
result.close();
} catch (SQLException e) {
//ignore
}
}
throw xe;
} catch (RuntimeException re) {
if (result != null) {
try {
result.close();
} catch (SQLException e) {
//ignore
}
}
throw re;
}
if (streamExecuteResults) {
synchronized(queryResultLookup) {
queryResultLookup.put(request, result);
}
} else {
writeQueryResult(result, request, response.getWriter());
}
}

public void writeQueryResult(XmlaRequest request, SaxWriter writer) {
QueryResult result;
synchronized(queryResultLookup) {
result = queryResultLookup.remove(request);
}
if (result != null) {
writeQueryResult(result, request, writer);
}
}

public void closeQueryResult(XmlaRequest request) {
QueryResult result = null;
if (queryResultLookup != null) {
synchronized(queryResultLookup) {
result = queryResultLookup.remove(request);
}
}
if (result != null) {
try {
result.close();
} catch (SQLException e) {
//ignore
}
}
}

private void writeQueryResult(QueryResult result,
XmlaRequest request, SaxWriter writer) {
try {
final Map<String, String> properties = request.getProperties();

// Default responseMimeType is SOAP.
Enumeration.ResponseMimeType responseMimeType =
getResponseMimeType(request);

// Default value is SchemaData, or Data for JSON responses.
final String contentName =
properties.get(PropertyDefinition.Content.name());
Content content = Util.lookup(
Content.class,
contentName,
responseMimeType == Enumeration.ResponseMimeType.JSON
? Content.Data
: Content.DEFAULT);


SaxWriter writer = response.getWriter();
writer.startDocument();

writer.startElement(
Expand Down
30 changes: 15 additions & 15 deletions src/main/java/mondrian/xmla/XmlaServlet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,7 @@ protected void doPost(
Phase phase = Phase.VALIDATE_HTTP_HEAD;
Enumeration.ResponseMimeType mimeType =
Enumeration.ResponseMimeType.SOAP;
Map<String, Object> context = new HashMap<String, Object>();

try {
if (charEncoding != null) {
Expand All @@ -172,8 +173,6 @@ protected void doPost(

response.setContentType(mimeType.getMimeType());

Map<String, Object> context = new HashMap<String, Object>();

try {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Invoking validate http header callbacks");
Expand All @@ -192,7 +191,7 @@ protected void doPost(
"Errors when invoking callbacks validateHttpHeader", xex);
handleFault(response, responseSoapParts, phase, xex);
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
} catch (Exception ex) {
LOGGER.error(
Expand All @@ -206,7 +205,7 @@ protected void doPost(
CHH_FAULT_FS,
ex));
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
}

Expand Down Expand Up @@ -249,7 +248,7 @@ protected void doPost(
LOGGER.error("Unable to unmarshall SOAP message", xex);
handleFault(response, responseSoapParts, phase, xex);
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
}

Expand All @@ -270,7 +269,7 @@ protected void doPost(
LOGGER.error("Errors when handling XML/A message", xex);
handleFault(response, responseSoapParts, phase, xex);
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
}

Expand All @@ -289,7 +288,7 @@ protected void doPost(
LOGGER.error("Errors when invoking callbacks preaction", xex);
handleFault(response, responseSoapParts, phase, xex);
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
} catch (Exception ex) {
LOGGER.error("Errors when invoking callbacks preaction", ex);
Expand All @@ -302,7 +301,7 @@ protected void doPost(
CPREA_FAULT_FS,
ex));
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
}

Expand All @@ -323,7 +322,7 @@ protected void doPost(
LOGGER.error("Errors when handling XML/A message", xex);
handleFault(response, responseSoapParts, phase, xex);
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
}

Expand All @@ -346,7 +345,7 @@ protected void doPost(
LOGGER.error("Errors when invoking callbacks postaction", xex);
handleFault(response, responseSoapParts, phase, xex);
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
} catch (Exception ex) {
LOGGER.error("Errors when invoking callbacks postaction", ex);
Expand All @@ -360,25 +359,25 @@ protected void doPost(
CPOSTA_FAULT_FS,
ex));
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
return;
}

phase = Phase.SEND_RESPONSE;

try {
response.setStatus(HttpServletResponse.SC_OK);
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
} catch (XmlaException xex) {
LOGGER.error("Errors when handling XML/A message", xex);
handleFault(response, responseSoapParts, phase, xex);
phase = Phase.SEND_ERROR;
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
}
} catch (Throwable t) {
LOGGER.error("Unknown Error when handling XML/A message", t);
handleFault(response, responseSoapParts, phase, t);
marshallSoapMessage(response, responseSoapParts, mimeType);
marshallSoapMessage(response, responseSoapParts, mimeType, context);
}
}

Expand Down Expand Up @@ -416,7 +415,8 @@ protected abstract void handleSoapBody(
protected abstract void marshallSoapMessage(
HttpServletResponse response,
byte[][] responseSoapParts,
Enumeration.ResponseMimeType responseMimeType)
Enumeration.ResponseMimeType responseMimeType,
Map<String, Object> context)
throws XmlaException;

/**
Expand Down
Loading