Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

JSR-356 cleaning up MessageInputStream and MessageReader

This commit is contained in:
Joakim Erdfelt 2013-06-05 10:34:38 -07:00
parent b01e6432a5
commit ffceb642c5
7 changed files with 263 additions and 499 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/WebSocketSession.java
View File

@ -132,6 +132,12 @@ public class WebSocketSession extends ContainerLifeCycle implements Session, Inc
notifyClose(StatusCode.NO_CLOSE,"Harsh disconnect");
}
public void dispatch(Runnable runnable)
{
// TODO Auto-generated method stub
runnable.run();
}
@Override
public void dump(Appendable out, String indent) throws IOException
{

25
jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/events/JettyAnnotatedEventDriver.java
View File

@ -65,6 +65,11 @@ public class JettyAnnotatedEventDriver extends AbstractEventDriver
}
}
private void dispatch(Runnable runnable)
{
session.dispatch(runnable);
}
@Override
public void onBinaryFrame(ByteBuffer buffer, boolean fin) throws IOException
{
@ -78,7 +83,15 @@ public class JettyAnnotatedEventDriver extends AbstractEventDriver
{
if (events.onBinary.isStreaming())
{
activeMessage = new MessageInputStream(this);
activeMessage = new MessageInputStream(session.getConnection());
dispatch(new Runnable()
{
@Override
public void run()
{
events.onBinary.call(websocket,session,activeMessage);
}
});
}
else
{
@ -171,7 +184,15 @@ public class JettyAnnotatedEventDriver extends AbstractEventDriver
{
if (events.onText.isStreaming())
{
activeMessage = new MessageReader(this);
activeMessage = new MessageReader(new MessageInputStream(session.getConnection()));
dispatch(new Runnable()
{
@Override
public void run()
{
events.onText.call(websocket,session,activeMessage);
}
});
}
else
{

130
jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageInputStream.java
View File

@ -21,97 +21,135 @@ package org.eclipse.jetty.websocket.common.message;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicBoolean;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.websocket.common.events.EventDriver;
import org.eclipse.jetty.websocket.common.LogicalConnection;
/**
* Support class for reading binary message data as an InputStream.
* Support class for reading a (single) WebSocket BINARY message via a InputStream.
* <p>
* An InputStream that can access a queue of ByteBuffer payloads, along with expected InputStream blocking behavior.
*/
public class MessageInputStream extends InputStream implements MessageAppender
{
/**
* Threshold (of bytes) to perform compaction at
* Used for controlling read suspend/resume behavior if the queue is full, but the read operations haven't caught up yet.
*/
private static final int COMPACT_THRESHOLD = 5;
private final EventDriver driver;
private final ByteBuffer buf;
private int size;
private boolean finished;
private boolean needsNotification;
private int readPosition;
@SuppressWarnings("unused")
private final LogicalConnection connection;
private final BlockingDeque<ByteBuffer> buffers = new LinkedBlockingDeque<>();
private AtomicBoolean closed = new AtomicBoolean(false);
// EOB / End of Buffers
private AtomicBoolean buffersExhausted = new AtomicBoolean(false);
private ByteBuffer activeBuffer = null;
public MessageInputStream(EventDriver driver)
public MessageInputStream(LogicalConnection connection)
{
this.driver = driver;
this.buf = ByteBuffer.allocate(driver.getPolicy().getMaxBinaryMessageBufferSize());
BufferUtil.clearToFill(this.buf);
size = 0;
readPosition = this.buf.position();
finished = false;
needsNotification = true;
this.connection = connection;
}
@Override
public void appendMessage(ByteBuffer payload, boolean isLast) throws IOException
{
if (finished)
if (buffersExhausted.get())
{
throw new IOException("Cannot append to finished buffer");
// This indicates a programming mistake/error and must be bug fixed
throw new RuntimeException("Last frame already received");
}
if (payload == null)
// if closed, we should just toss incoming payloads into the bit bucket.
if (closed.get())
{
// empty payload is valid
return;
}
driver.getPolicy().assertValidBinaryMessageSize(size + payload.remaining());
size += payload.remaining();
synchronized (buf)
// Put the payload into the queue
try
{
// TODO: grow buffer till max binary message size?
// TODO: compact this buffer to fit incoming buffer?
// TODO: tell connection to suspend if buffer too full?
BufferUtil.put(payload,buf);
buffers.put(payload);
if (isLast)
{
buffersExhausted.set(true);
}
}
if (needsNotification)
catch (InterruptedException e)
{
needsNotification = true;
this.driver.onInputStream(this);
throw new IOException(e);
}
}
@Override
public void close() throws IOException
{
finished = true;
closed.set(true);
super.close();
}
@Override
public synchronized void mark(int readlimit)
{
/* do nothing */
}
@Override
public boolean markSupported()
{
return false;
}
@Override
public void messageComplete()
{
finished = true;
buffersExhausted.set(true);
// toss an empty ByteBuffer into queue to let it drain
try
{
buffers.put(ByteBuffer.wrap(new byte[0]));
}
catch (InterruptedException ignore)
{
/* ignore */
}
}
@Override
public int read() throws IOException
{
synchronized (buf)
try
{
byte b = buf.get(readPosition);
readPosition++;
if (readPosition <= (buf.limit() - COMPACT_THRESHOLD))
if (closed.get())
{
int curPos = buf.position();
buf.compact();
int offsetPos = buf.position() - curPos;
readPosition += offsetPos;
return -1;
}
return b;
if (activeBuffer == null)
{
activeBuffer = buffers.take();
}
while (activeBuffer.remaining() <= 0)
{
if (buffersExhausted.get())
{
closed.set(true);
return -1;
}
activeBuffer = buffers.take();
}
return activeBuffer.get();
}
catch (InterruptedException e)
{
throw new IOException(e);
}
}
@Override
public synchronized void reset() throws IOException
{
throw new IOException("reset() not supported");
}
}

67
jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageReader.java
View File

@ -19,79 +19,36 @@
package org.eclipse.jetty.websocket.common.message;
import java.io.IOException;
import java.io.Reader;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import org.eclipse.jetty.util.Utf8StringBuilder;
import org.eclipse.jetty.websocket.common.events.EventDriver;
import org.eclipse.jetty.util.StringUtil;
/**
* Support class for reading text message data as an Reader.
* Support class for reading a (single) WebSocket TEXT message via a Reader.
* <p>
* Due to the spec, this reader is forced to use the UTF8 charset.
* In compliance to the WebSocket spec, this reader always uses the UTF8 {@link Charset}.
*/
public class MessageReader extends Reader implements MessageAppender
public class MessageReader extends InputStreamReader implements MessageAppender
{
private final EventDriver driver;
private final Utf8StringBuilder utf;
private int size;
private boolean finished;
private boolean needsNotification;
private final MessageInputStream stream;
public MessageReader(EventDriver driver)
public MessageReader(MessageInputStream stream)
{
this.driver = driver;
this.utf = new Utf8StringBuilder();
size = 0;
finished = false;
needsNotification = true;
super(stream,StringUtil.__UTF8_CHARSET);
this.stream = stream;
}
@Override
public void appendMessage(ByteBuffer payload, boolean isLast) throws IOException
{
if (finished)
{
throw new IOException("Cannot append to finished buffer");
}
if (payload == null)
{
// empty payload is valid
return;
}
driver.getPolicy().assertValidTextMessageSize(size + payload.remaining());
size += payload.remaining();
synchronized (utf)
{
utf.append(payload);
}
if (needsNotification)
{
needsNotification = true;
this.driver.onReader(this);
}
}
@Override
public void close() throws IOException
{
finished = true;
this.stream.appendMessage(payload,isLast);
}
@Override
public void messageComplete()
{
finished = true;
}
@Override
public int read(char[] cbuf, int off, int len) throws IOException
{
// TODO Auto-generated method stub
return 0;
this.stream.messageComplete();
}
}

153
jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/PayloadInputStream.java
View File

@ -1,153 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2013 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.websocket.common.message;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicBoolean;
import org.eclipse.jetty.websocket.common.LogicalConnection;
/**
* An InputStream that can access a queue of ByteBuffer payloads, along with expected InputStream blocking behavior.
*/
public class PayloadInputStream extends InputStream implements MessageAppender
{
/**
* Used for controlling read suspend/resume behavior if the queue is full, but the read operations haven't caught up yet.
*/
@SuppressWarnings("unused")
private final LogicalConnection connection;
private final BlockingDeque<ByteBuffer> buffers = new LinkedBlockingDeque<>();
private AtomicBoolean closed = new AtomicBoolean(false);
// EOB / End of Buffers
private AtomicBoolean buffersExhausted = new AtomicBoolean(false);
private ByteBuffer activeBuffer = null;
public PayloadInputStream(LogicalConnection connection)
{
this.connection = connection;
}
@Override
public void appendMessage(ByteBuffer payload, boolean isLast) throws IOException
{
if (buffersExhausted.get())
{
// This indicates a programming mistake/error and must be bug fixed
throw new RuntimeException("Last frame already received");
}
// if closed, we should just toss incoming payloads into the bit bucket.
if (closed.get())
{
return;
}
// Put the payload into the queue
try
{
buffers.put(payload);
if (isLast)
{
buffersExhausted.set(true);
}
}
catch (InterruptedException e)
{
throw new IOException(e);
}
}
@Override
public void close() throws IOException
{
closed.set(true);
super.close();
}
@Override
public synchronized void mark(int readlimit)
{
/* do nothing */
}
@Override
public boolean markSupported()
{
return false;
}
@Override
public void messageComplete()
{
buffersExhausted.set(true);
// toss an empty ByteBuffer into queue to let it drain
try
{
buffers.put(ByteBuffer.wrap(new byte[0]));
}
catch (InterruptedException ignore)
{
/* ignore */
}
}
@Override
public int read() throws IOException
{
try
{
if (closed.get())
{
return -1;
}
if (activeBuffer == null)
{
activeBuffer = buffers.take();
}
while (activeBuffer.remaining() <= 0)
{
if (buffersExhausted.get())
{
closed.set(true);
return -1;
}
activeBuffer = buffers.take();
}
return activeBuffer.get();
}
catch (InterruptedException e)
{
throw new IOException(e);
}
}
@Override
public synchronized void reset() throws IOException
{
throw new IOException("reset() not supported");
}
}

194
jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/message/MessageInputStreamTest.java
View File

@ -22,84 +22,166 @@ import static org.hamcrest.Matchers.*;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.eclipse.jetty.toolchain.test.TestTracker;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
import org.eclipse.jetty.websocket.common.events.EventDriver;
import org.eclipse.jetty.websocket.common.events.EventDriverFactory;
import org.eclipse.jetty.websocket.common.io.FramePipes;
import org.eclipse.jetty.websocket.common.io.LocalWebSocketSession;
import org.junit.After;
import org.eclipse.jetty.websocket.common.io.LocalWebSocketConnection;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
public class MessageInputStreamTest
{
@Rule
public TestTracker testtracker = new TestTracker();
private static final Charset UTF8 = StringUtil.__UTF8_CHARSET;
@Rule
public TestName testname = new TestName();
private WebSocketPolicy policy;
private TrackingInputStreamSocket socket;
private LocalWebSocketSession session;
private LocalWebSocketSession remoteSession;
@After
public void closeSession()
@Test
public void testBasicAppendRead() throws IOException
{
session.close();
remoteSession.close();
}
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
@Before
public void setupSession()
{
policy = WebSocketPolicy.newServerPolicy();
policy.setInputBufferSize(1024);
policy.setMaxBinaryMessageBufferSize(1024);
policy.setMaxTextMessageBufferSize(1024);
try (MessageInputStream stream = new MessageInputStream(conn))
{
// Append a message (simple, short)
ByteBuffer payload = BufferUtil.toBuffer("Hello World",UTF8);
System.out.printf("payload = %s%n",BufferUtil.toDetailString(payload));
boolean fin = true;
stream.appendMessage(payload,fin);
// Event Driver factory
EventDriverFactory factory = new EventDriverFactory(policy);
// Read it from the stream.
byte buf[] = new byte[32];
int len = stream.read(buf);
String message = new String(buf,0,len,UTF8);
// Local Socket
EventDriver localDriver = factory.wrap(new DummySocket());
// Remote socket & Session
socket = new TrackingInputStreamSocket("remote");
EventDriver remoteDriver = factory.wrap(socket);
remoteSession = new LocalWebSocketSession(testname,remoteDriver);
remoteSession.open();
OutgoingFrames socketPipe = FramePipes.to(remoteDriver);
// Local Session
session = new LocalWebSocketSession(testname,localDriver);
session.setPolicy(policy);
// talk to our remote socket
session.setOutgoingHandler(socketPipe);
// open connection
session.open();
// Test it
Assert.assertThat("Message",message,is("Hello World"));
}
}
@Test
@Ignore
public void testSimpleMessage() throws IOException
public void testBlockOnRead() throws IOException
{
ByteBuffer data = BufferUtil.toBuffer("Hello World",StringUtil.__UTF8_CHARSET);
session.getRemote().sendBytes(data);
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
Assert.assertThat("Socket.messageQueue.size",socket.messageQueue.size(),is(1));
String msg = socket.messageQueue.poll();
Assert.assertThat("Message",msg,is("Hello World"));
try (MessageInputStream stream = new MessageInputStream(conn))
{
final AtomicBoolean hadError = new AtomicBoolean(false);
new Thread(new Runnable()
{
@Override
public void run()
{
try
{
boolean fin = false;
TimeUnit.MILLISECONDS.sleep(200);
stream.appendMessage(BufferUtil.toBuffer("Saved",UTF8),fin);
TimeUnit.MILLISECONDS.sleep(200);
stream.appendMessage(BufferUtil.toBuffer(" by ",UTF8),fin);
fin = true;
TimeUnit.MILLISECONDS.sleep(200);
stream.appendMessage(BufferUtil.toBuffer("Zero",UTF8),fin);
}
catch (IOException | InterruptedException e)
{
hadError.set(true);
e.printStackTrace(System.err);
}
}
}).start();
// Read it from the stream.
byte buf[] = new byte[32];
int len = stream.read(buf);
String message = new String(buf,0,len,UTF8);
// Test it
Assert.assertThat("Error when appending",hadError.get(),is(false));
Assert.assertThat("Message",message,is("Saved by Zero"));
}
}
@Test
public void testBlockOnReadInitial() throws IOException
{
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
try (MessageInputStream stream = new MessageInputStream(conn))
{
final AtomicBoolean hadError = new AtomicBoolean(false);
new Thread(new Runnable()
{
@Override
public void run()
{
try
{
boolean fin = true;
// wait for a little bit before populating buffers
TimeUnit.MILLISECONDS.sleep(400);
stream.appendMessage(BufferUtil.toBuffer("I will conquer",UTF8),fin);
}
catch (IOException | InterruptedException e)
{
hadError.set(true);
e.printStackTrace(System.err);
}
}
}).start();
// Read byte from stream.
int b = stream.read();
// Should be a byte, blocking till byte received.
// Test it
Assert.assertThat("Error when appending",hadError.get(),is(false));
Assert.assertThat("Initial byte",b,is((int)'I'));
}
}
@Test
public void testReadByteNoBuffersClosed() throws IOException
{
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
try (MessageInputStream stream = new MessageInputStream(conn))
{
final AtomicBoolean hadError = new AtomicBoolean(false);
new Thread(new Runnable()
{
@Override
public void run()
{
try
{
// wait for a little bit before sending input closed
TimeUnit.MILLISECONDS.sleep(400);
stream.messageComplete();
}
catch (InterruptedException e)
{
hadError.set(true);
e.printStackTrace(System.err);
}
}
}).start();
// Read byte from stream.
int b = stream.read();
// Should be a -1, indicating the end of the stream.
// Test it
Assert.assertThat("Error when appending",hadError.get(),is(false));
Assert.assertThat("Initial byte",b,is(-1));
}
}
}

187
jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/message/PayloadInputStreamTest.java
View File

@ -1,187 +0,0 @@
//
// ========================================================================
// Copyright (c) 1995-2013 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// and Apache License v2.0 which accompanies this distribution.
//
// The Eclipse Public License is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// The Apache License v2.0 is available at
// http://www.opensource.org/licenses/apache2.0.php
//
// You may elect to redistribute this code under either of these licenses.
// ========================================================================
//
package org.eclipse.jetty.websocket.common.message;
import static org.hamcrest.Matchers.*;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.websocket.common.io.LocalWebSocketConnection;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
public class PayloadInputStreamTest
{
private static final Charset UTF8 = StringUtil.__UTF8_CHARSET;
@Rule
public TestName testname = new TestName();
@Test
public void testBasicAppendRead() throws IOException
{
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
try (PayloadInputStream stream = new PayloadInputStream(conn))
{
// Append a message (simple, short)
ByteBuffer payload = BufferUtil.toBuffer("Hello World",UTF8);
System.out.printf("payload = %s%n",BufferUtil.toDetailString(payload));
boolean fin = true;
stream.appendMessage(payload,fin);
// Read it from the stream.
byte buf[] = new byte[32];
int len = stream.read(buf);
String message = new String(buf,0,len,UTF8);
// Test it
Assert.assertThat("Message",message,is("Hello World"));
}
}
@Test
public void testBlockOnRead() throws IOException
{
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
try (PayloadInputStream stream = new PayloadInputStream(conn))
{
final AtomicBoolean hadError = new AtomicBoolean(false);
new Thread(new Runnable()
{
@Override
public void run()
{
try
{
boolean fin = false;
TimeUnit.MILLISECONDS.sleep(200);
stream.appendMessage(BufferUtil.toBuffer("Saved",UTF8),fin);
TimeUnit.MILLISECONDS.sleep(200);
stream.appendMessage(BufferUtil.toBuffer(" by ",UTF8),fin);
fin = true;
TimeUnit.MILLISECONDS.sleep(200);
stream.appendMessage(BufferUtil.toBuffer("Zero",UTF8),fin);
}
catch (IOException | InterruptedException e)
{
hadError.set(true);
e.printStackTrace(System.err);
}
}
}).start();
// Read it from the stream.
byte buf[] = new byte[32];
int len = stream.read(buf);
String message = new String(buf,0,len,UTF8);
// Test it
Assert.assertThat("Error when appending",hadError.get(),is(false));
Assert.assertThat("Message",message,is("Saved by Zero"));
}
}
@Test
public void testBlockOnReadInitial() throws IOException
{
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
try (PayloadInputStream stream = new PayloadInputStream(conn))
{
final AtomicBoolean hadError = new AtomicBoolean(false);
new Thread(new Runnable()
{
@Override
public void run()
{
try
{
boolean fin = true;
// wait for a little bit before populating buffers
TimeUnit.MILLISECONDS.sleep(400);
stream.appendMessage(BufferUtil.toBuffer("I will conquer",UTF8),fin);
}
catch (IOException | InterruptedException e)
{
hadError.set(true);
e.printStackTrace(System.err);
}
}
}).start();
// Read byte from stream.
int b = stream.read();
// Should be a byte, blocking till byte received.
// Test it
Assert.assertThat("Error when appending",hadError.get(),is(false));
Assert.assertThat("Initial byte",b,is((int)'I'));
}
}
@Test
public void testReadByteNoBuffersClosed() throws IOException
{
LocalWebSocketConnection conn = new LocalWebSocketConnection(testname);
try (PayloadInputStream stream = new PayloadInputStream(conn))
{
final AtomicBoolean hadError = new AtomicBoolean(false);
new Thread(new Runnable()
{
@Override
public void run()
{
try
{
// wait for a little bit before sending input closed
TimeUnit.MILLISECONDS.sleep(400);
stream.messageComplete();
}
catch (InterruptedException e)
{
hadError.set(true);
e.printStackTrace(System.err);
}
}
}).start();
// Read byte from stream.
int b = stream.read();
// Should be a -1, indicating the end of the stream.
// Test it
Assert.assertThat("Error when appending",hadError.get(),is(false));
Assert.assertThat("Initial byte",b,is(-1));
}
}
}