Revised exception handling.
This commit is contained in:
parent
1238de8a28
commit
3b8c6dfd4c
|
@ -142,17 +142,8 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
int streamId = streamIds.getAndAdd(2);
|
int streamId = streamIds.getAndAdd(2);
|
||||||
SynStreamFrame synStream = new SynStreamFrame(version, synInfo.getFlags(), streamId, 0, synInfo.getPriority(), synInfo.getHeaders());
|
SynStreamFrame synStream = new SynStreamFrame(version, synInfo.getFlags(), streamId, 0, synInfo.getPriority(), synInfo.getHeaders());
|
||||||
final IStream stream = createStream(synStream, listener);
|
final IStream stream = createStream(synStream, listener);
|
||||||
try
|
|
||||||
{
|
|
||||||
// May throw if wrong version or headers too big
|
|
||||||
control(stream, synStream, timeout, unit, handler, stream);
|
control(stream, synStream, timeout, unit, handler, stream);
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
removeStream(stream);
|
|
||||||
handler.failed(x);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -166,8 +157,6 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void rst(RstInfo rstInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
public void rst(RstInfo rstInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
{
|
||||||
// SPEC v3, 2.2.2
|
// SPEC v3, 2.2.2
|
||||||
if (goAwaySent.get())
|
if (goAwaySent.get())
|
||||||
|
@ -180,12 +169,6 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
control(null, frame, timeout, unit, handler, null);
|
control(null, frame, timeout, unit, handler, null);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
logger.info("Could not send reset on stream " + rstInfo.getStreamId(), x);
|
|
||||||
handler.failed(x);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Future<Void> settings(SettingsInfo settingsInfo)
|
public Future<Void> settings(SettingsInfo settingsInfo)
|
||||||
|
@ -197,17 +180,10 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void settings(SettingsInfo settingsInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
public void settings(SettingsInfo settingsInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
{
|
||||||
SettingsFrame frame = new SettingsFrame(version, settingsInfo.getFlags(), settingsInfo.getSettings());
|
SettingsFrame frame = new SettingsFrame(version, settingsInfo.getFlags(), settingsInfo.getSettings());
|
||||||
control(null, frame, timeout, unit, handler, null);
|
control(null, frame, timeout, unit, handler, null);
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
handler.failed(x);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Future<PingInfo> ping()
|
public Future<PingInfo> ping()
|
||||||
|
@ -222,16 +198,9 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
{
|
{
|
||||||
int pingId = pingIds.getAndAdd(2);
|
int pingId = pingIds.getAndAdd(2);
|
||||||
PingInfo pingInfo = new PingInfo(pingId);
|
PingInfo pingInfo = new PingInfo(pingId);
|
||||||
try
|
|
||||||
{
|
|
||||||
PingFrame frame = new PingFrame(version, pingId);
|
PingFrame frame = new PingFrame(version, pingId);
|
||||||
control(null, frame, timeout, unit, handler, pingInfo);
|
control(null, frame, timeout, unit, handler, pingInfo);
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
handler.failed(x);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Future<Void> goAway()
|
public Future<Void> goAway()
|
||||||
|
@ -247,18 +216,11 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
if (goAwaySent.compareAndSet(false, true))
|
if (goAwaySent.compareAndSet(false, true))
|
||||||
{
|
{
|
||||||
if (!goAwayReceived.get())
|
if (!goAwayReceived.get())
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
{
|
||||||
GoAwayFrame frame = new GoAwayFrame(version, lastStreamId.get(), SessionStatus.OK.getCode());
|
GoAwayFrame frame = new GoAwayFrame(version, lastStreamId.get(), SessionStatus.OK.getCode());
|
||||||
control(null, frame, timeout, unit, handler, null);
|
control(null, frame, timeout, unit, handler, null);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
handler.failed(x);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
handler.completed(null);
|
handler.completed(null);
|
||||||
}
|
}
|
||||||
|
@ -554,8 +516,6 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
}
|
}
|
||||||
|
|
||||||
private void onPing(final PingFrame frame)
|
private void onPing(final PingFrame frame)
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
{
|
||||||
int pingId = frame.getPingId();
|
int pingId = frame.getPingId();
|
||||||
if (pingId % 2 == pingIds.get() % 2)
|
if (pingId % 2 == pingIds.get() % 2)
|
||||||
|
@ -576,11 +536,6 @@ public class StandardSession implements ISession, Parser.Listener, Handler<Stand
|
||||||
control(null, frame, 0, TimeUnit.MILLISECONDS, new Promise<>(), null);
|
control(null, frame, 0, TimeUnit.MILLISECONDS, new Promise<>(), null);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
throw new SPDYException(x);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private void onGoAway(final GoAwayFrame frame)
|
private void onGoAway(final GoAwayFrame frame)
|
||||||
{
|
{
|
||||||
|
|
|
@ -203,14 +203,13 @@ public class StandardStream implements IStream
|
||||||
super.consume(delta);
|
super.consume(delta);
|
||||||
|
|
||||||
// This is the algorithm for flow control.
|
// This is the algorithm for flow control.
|
||||||
// This method may be called multiple times
|
// This method may be called multiple times with delta=1, but we only send a window
|
||||||
// with delta=1, but we only send a window
|
// update when the whole dataInfo has been consumed.
|
||||||
// update when the whole dataInfo has been
|
// Other policies may be to send window updates when consumed() is greater than
|
||||||
// consumed.
|
// a certain threshold, etc. but for now the policy is not pluggable for simplicity.
|
||||||
// Other policies may be to send window
|
// Note that the frequency of window updates depends on the read buffer, that
|
||||||
// updates when consumed() is greater than
|
// should not be too smaller than the window size to avoid frequent window updates.
|
||||||
// a certain threshold, etc. but for now
|
// Therefore, a pluggable policy should be able to modify the read buffer capacity.
|
||||||
// the policy is not pluggable for simplicity.
|
|
||||||
if (consumed() == length() && !isClosed())
|
if (consumed() == length() && !isClosed())
|
||||||
windowUpdate(length());
|
windowUpdate(length());
|
||||||
}
|
}
|
||||||
|
@ -259,24 +258,13 @@ public class StandardStream implements IStream
|
||||||
}
|
}
|
||||||
|
|
||||||
private void windowUpdate(int delta)
|
private void windowUpdate(int delta)
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
{
|
||||||
if (delta > 0)
|
if (delta > 0)
|
||||||
{
|
{
|
||||||
// TODO: if the read buffer is small, but the default window size is big,
|
|
||||||
// we will send many window update frames... perhaps we can delay
|
|
||||||
// window update frames until we have a bigger delta to send
|
|
||||||
WindowUpdateFrame windowUpdateFrame = new WindowUpdateFrame(session.getVersion(), getId(), delta);
|
WindowUpdateFrame windowUpdateFrame = new WindowUpdateFrame(session.getVersion(), getId(), delta);
|
||||||
session.control(this, windowUpdateFrame, 0, TimeUnit.MILLISECONDS, new Promise<>(), null);
|
session.control(this, windowUpdateFrame, 0, TimeUnit.MILLISECONDS, new Promise<>(), null);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
logger.debug("Could not send window update on stream " + this, x);
|
|
||||||
session.rst(new RstInfo(getId(), x.getStreamStatus()));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private void notifyOnReply(ReplyInfo replyInfo)
|
private void notifyOnReply(ReplyInfo replyInfo)
|
||||||
{
|
{
|
||||||
|
@ -340,20 +328,11 @@ public class StandardStream implements IStream
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void reply(ReplyInfo replyInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
public void reply(ReplyInfo replyInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
{
|
||||||
updateCloseState(replyInfo.isClose());
|
updateCloseState(replyInfo.isClose());
|
||||||
SynReplyFrame frame = new SynReplyFrame(session.getVersion(), replyInfo.getFlags(), getId(), replyInfo.getHeaders());
|
SynReplyFrame frame = new SynReplyFrame(session.getVersion(), replyInfo.getFlags(), getId(), replyInfo.getHeaders());
|
||||||
session.control(this, frame, timeout, unit, handler, null);
|
session.control(this, frame, timeout, unit, handler, null);
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
logger.debug("Could not send reply on stream " + this, x);
|
|
||||||
handler.failed(x);
|
|
||||||
session.rst(new RstInfo(getId(), x.getStreamStatus()));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Future<Void> data(DataInfo dataInfo)
|
public Future<Void> data(DataInfo dataInfo)
|
||||||
|
@ -381,20 +360,11 @@ public class StandardStream implements IStream
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void headers(HeadersInfo headersInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
public void headers(HeadersInfo headersInfo, long timeout, TimeUnit unit, Handler<Void> handler)
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
{
|
||||||
updateCloseState(headersInfo.isClose());
|
updateCloseState(headersInfo.isClose());
|
||||||
HeadersFrame frame = new HeadersFrame(session.getVersion(), headersInfo.getFlags(), getId(), headersInfo.getHeaders());
|
HeadersFrame frame = new HeadersFrame(session.getVersion(), headersInfo.getFlags(), getId(), headersInfo.getHeaders());
|
||||||
session.control(this, frame, timeout, unit, handler, null);
|
session.control(this, frame, timeout, unit, handler, null);
|
||||||
}
|
}
|
||||||
catch (StreamException x)
|
|
||||||
{
|
|
||||||
logger.debug("Could not send headers on stream " + this, x);
|
|
||||||
handler.failed(x);
|
|
||||||
session.rst(new RstInfo(getId(), x.getStreamStatus()));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean isClosed()
|
public boolean isClosed()
|
||||||
|
|
Loading…
Reference in New Issue