Repository: cxf Updated Branches: refs/heads/master 1bcb4ec0d -> 95bfefb65
[CXF-5893] Websocket transport sporadically fails with corrupted data Project: http://git-wip-us.apache.org/repos/asf/cxf/repo Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/95bfefb6 Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/95bfefb6 Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/95bfefb6 Branch: refs/heads/master Commit: 95bfefb653417b8b23ad481ee3f61139ef0eb5ed Parents: 1bcb4ec Author: Akitoshi Yoshida <[email protected]> Authored: Fri Jul 18 23:57:40 2014 +0200 Committer: Akitoshi Yoshida <[email protected]> Committed: Sat Jul 19 00:01:46 2014 +0200 ---------------------------------------------------------------------- .../websocket/atmosphere/AtmosphereWebSocketHandler.java | 9 +++++++-- .../cxf/transport/websocket/jetty/JettyWebSocket.java | 7 ++++++- 2 files changed, 13 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cxf/blob/95bfefb6/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/atmosphere/AtmosphereWebSocketHandler.java ---------------------------------------------------------------------- diff --git a/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/atmosphere/AtmosphereWebSocketHandler.java b/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/atmosphere/AtmosphereWebSocketHandler.java index 6a5550d..ee4e358 100644 --- a/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/atmosphere/AtmosphereWebSocketHandler.java +++ b/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/atmosphere/AtmosphereWebSocketHandler.java @@ -86,18 +86,23 @@ public class AtmosphereWebSocketHandler implements WebSocketProtocol { } catch (UnsupportedEncodingException e) { // will not happen } - return onMessage(webSocket, bdata, 0, bdata.length); + return invokeService(webSocket, new ByteArrayInputStream(bdata, 0, bdata.length)); } /** {@inheritDoc}*/ @Override public List<AtmosphereRequest> onMessage(WebSocket webSocket, byte[] data, int offset, int length) { - return invokeService(webSocket, new ByteArrayInputStream(data, offset, length)); + final byte[] safedata = new byte[length]; + System.arraycopy(data, offset, safedata, 0, length); + return invokeService(webSocket, new ByteArrayInputStream(safedata, 0, safedata.length)); } protected List<AtmosphereRequest> invokeService(final WebSocket webSocket, final InputStream stream) { LOG.info("invokeService(WebSocket, InputStream)"); // invoke the service directly as onMessage is synchronously blocked (in jetty) + // make sure the byte array passed to this method is immutable, as the websocket framework + // may corrupt the byte array after this method is returned (i.e., before the data is returned in + // the executor's thread. destination.getExecutor().execute(new Runnable() { @Override public void run() { http://git-wip-us.apache.org/repos/asf/cxf/blob/95bfefb6/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/jetty/JettyWebSocket.java ---------------------------------------------------------------------- diff --git a/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/jetty/JettyWebSocket.java b/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/jetty/JettyWebSocket.java index 3f2a79f..8d0f596 100644 --- a/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/jetty/JettyWebSocket.java +++ b/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/jetty/JettyWebSocket.java @@ -99,11 +99,16 @@ class JettyWebSocket implements WebSocket.OnBinaryMessage, WebSocket.OnTextMessa if (LOG.isLoggable(Level.INFO)) { LOG.log(Level.INFO, "onMessage({0}, {1}, {2})", new Object[]{data, offset, length}); } - invokeService(data, offset, length); + final byte[] safedata = new byte[length]; + System.arraycopy(data, offset, safedata, 0, length); + invokeService(safedata, 0, safedata.length); } private void invokeService(final byte[] data, final int offset, final int length) { // invoke the service asynchronously as the jetty websocket's onMessage is synchronously blocked + // make sure the byte array passed to this method is immutable, as the websocket framework + // may corrupt the byte array after this method is returned (i.e., before the data is returned in + // the executor's thread. manager.getExecutor().execute(new Runnable() { @Override public void run() {
