I have a Spring Boot application where I created a POST method that sends data in a streaming fashion to the caller. Code below:
@RequestMapping(value = "/mapmatchstreaming", method = RequestMethod.POST)
public ResponseEntity<StreamingResponseBody> handleRequest(@RequestParam(value = "data", required = true) String data, @RequestParam(value = "mnr", required = true) Boolean mnr) {
logger.info("/mapmatchstreaming endpoint");
try {
Semaphore semaphore = new Semaphore(1);
ObjectMapper mapper = new ObjectMapper();
StreamingResponseBody responseBody = new StreamingResponseBody() {
@Override
public void writeTo (OutputStream outputStream) throws IOException {
// For each map
DataReader dataReader = new DataReader(data, "2020.06.011");
for(String mapRoot: dataReader.getMapsFolders()) {
dataReader = new DataReader(data, "2020.06.011");
DistributedMapMatcherStreaming distributedMapMatcher = new DistributedMapMatcherStreaming(dataReader.getTraces(), mapRoot, dataReader.getBoundingBox());
distributedMapMatcher.mapMatchBatch(new DistributedMapMatcherResult() {
@Override
public void onCorrectlyMapMatched(MapMatchedTrajectory mapMatchedTrajectory) {
try {
semaphore.acquire();
outputStream.write(mapper.writeValueAsString(mapMatchedTrajectory).getBytes());
outputStream.flush();
}
catch (Exception e) {
e.printStackTrace();
logger.error(String.format("Writing to output stream error: %s", e.getMessage()));
} finally{
semaphore.release();
}
}
});
}
}
};
return new ResponseEntity<StreamingResponseBody>(responseBody, HttpStatus.OK);
}
catch (Exception e) {
logger.error(String.format("Map-matching result ERROR: %s", ExceptionUtils.getStackTrace(e)));
return new ResponseEntity<StreamingResponseBody>(HttpStatus.BAD_REQUEST);
}
}
It works nicely, but the problem is that if multiple calls arrive to this method, all of them are run in parallel even if I have set server.tomcat.threads.max=1
. In the non-streaming version, every next call waits for the current one to complete.
Is it possible to have blocking streaming calls in Spring? Thanks.
EDIT: I temporarily solved by using a global semaphore with only 1 permit, but I don't think this is the ideal solution.