andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

* Made sure ChannelFutures are notified when no lock is acquired in HttpTunnelingChannelHandler

Browse Source 
* Reduced the visibility of some methods in HttpTunnelingChannelHandler
This commit is contained in:
Trustin Lee 2009-03-12 07:01:20 +00:00
parent 937ee06d9d
commit 5f1dd20d74
1 changed files with 33 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
src/main/java/org/jboss/netty/channel/socket/http/HttpTunnelingChannelHandler.java
View File

@ -56,7 +56,7 @@ class HttpTunnelingChannelHandler extends SimpleChannelHandler {
private final Condition reconnectCondition = reconnectLock.newCondition(); private final Condition reconnectCondition = reconnectLock.newCondition();
private final long reconnectTimeout; private final long reconnectTimeoutMillis;
private volatile boolean connected = false; private volatile boolean connected = false;
@ -68,10 +68,11 @@ class HttpTunnelingChannelHandler extends SimpleChannelHandler {
private final HttpSession session; private final HttpSession session;
public HttpTunnelingChannelHandler(boolean stream, HttpSession session, long reconnectTimeout) { public HttpTunnelingChannelHandler(
boolean stream, HttpSession session, long reconnectTimeoutMillis) {
this.stream = stream; this.stream = stream;
this.session = session; this.session = session;
this.reconnectTimeout = reconnectTimeout; this.reconnectTimeoutMillis = reconnectTimeoutMillis;
} }
@Override @Override
@ -79,6 +80,8 @@ class HttpTunnelingChannelHandler extends SimpleChannelHandler {
ChannelBuffer buffer = (ChannelBuffer) e.getMessage(); ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
if (stream) { if (stream) {
boolean success = false;
Throwable cause = null;
byte[] b = null; byte[] b = null;
reconnectLock.lock(); reconnectLock.lock();
try { try {
@ -90,30 +93,35 @@ class HttpTunnelingChannelHandler extends SimpleChannelHandler {
buffer.readBytes(b); buffer.readBytes(b);
outputStream.write(b); outputStream.write(b);
outputStream.flush(); outputStream.flush();
e.getFuture().setSuccess(); success = true;
} } catch (Throwable t) {
catch (IOException e1) { success = false;
connected = false; cause = t;
reconnectCondition.await(reconnectTimeout, TimeUnit.MILLISECONDS); if (awaitReconnect()) {
if (connected) { try {
outputStream.write(b); outputStream.write(b);
outputStream.flush(); outputStream.flush();
e.getFuture().setSuccess(); success = true;
} catch (Throwable t2) {
success = false;
cause = t2;
} }
else { } else {
e.getFuture().setFailure(e1);
if (invalidated.compareAndSet(false, true)) { if (invalidated.compareAndSet(false, true)) {
session.invalidate(); session.invalidate();
} }
e.getChannel().close(); e.getChannel().close();
} }
} } finally {
finally {
reconnectLock.unlock(); reconnectLock.unlock();
if (success) {
e.getFuture().setSuccess();
} else {
assert cause != null;
e.getFuture().setFailure(cause);
} }
} }
} else {
else {
awaitingEvents.add(e); awaitingEvents.add(e);
} }
@ -134,14 +142,14 @@ class HttpTunnelingChannelHandler extends SimpleChannelHandler {
} }
} }
public synchronized List<MessageEvent> getAwaitingEvents() { synchronized List<MessageEvent> getAwaitingEvents() {
List<MessageEvent> list = new ArrayList<MessageEvent>(); List<MessageEvent> list = new ArrayList<MessageEvent>();
list.addAll(awaitingEvents); list.addAll(awaitingEvents);
awaitingEvents.clear(); awaitingEvents.clear();
return list; return list;
} }
public void setOutputStream(ServletOutputStream outputStream) { void setOutputStream(ServletOutputStream outputStream) {
reconnectLock.lock(); reconnectLock.lock();
try { try {
this.outputStream = outputStream; this.outputStream = outputStream;
@ -166,19 +174,15 @@ class HttpTunnelingChannelHandler extends SimpleChannelHandler {
} }
} }
public boolean isStreaming() { boolean isStreaming() {
return stream; return stream;
} }
public ServletOutputStream getOutputStream() { boolean awaitReconnect() {
return outputStream;
}
public boolean awaitReconnect() {
reconnectLock.lock(); reconnectLock.lock();
try { try {
connected = false; connected = false;
reconnectCondition.await(reconnectTimeout, TimeUnit.MILLISECONDS); reconnectCondition.await(reconnectTimeoutMillis, TimeUnit.MILLISECONDS);
} }
catch (InterruptedException e) { catch (InterruptedException e) {
return connected; return connected;