You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ph...@apache.org on 2013/05/21 17:49:53 UTC
svn commit: r1484842 [2/4] - in /qpid/proton/trunk:
proton-c/bindings/java/src/main/java/org/apache/qpid/proton/engine/jni/
proton-j/proton-api/src/main/java/org/apache/qpid/proton/engine/
proton-j/proton-api/src/main/resources/ proton-j/proton/ proton...
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/SaslImpl.java Tue May 21 15:49:52 2013
@@ -1,4 +1,3 @@
-package org.apache.qpid.proton.engine.impl;
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
@@ -20,39 +19,50 @@ package org.apache.qpid.proton.engine.im
*
*/
+package org.apache.qpid.proton.engine.impl;
+
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.newReadableBuffer;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.newWriteableBuffer;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.pour;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.pourAll;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.pourBufferToArray;
import java.nio.ByteBuffer;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.security.SaslChallenge;
+import org.apache.qpid.proton.amqp.security.SaslCode;
+import org.apache.qpid.proton.amqp.security.SaslFrameBody;
+import org.apache.qpid.proton.amqp.security.SaslInit;
+import org.apache.qpid.proton.amqp.security.SaslMechanisms;
+import org.apache.qpid.proton.amqp.security.SaslResponse;
+import org.apache.qpid.proton.codec.AMQPDefinedTypes;
import org.apache.qpid.proton.codec.CompositeWritableBuffer;
import org.apache.qpid.proton.codec.DecoderImpl;
import org.apache.qpid.proton.codec.EncoderImpl;
import org.apache.qpid.proton.codec.WritableBuffer;
import org.apache.qpid.proton.engine.Sasl;
-import org.apache.qpid.proton.codec.AMQPDefinedTypes;
-import org.apache.qpid.proton.amqp.Binary;
-import org.apache.qpid.proton.amqp.Symbol;
-import org.apache.qpid.proton.amqp.security.*;
+import org.apache.qpid.proton.engine.TransportResult;
+import org.apache.qpid.proton.engine.TransportResultFactory;
-public class SaslImpl implements Sasl, SaslFrameBody.SaslFrameBodyHandler<Void>
+public class SaslImpl implements Sasl, SaslFrameBody.SaslFrameBodyHandler<Void>, SaslFrameHandler
{
- public static final byte SASL_FRAME_TYPE = (byte) 1;
+ private static final Logger _logger = Logger.getLogger(SaslImpl.class.getName());
- public static final byte[] HEADER =
- new byte[] { (byte) 'A',
- (byte) 'M',
- (byte) 'Q',
- (byte) 'P',
- 3,
- 1,
- 0,
- 0
- };
+ public static final byte SASL_FRAME_TYPE = (byte) 1;
- private ByteBuffer _pending;
private final DecoderImpl _decoder = new DecoderImpl();
private final EncoderImpl _encoder = new EncoderImpl(_decoder);
- private int _maxFrameSize = 4096;
- private final ByteBuffer _overflowBuffer = ByteBuffer.wrap(new byte[_maxFrameSize]);
+
+ private final ByteBuffer _inputBuffer;
+ private final ByteBuffer _outputBuffer;
+ private final ByteBuffer _outputOverflowBuffer;
+
+ private ByteBuffer _pending;
+
private boolean _headerWritten;
private Binary _challengeResponse;
private SaslFrameParser _frameParser;
@@ -60,7 +70,6 @@ public class SaslImpl implements Sasl, S
private boolean _mechanismsSent;
private boolean _initSent;
-
enum Role { CLIENT, SERVER };
private SaslOutcome _outcome = SaslOutcome.PN_SASL_NONE;
@@ -72,72 +81,67 @@ public class SaslImpl implements Sasl, S
private Symbol _chosenMechanism;
-
private Role _role;
- SaslImpl()
- {
- _frameParser = new SaslFrameParser(this);
+ /**
+ * @param maxFrameSize the size of the input and output buffers
+ * returned by {@link SaslTransportWrapper#getInputBuffer()} and
+ * {@link SaslTransportWrapper#getOutputBuffer()}.
+ */
+ SaslImpl(int maxFrameSize)
+ {
+ _inputBuffer = newWriteableBuffer(maxFrameSize);
+ _outputBuffer = newWriteableBuffer(maxFrameSize);
+ _outputOverflowBuffer = newReadableBuffer(maxFrameSize);
+
AMQPDefinedTypes.registerAllTypes(_decoder,_encoder);
- _overflowBuffer.flip();
+ _frameParser = new SaslFrameParser(this, _decoder);
}
- boolean isDone()
+ @Override
+ public boolean isDone()
{
return _done && (_role==Role.CLIENT || _initReceived);
}
- private final int input(byte[] bytes, int offset, int size)
+ private void writeSaslOutput()
{
- if(isDone())
+ if(_outputOverflowBuffer.hasRemaining())
{
- return TransportImpl.END_OF_STREAM;
+ _outputBuffer.put(_outputOverflowBuffer);
}
- else
+ if(!_outputOverflowBuffer.hasRemaining())
{
- return getFrameParser().input(bytes, offset, size);
- }
- }
-
- private final int output(byte[] bytes, int offset, int size)
- {
+ _outputOverflowBuffer.clear();
- int written = 0;
- if(_overflowBuffer.hasRemaining())
- {
- final int overflowWritten = Math.min(size, _overflowBuffer.remaining());
- _overflowBuffer.get(bytes, offset, overflowWritten);
- written+=overflowWritten;
- }
- if(!_overflowBuffer.hasRemaining())
- {
- _overflowBuffer.rewind();
-
- CompositeWritableBuffer outputBuffer =
+ CompositeWritableBuffer compositeOutputBuffer =
new CompositeWritableBuffer(
- new WritableBuffer.ByteBufferWrapper(ByteBuffer.wrap(bytes, offset + written, size - written)),
- new WritableBuffer.ByteBufferWrapper(_overflowBuffer));
+ new WritableBuffer.ByteBufferWrapper(_outputBuffer),
+ new WritableBuffer.ByteBufferWrapper(_outputOverflowBuffer));
+ process(compositeOutputBuffer);
- written += process(outputBuffer);
+ _outputOverflowBuffer.flip();
}
- return written;
- }
+ if(_logger.isLoggable(Level.FINER))
+ {
+ _logger.log(Level.FINER, "Finished writing SASL output. Output Buffer : " + _outputBuffer);
+ }
+ }
- protected int process(WritableBuffer buffer)
+ private void process(WritableBuffer buffer)
{
- int written = processHeader(buffer);
+ processHeader(buffer);
if(_role == Role.SERVER)
{
-
if(!_mechanismsSent && _mechanisms != null)
{
SaslMechanisms mechanisms = new SaslMechanisms();
mechanisms.setSaslServerMechanisms(_mechanisms);
- written += writeFrame(buffer, mechanisms);
+ writeFrame(buffer, mechanisms);
_mechanismsSent = true;
_state = SaslState.PN_SASL_STEP;
}
@@ -146,7 +150,7 @@ public class SaslImpl implements Sasl, S
{
SaslChallenge challenge = new SaslChallenge();
challenge.setChallenge(getChallengeResponse());
- written+=writeFrame(buffer, challenge);
+ writeFrame(buffer, challenge);
setChallengeResponse(null);
}
@@ -155,26 +159,24 @@ public class SaslImpl implements Sasl, S
org.apache.qpid.proton.amqp.security.SaslOutcome outcome =
new org.apache.qpid.proton.amqp.security.SaslOutcome();
outcome.setCode(SaslCode.values()[_outcome.getCode()]);
- written+=writeFrame(buffer, outcome);
+ writeFrame(buffer, outcome);
}
}
else if(_role == Role.CLIENT)
{
if(getState() == SaslState.PN_SASL_IDLE && _chosenMechanism != null)
{
- written += processInit(buffer);
+ processInit(buffer);
_state = SaslState.PN_SASL_STEP;
}
if(getState() == SaslState.PN_SASL_STEP && getChallengeResponse() != null)
{
- written += processResponse(buffer);
+ processResponse(buffer);
}
}
-
- return written;
}
- int writeFrame(WritableBuffer buffer, SaslFrameBody frameBody)
+ private void writeFrame(WritableBuffer buffer, SaslFrameBody frameBody)
{
int oldPosition = buffer.position();
buffer.position(buffer.position()+8);
@@ -190,17 +192,20 @@ public class SaslImpl implements Sasl, S
buffer.putShort((short) 0);
buffer.position(limit);
- return frameSize;
+ if(_logger.isLoggable(Level.FINE))
+ {
+ _logger.fine(this + " wrote frame " + frameBody + " (" + frameSize + ") bytes");
+ }
}
+ @Override
final public int recv(byte[] bytes, int offset, int size)
{
if(_pending == null)
{
return -1;
}
- final int written = Math.min(size, _pending.remaining());
- _pending.get(bytes, offset, written);
+ final int written = pourBufferToArray(_pending, bytes, offset, size);
if(!_pending.hasRemaining())
{
_pending = null;
@@ -208,6 +213,7 @@ public class SaslImpl implements Sasl, S
return written;
}
+ @Override
final public int send(byte[] bytes, int offset, int size)
{
byte[] data = new byte[size];
@@ -221,10 +227,10 @@ public class SaslImpl implements Sasl, S
if(!_headerWritten)
{
- outputBuffer.put(HEADER,0, HEADER.length);
+ outputBuffer.put(AmqpHeader.SASL_HEADER,0, AmqpHeader.SASL_HEADER.length);
_headerWritten = true;
- return HEADER.length;
+ return AmqpHeader.SASL_HEADER.length;
}
else
{
@@ -232,6 +238,7 @@ public class SaslImpl implements Sasl, S
}
}
+ @Override
public int pending()
{
return _pending == null ? 0 : _pending.remaining();
@@ -242,17 +249,12 @@ public class SaslImpl implements Sasl, S
_pending = pending;
}
+ @Override
public SaslState getState()
{
return _state;
}
-
- final DecoderImpl getDecoder()
- {
- return _decoder;
- }
-
final Binary getChallengeResponse()
{
return _challengeResponse;
@@ -263,13 +265,7 @@ public class SaslImpl implements Sasl, S
_challengeResponse = challengeResponse;
}
- final SaslFrameParser getFrameParser()
- {
- return _frameParser;
- }
-
-
-
+ @Override
public void setMechanisms(String[] mechanisms)
{
if(mechanisms != null)
@@ -290,6 +286,7 @@ public class SaslImpl implements Sasl, S
}
}
+ @Override
public String[] getRemoteMechanisms()
{
if(_role == Role.SERVER)
@@ -333,7 +330,13 @@ public class SaslImpl implements Sasl, S
setPending(initialResponse.asByteBuffer());
}
+ @Override
+ public void handle(SaslFrameBody frameBody, Binary payload)
+ {
+ frameBody.invoke(this, payload, null);
+ }
+ @Override
public void handleInit(SaslInit saslInit, Binary payload, Void context)
{
if(_role == null)
@@ -351,20 +354,21 @@ public class SaslImpl implements Sasl, S
}
}
-
+ @Override
public void handleResponse(SaslResponse saslResponse, Binary payload, Void context)
{
checkRole(Role.SERVER);
setPending(saslResponse.getResponse() == null ? null : saslResponse.getResponse().asByteBuffer());
}
-
+ @Override
public void done(SaslOutcome outcome)
{
checkRole(Role.SERVER);
_outcome = outcome;
_done = true;
_state = outcome == SaslOutcome.PN_SASL_OK ? SaslState.PN_SASL_PASS : SaslState.PN_SASL_FAIL;
+ _logger.fine("SASL negotiation done: " + this);
}
private void checkRole(Role role)
@@ -375,7 +379,7 @@ public class SaslImpl implements Sasl, S
}
}
-
+ @Override
public void handleMechanisms(SaslMechanisms saslMechanisms, Binary payload, Void context)
{
if(_role == null)
@@ -386,14 +390,14 @@ public class SaslImpl implements Sasl, S
_mechanisms = saslMechanisms.getSaslServerMechanisms();
}
-
+ @Override
public void handleChallenge(SaslChallenge saslChallenge, Binary payload, Void context)
{
checkRole(Role.CLIENT);
setPending(saslChallenge.getChallenge() == null ? null : saslChallenge.getChallenge().asByteBuffer());
}
-
+ @Override
public void handleOutcome(org.apache.qpid.proton.amqp.security.SaslOutcome saslOutcome,
Binary payload,
Void context)
@@ -408,16 +412,22 @@ public class SaslImpl implements Sasl, S
}
}
_done = true;
+
+ if(_logger.isLoggable(Level.FINE))
+ {
+ _logger.fine("Handled outcome: " + this);
+ }
}
- private int processResponse(WritableBuffer buffer)
+
+ private void processResponse(WritableBuffer buffer)
{
SaslResponse response = new SaslResponse();
response.setResponse(getChallengeResponse());
setChallengeResponse(null);
- return writeFrame(buffer, response);
+ writeFrame(buffer, response);
}
- private int processInit(WritableBuffer buffer)
+ private void processInit(WritableBuffer buffer)
{
SaslInit init = new SaslInit();
init.setHostname(_hostname);
@@ -428,9 +438,10 @@ public class SaslImpl implements Sasl, S
setChallengeResponse(null);
}
_initSent = true;
- return writeFrame(buffer, init);
+ writeFrame(buffer, init);
}
+ @Override
public void plain(String username, String password)
{
client();
@@ -445,11 +456,13 @@ public class SaslImpl implements Sasl, S
}
+ @Override
public SaslOutcome getOutcome()
{
return _outcome;
}
+ @Override
public void client()
{
_role = Role.CLIENT;
@@ -461,6 +474,7 @@ public class SaslImpl implements Sasl, S
}
}
+ @Override
public void server()
{
_role = Role.SERVER;
@@ -469,41 +483,151 @@ public class SaslImpl implements Sasl, S
public TransportWrapper wrap(final TransportInput input, final TransportOutput output)
{
- return new TransportWrapper()
+ return new SaslTransportWrapper(input, output);
+ }
+
+ @Override
+ public String toString()
+ {
+ StringBuilder builder = new StringBuilder();
+ builder
+ .append("SaslImpl [_outcome=").append(_outcome)
+ .append(", state=").append(_state)
+ .append(", done=").append(_done)
+ .append(", role=").append(_role)
+ .append("]");
+ return builder.toString();
+ }
+
+
+ private class SaslTransportWrapper implements TransportWrapper
+ {
+ private final TransportInput _underlyingInput;
+ private final TransportOutput _underlyingOutput;
+ private boolean _outputComplete;
+
+ /**
+ * We store the last result when processing input so that
+ * we know not to process any more input if it was an error.
+ */
+ private TransportResult _lastInputResult = TransportResultFactory.ok();
+ private ByteBuffer _readOnlyOutputBufferView;
+
+ private SaslTransportWrapper(TransportInput input, TransportOutput output)
{
- private boolean _outputComplete;
+ _underlyingInput = input;
+ _underlyingOutput = output;
+ }
- @Override
- public int input(byte[] bytes, int offset, int size)
+ private void fillOutputBuffer()
+ {
+ if(isOutputInSaslMode())
{
- if(_role == null || (_role == Role.CLIENT && !_done) ||(_role == Role.SERVER && (!_initReceived || !_done)))
+ SaslImpl.this.writeSaslOutput();
+ if(_done)
{
- return SaslImpl.this.input(bytes, offset, size);
+ _outputComplete = true;
}
- else
+
+ // TODO if sasl is now 'done', it would be more efficient if any remaining space is
+ // now offered directly to _output rather than awaiting the next invocation.
+ }
+ else
+ {
+ ByteBuffer outputBuffer = _underlyingOutput.getOutputBuffer();
+ pour(outputBuffer, _outputBuffer);
+
+ _underlyingOutput.outputConsumed();
+
+ if(_logger.isLoggable(Level.FINER))
{
- return input.input(bytes, offset, size);
+ _logger.log(Level.FINER, SaslImpl.this + " filled output buffer with plain output");
}
}
+ }
+
+ /**
+ * TODO rationalise this method with respect to the other similar checks of _role/_initReceived etc
+ * @see SaslImpl#isDone()
+ */
+ private boolean isInputInSaslMode()
+ {
+ return _role == null || (_role == Role.CLIENT && !_done) ||(_role == Role.SERVER && (!_initReceived || !_done));
+ }
+
+ private boolean isOutputInSaslMode()
+ {
+ return _role == null || (_role == Role.CLIENT && (!_done || !_initSent)) || (_role == Role.SERVER && !_outputComplete);
+ }
+
+ @Override
+ public ByteBuffer getInputBuffer()
+ {
+ // TODO if the SASL negotiation is complete, it would be more efficient
+ // to return the underlying transport input's buffer.
+ // The same optimisation would be possible in getOutputBuffer
+ _lastInputResult.checkIsOk();
+ return _inputBuffer;
+ }
+
+ @Override
+ public TransportResult processInput()
+ {
+ _inputBuffer.flip();
+
+ try
+ {
+ _lastInputResult = reallyProcessInput();
+ return _lastInputResult;
+ }
+ finally
+ {
+ _inputBuffer.compact();
+ }
+ }
+ private TransportResult reallyProcessInput()
+ {
+ TransportResult transportResult = TransportResultFactory.ok();
- @Override
- public int output(byte[] bytes, int offset, int size)
+ if(isInputInSaslMode())
{
- if(_role == null || (_role == Role.CLIENT && (!_done || !_initSent)) || (_role == Role.SERVER && !_outputComplete))
+ if(_logger.isLoggable(Level.FINER))
{
- int written = SaslImpl.this.output(bytes, offset, size);
- if(_done && !_overflowBuffer.hasRemaining())
- {
- _outputComplete = true;
- }
- return written;
+ _logger.log(Level.FINER, SaslImpl.this + " about to call input.");
}
- else
+
+ transportResult = _frameParser.input(_inputBuffer);
+ }
+
+ if(!isInputInSaslMode() && transportResult.isOk())
+ {
+ if(_logger.isLoggable(Level.FINER))
{
- return output.output(bytes, offset, size);
+ _logger.log(Level.FINER, SaslImpl.this + " about to call plain input");
}
+
+ transportResult = pourAll(_inputBuffer, _underlyingInput);
}
- };
+ return transportResult;
+ }
+
+ @Override
+ public ByteBuffer getOutputBuffer()
+ {
+ fillOutputBuffer();
+ _outputBuffer.flip();
+
+ _readOnlyOutputBufferView = _outputBuffer.asReadOnlyBuffer();
+ return _readOnlyOutputBufferView;
+ }
+
+ @Override
+ public void outputConsumed()
+ {
+ _outputBuffer.position(_readOnlyOutputBufferView.position());
+ _outputBuffer.compact();
+ }
+
}
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java Tue May 21 15:49:52 2013
@@ -17,12 +17,30 @@
package org.apache.qpid.proton.engine.impl;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.newReadableBuffer;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.pour;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.pourArrayToBuffer;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.pourBufferToArray;
+
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
+
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.UnsignedShort;
+import org.apache.qpid.proton.amqp.transport.Attach;
+import org.apache.qpid.proton.amqp.transport.Begin;
+import org.apache.qpid.proton.amqp.transport.Close;
+import org.apache.qpid.proton.amqp.transport.Detach;
+import org.apache.qpid.proton.amqp.transport.Disposition;
+import org.apache.qpid.proton.amqp.transport.End;
+import org.apache.qpid.proton.amqp.transport.ErrorCondition;
+import org.apache.qpid.proton.amqp.transport.Flow;
+import org.apache.qpid.proton.amqp.transport.FrameBody;
+import org.apache.qpid.proton.amqp.transport.Open;
+import org.apache.qpid.proton.amqp.transport.Role;
+import org.apache.qpid.proton.amqp.transport.Transfer;
import org.apache.qpid.proton.codec.AMQPDefinedTypes;
import org.apache.qpid.proton.codec.CompositeWritableBuffer;
import org.apache.qpid.proton.codec.DecoderImpl;
@@ -36,38 +54,21 @@ import org.apache.qpid.proton.engine.Sas
import org.apache.qpid.proton.engine.Ssl;
import org.apache.qpid.proton.engine.SslDomain;
import org.apache.qpid.proton.engine.SslPeerDetails;
+import org.apache.qpid.proton.engine.Transport;
import org.apache.qpid.proton.engine.TransportException;
+import org.apache.qpid.proton.engine.TransportResult;
+import org.apache.qpid.proton.engine.TransportResultFactory;
import org.apache.qpid.proton.engine.impl.ssl.ProtonSslEngineProvider;
import org.apache.qpid.proton.engine.impl.ssl.SslImpl;
import org.apache.qpid.proton.framing.TransportFrame;
-import org.apache.qpid.proton.amqp.transport.Attach;
-import org.apache.qpid.proton.amqp.transport.Begin;
-import org.apache.qpid.proton.amqp.transport.Close;
-import org.apache.qpid.proton.amqp.transport.Detach;
-import org.apache.qpid.proton.amqp.transport.Disposition;
-import org.apache.qpid.proton.amqp.transport.End;
-import org.apache.qpid.proton.amqp.transport.ErrorCondition;
-import org.apache.qpid.proton.amqp.transport.Flow;
-import org.apache.qpid.proton.amqp.transport.FrameBody;
-import org.apache.qpid.proton.amqp.transport.Open;
-import org.apache.qpid.proton.amqp.transport.Role;
-import org.apache.qpid.proton.amqp.transport.Transfer;
-public class TransportImpl extends EndpointImpl implements ProtonJTransport, FrameBody.FrameBodyHandler<Integer>,FrameTransport
+public class TransportImpl extends EndpointImpl
+ implements ProtonJTransport, FrameBody.FrameBodyHandler<Integer>,
+ FrameHandler, TransportOutputWriter
{
- public static final byte[] HEADER = new byte[8];
+ private static final byte AMQP_FRAME_TYPE = 0;
- static
- {
- HEADER[0] = (byte) 'A';
- HEADER[1] = (byte) 'M';
- HEADER[2] = (byte) 'Q';
- HEADER[3] = (byte) 'P';
- HEADER[4] = 0;
- HEADER[5] = 1;
- HEADER[6] = 0;
- HEADER[7] = 0;
- }
+ private final FrameParser _frameParser;
private ConnectionImpl _connectionEndpoint;
@@ -82,28 +83,25 @@ public class TransportImpl extends Endpo
private TransportOutput _outputProcessor;
private Map<SessionImpl, TransportSession> _transportSessionState = new HashMap<SessionImpl, TransportSession>();
- private Map<LinkImpl, TransportLink> _transportLinkState = new HashMap<LinkImpl, TransportLink>();
+ private Map<LinkImpl, TransportLink<?>> _transportLinkState = new HashMap<LinkImpl, TransportLink<?>>();
private DecoderImpl _decoder = new DecoderImpl();
private EncoderImpl _encoder = new EncoderImpl(_decoder);
- private int _maxFrameSize = 16 * 1024;
+ private int _maxFrameSize = DEFAULT_MAX_FRAME_SIZE;
+
+ private final ByteBuffer _outputOverflowBuffer;
- private final ByteBuffer _overflowBuffer = ByteBuffer.wrap(new byte[_maxFrameSize]);
- private static final byte AMQP_FRAME_TYPE = 0;
private boolean _closeReceived;
private Open _open;
private SaslImpl _sasl;
private SslImpl _ssl;
- private TransportException _inputException;
private ProtocolTracer _protocolTracer = null;
+ private ByteBuffer _lastInputBuffer;
- {
- AMQPDefinedTypes.registerAllTypes(_decoder, _encoder);
- _overflowBuffer.flip();
- }
+ private TransportResult _lastTransportResult = TransportResultFactory.ok();
/**
* @deprecated This constructor's visibility will be reduced to the default scope in a future release.
@@ -111,17 +109,23 @@ public class TransportImpl extends Endpo
*/
@Deprecated public TransportImpl()
{
- FrameParser frameParser = new FrameParser(this);
+ this(DEFAULT_MAX_FRAME_SIZE);
+ }
- _inputProcessor = frameParser;
- _outputProcessor = new TransportOutput()
- {
- @Override
- public int output(byte[] bytes, int offset, int size)
- {
- return transportOutput(bytes, offset, size);
- }
- };
+ /**
+ * Creates a transport with the given maximum frame size.
+ * Note that the maximumFrameSize also determines the size of the output buffer.
+ */
+ TransportImpl(int maxFrameSize)
+ {
+ AMQPDefinedTypes.registerAllTypes(_decoder, _encoder);
+
+ _maxFrameSize = maxFrameSize;
+ _frameParser = new FrameParser(this, _decoder, maxFrameSize);
+
+ _inputProcessor = _frameParser;
+ _outputProcessor = new TransportOutputAdaptor(this, maxFrameSize);
+ _outputOverflowBuffer = newReadableBuffer(maxFrameSize);
}
@Override
@@ -143,33 +147,37 @@ public class TransportImpl extends Endpo
{
_connectionEndpoint.setRemoteState(EndpointState.CLOSED);
}
- _inputProcessor.input(new byte[0],0,0);
+
+ _frameParser.flush();
}
}
@Override
public int input(byte[] bytes, int offset, int length)
{
- if(_inputException != null)
- {
- throw _inputException;
- }
- if(length == 0)
+ oldApiCheckStateBeforeInput(length).checkIsOk();
+
+ ByteBuffer inputBuffer = getInputBuffer();
+ int numberOfBytesConsumed = pourArrayToBuffer(bytes, offset, length, inputBuffer);
+ processInput().checkIsOk();
+ return numberOfBytesConsumed;
+ }
+
+ /**
+ * This method is public as it is used by Python layer.
+ * @see Transport#input(byte[], int, int)
+ */
+ public TransportResult oldApiCheckStateBeforeInput(int inputLength)
+ {
+ _lastTransportResult.checkIsOk();
+ if(inputLength == 0)
{
if(_connectionEndpoint == null || _connectionEndpoint.getRemoteState() != EndpointState.CLOSED)
{
- throw new TransportException("Unexpected EOS: connection aborted");
+ return TransportResultFactory.error(new TransportException("Unexpected EOS when remote connection not closed: connection aborted"));
}
}
- try
- {
- return _inputProcessor.input(bytes, offset, length);
- }
- catch (TransportException e)
- {
- _inputException = e;
- throw e;
- }
+ return TransportResultFactory.ok();
}
//==================================================================================================================
@@ -178,62 +186,52 @@ public class TransportImpl extends Endpo
@Override
public int output(byte[] bytes, final int offset, final int size)
{
- try
- {
- return _outputProcessor.output(bytes, offset, size);
- }
- catch (RuntimeException e)
- {
- e.printStackTrace();
- throw e;
- }
+ ByteBuffer outputBuffer = getOutputBuffer();
+ int numberOfBytesOutput = pourBufferToArray(outputBuffer, bytes, offset, size);
+ outputConsumed();
+ return numberOfBytesOutput;
}
- private int transportOutput(byte[] bytes, int offset, int size)
+ @Override
+ public void writeInto(ByteBuffer outputBuffer)
{
- int written = 0;
-
- if(_overflowBuffer.hasRemaining())
+ if(_outputOverflowBuffer.hasRemaining())
{
- final int overflowWritten = Math.min(size, _overflowBuffer.remaining());
- _overflowBuffer.get(bytes, offset, overflowWritten);
- written+=overflowWritten;
+ pour(_outputOverflowBuffer, outputBuffer);
}
- if(!_overflowBuffer.hasRemaining())
+ if(!_outputOverflowBuffer.hasRemaining())
{
- _overflowBuffer.clear();
+ _outputOverflowBuffer.clear();
- CompositeWritableBuffer outputBuffer =
+ CompositeWritableBuffer compositeOutputBuffer =
new CompositeWritableBuffer(
- new WritableBuffer.ByteBufferWrapper(ByteBuffer.wrap(bytes, offset + written, size - written)),
- new WritableBuffer.ByteBufferWrapper(_overflowBuffer));
+ new WritableBuffer.ByteBufferWrapper(outputBuffer),
+ new WritableBuffer.ByteBufferWrapper(_outputOverflowBuffer));
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processHeader(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processOpen(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processBegin(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processAttach(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processReceiverDisposition(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processReceiverFlow(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processMessageData(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processSenderDisposition(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processSenderFlow(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processDetach(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processEnd(outputBuffer);
- if( outputBuffer.remaining() >= _maxFrameSize ) { written += processClose(outputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processHeader(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processOpen(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processBegin(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processAttach(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processReceiverDisposition(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processReceiverFlow(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processMessageData(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processSenderDisposition(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processSenderFlow(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processDetach(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processEnd(compositeOutputBuffer);
+ if( compositeOutputBuffer.remaining() >= _maxFrameSize ) { processClose(compositeOutputBuffer);
}}}}}}}}}}}}
- _overflowBuffer.flip();
- written -= _overflowBuffer.remaining();
+ _outputOverflowBuffer.flip();
}
-
- return written;
}
+ @Override
public Sasl sasl()
{
if(_sasl == null)
{
- _sasl = new SaslImpl();
+ _sasl = new SaslImpl(_maxFrameSize);
TransportWrapper transportWrapper = _sasl.wrap(_inputProcessor, _outputProcessor);
_inputProcessor = transportWrapper;
_outputProcessor = transportWrapper;
@@ -268,20 +266,8 @@ public class TransportImpl extends Endpo
return ssl(sslDomain, null);
}
- private void clearTransportWorkList()
+ private void processDetach(WritableBuffer buffer)
{
- DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
- while(delivery != null)
- {
- DeliveryImpl transportWorkNext = delivery.getTransportWorkNext();
- delivery.clearTransportWork();
- delivery = transportWorkNext;
- }
- }
-
- private int processDetach(WritableBuffer buffer)
- {
- int written = 0;
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
@@ -291,7 +277,7 @@ public class TransportImpl extends Endpo
if(endpoint instanceof LinkImpl)
{
LinkImpl link = (LinkImpl) endpoint;
- TransportLink transportLink = getTransportState(link);
+ TransportLink<?> transportLink = getTransportState(link);
SessionImpl session = link.getSession();
TransportSession transportSession = getTransportState(session);
@@ -321,8 +307,7 @@ public class TransportImpl extends Endpo
}
- int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), detach, null, null);
- written += frameBytes;
+ writeFrame(buffer, transportSession.getLocalChannel(), detach, null, null);
endpoint.clearModified();
// TODO - temporary hack for PROTON-154, this line should be removed and replaced
@@ -335,12 +320,10 @@ public class TransportImpl extends Endpo
endpoint = endpoint.transportNext();
}
}
- return written;
}
- private int processSenderFlow(WritableBuffer buffer)
+ private void processSenderFlow(WritableBuffer buffer)
{
- int written = 0;
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
@@ -368,8 +351,7 @@ public class TransportImpl extends Endpo
flow.setLinkCredit(transportLink.getLinkCredit());
flow.setDrain(sender.getDrain());
flow.setNextOutgoingId(transportSession.getNextOutgoingId());
- int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
- written += frameBytes;
+ writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
endpoint.clearModified();
}
@@ -378,12 +360,10 @@ public class TransportImpl extends Endpo
endpoint = endpoint.transportNext();
}
}
- return written; //TODO - Implement
}
- private int processSenderDisposition(WritableBuffer buffer)
+ private void processSenderDisposition(WritableBuffer buffer)
{
- int written = 0;
if(_connectionEndpoint != null)
{
DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
@@ -402,10 +382,11 @@ public class TransportImpl extends Endpo
transportDelivery.settled();
}
disposition.setState(delivery.getLocalState());
- int frameBytes = writeFrame(buffer, delivery.getLink().getSession()
- .getTransportSession().getLocalChannel(),
- disposition, null, null);
- written += frameBytes;
+
+ writeFrame(buffer,
+ delivery.getLink().getSession().getTransportSession().getLocalChannel(),
+ disposition, null, null);
+
delivery = delivery.clearTransportWork();
}
else
@@ -414,13 +395,10 @@ public class TransportImpl extends Endpo
}
}
}
- return written;
-
}
- private int processMessageData(WritableBuffer buffer)
+ private void processMessageData(WritableBuffer buffer)
{
- int written = 0;
if(_connectionEndpoint != null)
{
DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
@@ -435,7 +413,7 @@ public class TransportImpl extends Endpo
sender.decrementQueued();
- TransportLink transportLink = sender.getTransportLink();
+ TransportLink<?> transportLink = sender.getTransportLink();
UnsignedInteger deliveryId = transportLink.getDeliveryCount();
TransportDelivery transportDelivery = new TransportDelivery(deliveryId, delivery, transportLink);
@@ -465,13 +443,11 @@ public class TransportImpl extends Endpo
// TODO - large frames
ByteBuffer payload = delivery.getData() == null ? null : ByteBuffer.wrap(delivery.getData(), delivery.getDataOffset(), delivery.getDataLength());
- int frameBytes = writeFrame(buffer,
- sender.getSession().getTransportSession().getLocalChannel(),
- transfer, payload, new PartialTransfer(transfer));
+ writeFrame(buffer,
+ sender.getSession().getTransportSession().getLocalChannel(),
+ transfer, payload, new PartialTransfer(transfer));
sender.getSession().getTransportSession().incrementOutgoingId();
- written += frameBytes;
-
if(payload == null || !payload.hasRemaining())
{
delivery.setData(null);
@@ -492,11 +468,6 @@ public class TransportImpl extends Endpo
delivery.setDataOffset(delivery.getDataOffset()+delivery.getDataLength()-payload.remaining());
delivery.setDataLength(payload.remaining());
}
-
-
-
-
-
}
else
{
@@ -504,21 +475,17 @@ public class TransportImpl extends Endpo
}
}
}
- return written;
}
- private int processReceiverDisposition(WritableBuffer buffer)
+ private void processReceiverDisposition(WritableBuffer buffer)
{
- int written = 0;
if(_connectionEndpoint != null)
{
DeliveryImpl delivery = _connectionEndpoint.getTransportWorkHead();
while(delivery != null && buffer.remaining() >= _maxFrameSize)
{
- boolean remove = false;
if((delivery.getLink() instanceof ReceiverImpl) && delivery.isLocalStateChange())
{
- remove = true;
TransportDelivery transportDelivery = delivery.getTransportDelivery();
Disposition disposition = new Disposition();
disposition.setFirst(transportDelivery.getDeliveryId());
@@ -527,16 +494,14 @@ public class TransportImpl extends Endpo
disposition.setSettled(delivery.isSettled());
disposition.setState(delivery.getLocalState());
- int frameBytes = writeFrame(buffer, delivery.getLink().getSession()
- .getTransportSession().getLocalChannel(),
- disposition, null, null);
- written += frameBytes;
+ writeFrame(buffer,
+ delivery.getLink().getSession().getTransportSession().getLocalChannel(),
+ disposition, null, null);
if(delivery.isSettled())
{
transportDelivery.settled();
}
delivery = delivery.clearTransportWork();
-
}
else
{
@@ -544,23 +509,19 @@ public class TransportImpl extends Endpo
}
}
}
- return written;
}
- private int processReceiverFlow(WritableBuffer buffer)
+ private void processReceiverFlow(WritableBuffer buffer)
{
- int written = 0;
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
while(endpoint != null && buffer.remaining() >= _maxFrameSize)
{
-
if(endpoint instanceof ReceiverImpl)
{
-
ReceiverImpl receiver = (ReceiverImpl) endpoint;
- TransportLink transportLink = getTransportState(receiver);
+ TransportLink<?> transportLink = getTransportState(receiver);
TransportSession transportSession = getTransportState(receiver.getSession());
if(receiver.getLocalState() == EndpointState.ACTIVE)
@@ -579,14 +540,12 @@ public class TransportImpl extends Endpo
flow.setLinkCredit(transportLink.getLinkCredit());
flow.setDrain(receiver.getDrain());
flow.setNextOutgoingId(transportSession.getNextOutgoingId());
- int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
- written += frameBytes;
+ writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
if(receiver.getLocalState() == EndpointState.ACTIVE)
{
endpoint.clearModified();
}
}
-
}
}
endpoint = endpoint.transportNext();
@@ -594,7 +553,6 @@ public class TransportImpl extends Endpo
endpoint = _connectionEndpoint.getTransportHead();
while(endpoint != null && buffer.remaining() >= _maxFrameSize)
{
-
if(endpoint instanceof SessionImpl)
{
@@ -611,20 +569,17 @@ public class TransportImpl extends Endpo
flow.setOutgoingWindow(transportSession.getOutgoingWindowSize());
flow.setNextOutgoingId(transportSession.getNextOutgoingId());
flow.setNextIncomingId(transportSession.getNextIncomingId());
- int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
- written += frameBytes;
+ writeFrame(buffer, transportSession.getLocalChannel(), flow, null, null);
}
}
}
endpoint = endpoint.transportNext();
}
}
- return written;
}
- private int processAttach(WritableBuffer buffer)
+ private void processAttach(WritableBuffer buffer)
{
- int written = 0;
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
@@ -635,7 +590,7 @@ public class TransportImpl extends Endpo
{
LinkImpl link = (LinkImpl) endpoint;
- TransportLink transportLink = getTransportState(link);
+ TransportLink<?> transportLink = getTransportState(link);
if(link.getLocalState() != EndpointState.UNINITIALIZED && !transportLink.attachSent())
{
@@ -683,8 +638,7 @@ public class TransportImpl extends Endpo
attach.setInitialDeliveryCount(UnsignedInteger.ZERO);
}
- int frameBytes = writeFrame(buffer, transportSession.getLocalChannel(), attach, null, null);
- written += frameBytes;
+ writeFrame(buffer, transportSession.getLocalChannel(), attach, null, null);
transportLink.sentAttach();
if(link.getLocalState() == EndpointState.ACTIVE && (link instanceof SenderImpl || !link.hasCredit()))
{
@@ -692,40 +646,22 @@ public class TransportImpl extends Endpo
}
}
}
-
}
-
endpoint = endpoint.transportNext();
}
}
- return written;
- }
-
- private void clearInterestList()
- {
- EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
- while(endpoint != null)
- {
- endpoint.clearModified();
- endpoint = endpoint.transportNext();
- }
}
- private int processHeader(WritableBuffer buffer)
+ private void processHeader(WritableBuffer buffer)
{
if(!_headerWritten)
{
- buffer.put(HEADER, 0, HEADER.length);
+ buffer.put(AmqpHeader.HEADER, 0, AmqpHeader.HEADER.length);
_headerWritten = true;
- return HEADER.length;
- }
- else
- {
- return 0;
}
}
- private int processOpen(WritableBuffer buffer)
+ private void processOpen(WritableBuffer buffer)
{
if(_connectionEndpoint != null && _connectionEndpoint.getLocalState() != EndpointState.UNINITIALIZED && !_isOpenSent)
{
@@ -739,16 +675,13 @@ public class TransportImpl extends Endpo
_isOpenSent = true;
- return writeFrame(buffer, 0, open, null, null);
+ writeFrame(buffer, 0, open, null, null);
}
- return 0;
}
- private int processBegin(WritableBuffer buffer)
+ private void processBegin(WritableBuffer buffer)
{
- int written = 0;
-
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
@@ -772,7 +705,7 @@ public class TransportImpl extends Endpo
begin.setOutgoingWindow(transportSession.getOutgoingWindowSize());
begin.setNextOutgoingId(transportSession.getNextOutgoingId());
- written += writeFrame(buffer, channelId, begin, null, null);
+ writeFrame(buffer, channelId, begin, null, null);
transportSession.sentBegin();
if(session.getLocalState() == EndpointState.ACTIVE)
{
@@ -783,7 +716,6 @@ public class TransportImpl extends Endpo
endpoint = endpoint.transportNext();
}
}
- return written;
}
private TransportSession getTransportState(SessionImpl session)
@@ -798,9 +730,9 @@ public class TransportImpl extends Endpo
return transportSession;
}
- private TransportLink getTransportState(LinkImpl link)
+ private TransportLink<?> getTransportState(LinkImpl link)
{
- TransportLink transportLink = _transportLinkState.get(link);
+ TransportLink<?> transportLink = _transportLinkState.get(link);
if(transportLink == null)
{
transportLink = TransportLink.createTransportLink(link);
@@ -831,9 +763,8 @@ public class TransportImpl extends Endpo
return channel;
}
- private int processEnd(WritableBuffer buffer)
+ private void processEnd(WritableBuffer buffer)
{
- int written = 0;
if(_connectionEndpoint != null)
{
EndpointImpl endpoint = _connectionEndpoint.getTransportHead();
@@ -855,16 +786,13 @@ public class TransportImpl extends Endpo
end.setError(localError);
}
- int frameBytes = writeFrame(buffer, channel, end, null, null);
- written += frameBytes;
+ writeFrame(buffer, channel, end, null, null);
endpoint.clearModified();
}
endpoint = endpoint.transportNext();
-
}
}
- return written;
}
private boolean hasSendableMessages(SessionImpl session)
@@ -891,7 +819,7 @@ public class TransportImpl extends Endpo
return false;
}
- private int processClose(WritableBuffer buffer)
+ private void processClose(WritableBuffer buffer)
{
if(_connectionEndpoint != null && _connectionEndpoint.getLocalState() == EndpointState.CLOSED && !_isCloseSent)
{
@@ -907,14 +835,12 @@ public class TransportImpl extends Endpo
_isCloseSent = true;
- return writeFrame(buffer, 0, close, null, null);
+ writeFrame(buffer, 0, close, null, null);
}
}
- return 0;
-
}
- private int writeFrame(WritableBuffer buffer,
+ private void writeFrame(WritableBuffer buffer,
int channel,
FrameBody frameBody,
ByteBuffer payload,
@@ -943,7 +869,7 @@ public class TransportImpl extends Endpo
{
originalPayload = payload.duplicate();
}
- _protocolTracer.sentFrame(new TransportFrame(channel, (FrameBody) frameBody, Binary.create(originalPayload)));
+ _protocolTracer.sentFrame(new TransportFrame(channel, frameBody, Binary.create(originalPayload)));
}
int payloadSize = Math.min(payload == null ? 0 : payload.remaining(), _maxFrameSize - (buffer.position() - oldPosition));
@@ -962,8 +888,6 @@ public class TransportImpl extends Endpo
buffer.put(AMQP_FRAME_TYPE);
buffer.putShort((short) channel);
buffer.position(limit);
-
- return frameSize;
}
//==================================================================================================================
@@ -997,7 +921,7 @@ public class TransportImpl extends Endpo
_open = open;
}
- if(open.getMaxFrameSize().longValue() > 0 && open.getMaxFrameSize().longValue() < (long) _maxFrameSize)
+ if(open.getMaxFrameSize().longValue() > 0 && open.getMaxFrameSize().longValue() < _maxFrameSize)
{
_maxFrameSize = (int) open.getMaxFrameSize().longValue();
}
@@ -1048,7 +972,7 @@ public class TransportImpl extends Endpo
else
{
SessionImpl session = transportSession.getSession();
- TransportLink transportLink = transportSession.getLinkFromRemoteHandle(attach.getHandle());
+ TransportLink<?> transportLink = transportSession.getLinkFromRemoteHandle(attach.getHandle());
LinkImpl link = null;
if(transportLink != null)
@@ -1144,7 +1068,7 @@ public class TransportImpl extends Endpo
}
else
{
- TransportLink transportLink = transportSession.getLinkFromRemoteHandle(detach.getHandle());
+ TransportLink<?> transportLink = transportSession.getLinkFromRemoteHandle(detach.getHandle());
if(transportLink != null)
{
@@ -1203,21 +1127,75 @@ public class TransportImpl extends Endpo
}
@Override
- public boolean input(TransportFrame frame)
+ public void handleFrame(TransportFrame frame)
{
- if( _protocolTracer!=null )
+ if (!isHandlingFrames())
{
- _protocolTracer.receivedFrame(frame);
+ throw new IllegalStateException("Transport cannot accept frame: " + frame);
}
- if(_connectionEndpoint != null || getRemoteState() == EndpointState.UNINITIALIZED)
+
+ if( _protocolTracer != null )
{
- frame.getBody().invoke(this,frame.getPayload(), frame.getChannel());
- return true;
+ _protocolTracer.receivedFrame(frame);
}
- else
+
+ frame.getBody().invoke(this,frame.getPayload(), frame.getChannel());
+ }
+
+ @Override
+ public boolean isHandlingFrames()
+ {
+ return _connectionEndpoint != null || getRemoteState() == EndpointState.UNINITIALIZED;
+ }
+
+ @Override
+ public ProtocolTracer getProtocolTracer()
+ {
+ return _protocolTracer;
+ }
+
+ @Override
+ public void setProtocolTracer(ProtocolTracer protocolTracer)
+ {
+ this._protocolTracer = protocolTracer;
+ }
+
+ @Override
+ public ByteBuffer getInputBuffer()
+ {
+ _lastTransportResult.checkIsOk();
+ _lastInputBuffer = _inputProcessor.getInputBuffer();
+ return _lastInputBuffer;
+ }
+
+ @Override
+ public TransportResult processInput()
+ {
+ if (_lastInputBuffer == null)
{
- return false;
+ throw new IllegalStateException("Unexpected invocation of processInput(), it is required to invoke getInputBuffer() first");
}
+
+ _lastTransportResult = _inputProcessor.processInput();
+ return _lastTransportResult;
+ }
+
+ @Override
+ public ByteBuffer getOutputBuffer()
+ {
+ return _outputProcessor.getOutputBuffer();
+ }
+
+ @Override
+ public void outputConsumed()
+ {
+ _outputProcessor.outputConsumed();
+ }
+
+ @Override
+ public String toString()
+ {
+ return "TransportImpl [_connectionEndpoint=" + _connectionEndpoint + ", " + super.toString() + "]";
}
private static class PartialTransfer implements Runnable
@@ -1229,22 +1207,10 @@ public class TransportImpl extends Endpo
_transfer = transfer;
}
+ @Override
public void run()
{
_transfer.setMore(true);
}
}
-
- @Override
- public ProtocolTracer getProtocolTracer()
- {
- return _protocolTracer;
- }
-
- @Override
- public void setProtocolTracer(ProtocolTracer protocolTracer)
- {
- this._protocolTracer = protocolTracer;
- }
-
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportInput.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportInput.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportInput.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportInput.java Tue May 21 15:49:52 2013
@@ -20,18 +20,22 @@
*/
package org.apache.qpid.proton.engine.impl;
+import java.nio.ByteBuffer;
+
+import org.apache.qpid.proton.engine.Transport;
+import org.apache.qpid.proton.engine.TransportResult;
+
public interface TransportInput
{
-
/**
- * @param bytes input bytes for consumption
- * @param offset the offset within bytes where input begins
- * @param size the number of bytes available for input
- *
- * @return the number of bytes consumed
+ * @see Transport#getInputBuffer()
*/
- public int input(byte[] bytes, int offset, int size);
+ ByteBuffer getInputBuffer();
+ /**
+ * @see Transport#processInput()
+ */
+ TransportResult processInput();
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutput.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutput.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutput.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutput.java Tue May 21 15:49:52 2013
@@ -20,18 +20,19 @@
*/
package org.apache.qpid.proton.engine.impl;
+import java.nio.ByteBuffer;
+
+import org.apache.qpid.proton.engine.Transport;
public interface TransportOutput
{
-
/**
- * @param bytes input bytes for consumption
- * @param offset the offset within bytes where input begins
- * @param size the number of bytes available for input
- *
- * @return the number of bytes consumed
+ * @see Transport#getOutputBuffer()
*/
- public int output(byte[] bytes, int offset, int size);
-
+ ByteBuffer getOutputBuffer();
+ /**
+ * @see Transport#outputConsumed()
+ */
+ void outputConsumed();
}
Added: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java?rev=1484842&view=auto
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java (added)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputAdaptor.java Tue May 21 15:49:52 2013
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.proton.engine.impl;
+
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.*;
+
+import java.nio.ByteBuffer;
+
+class TransportOutputAdaptor implements TransportOutput
+{
+ private TransportOutputWriter _transportOutputWriter;
+
+ private final ByteBuffer _outputBuffer;
+
+ private ByteBuffer _readOnlyOutputBufferView;
+
+ TransportOutputAdaptor(TransportOutputWriter transportOutputWriter, int maxFrameSize)
+ {
+ _transportOutputWriter = transportOutputWriter;
+ _outputBuffer = newWriteableBuffer(maxFrameSize);
+ }
+
+ @Override
+ public ByteBuffer getOutputBuffer()
+ {
+ _transportOutputWriter.writeInto(_outputBuffer);
+ _outputBuffer.flip();
+
+ _readOnlyOutputBufferView = _outputBuffer.asReadOnlyBuffer();
+ return _readOnlyOutputBufferView;
+ }
+
+ @Override
+ public void outputConsumed()
+ {
+ if(_readOnlyOutputBufferView == null)
+ {
+ throw new IllegalStateException("Illegal invocation with previously calling getOutputBuffer");
+ }
+
+ _outputBuffer.position(_readOnlyOutputBufferView.position());
+ _readOnlyOutputBufferView = null;
+ _outputBuffer.compact();
+ }
+
+}
Copied: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputWriter.java (from r1483522, qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameTransport.java)
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputWriter.java?p2=qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputWriter.java&p1=qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameTransport.java&r1=1483522&r2=1484842&rev=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/FrameTransport.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportOutputWriter.java Tue May 21 15:49:52 2013
@@ -1,5 +1,4 @@
/*
- *
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -16,13 +15,15 @@
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
- *
*/
package org.apache.qpid.proton.engine.impl;
-import org.apache.qpid.proton.framing.TransportFrame;
+import java.nio.ByteBuffer;
-public interface FrameTransport
+interface TransportOutputWriter
{
- boolean input(TransportFrame frame);
+ /**
+ * Writes my pending output bytes into outputBuffer. Does not subsequently flip it.
+ */
+ void writeInto(ByteBuffer outputBuffer);
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/DefaultSslEngineFacade.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/DefaultSslEngineFacade.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/DefaultSslEngineFacade.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/DefaultSslEngineFacade.java Tue May 21 15:49:52 2013
@@ -24,13 +24,23 @@ import java.nio.ByteBuffer;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
+import javax.net.ssl.SSLEngineResult.Status;
import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLSession;
class DefaultSslEngineFacade implements ProtonSslEngine
{
private final SSLEngine _sslEngine;
+ /**
+ * Our testing has shown that application buffers need to be a bit larger
+ * than that provided by {@link SSLSession#getApplicationBufferSize()} otherwise
+ * {@link Status#BUFFER_OVERFLOW} will result on {@link SSLEngine#unwrap()}.
+ * Sun's own example uses 50, so we use the same.
+ */
+ private static final int APPLICATION_BUFFER_EXTRA = 50;
+
DefaultSslEngineFacade(SSLEngine sslEngine)
{
_sslEngine = sslEngine;
@@ -48,8 +58,16 @@ class DefaultSslEngineFacade implements
return _sslEngine.unwrap(src, dst);
}
+ /**
+ * @see #APPLICATION_BUFFER_EXTRA
+ */
@Override
- public int getApplicationBufferSize()
+ public int getEffectiveApplicationBufferSize()
+ {
+ return getApplicationBufferSize() + APPLICATION_BUFFER_EXTRA;
+ }
+
+ private int getApplicationBufferSize()
{
return _sslEngine.getSession().getApplicationBufferSize();
}
@@ -89,4 +107,13 @@ class DefaultSslEngineFacade implements
{
return _sslEngine.getUseClientMode();
}
+
+ @Override
+ public String toString()
+ {
+ StringBuilder builder = new StringBuilder();
+ builder.append("DefaultSslEngineFacade [_sslEngine=").append(_sslEngine).append("]");
+ return builder.toString();
+ }
+
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/ProtonSslEngine.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/ProtonSslEngine.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/ProtonSslEngine.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/ProtonSslEngine.java Tue May 21 15:49:52 2013
@@ -56,9 +56,14 @@ public interface ProtonSslEngine
Runnable getDelegatedTask();
HandshakeStatus getHandshakeStatus();
- int getApplicationBufferSize();
+ /**
+ * Gets the application buffer size.
+ */
+ int getEffectiveApplicationBufferSize();
+
int getPacketBufferSize();
String getCipherSuite();
String getProtocol();
boolean getUseClientMode();
+
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SimpleSslTransportWrapper.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SimpleSslTransportWrapper.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SimpleSslTransportWrapper.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SimpleSslTransportWrapper.java Tue May 21 15:49:52 2013
@@ -20,53 +20,51 @@
*/
package org.apache.qpid.proton.engine.impl.ssl;
+
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.newReadableBuffer;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.newWriteableBuffer;
+import static org.apache.qpid.proton.engine.impl.ByteBufferUtils.pourAll;
+
import java.nio.ByteBuffer;
import java.util.logging.Level;
import java.util.logging.Logger;
-import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
import javax.net.ssl.SSLEngineResult.Status;
import javax.net.ssl.SSLException;
-import javax.net.ssl.SSLSession;
import org.apache.qpid.proton.engine.TransportException;
+import org.apache.qpid.proton.engine.TransportResult;
+import org.apache.qpid.proton.engine.TransportResultFactory;
import org.apache.qpid.proton.engine.impl.TransportInput;
import org.apache.qpid.proton.engine.impl.TransportOutput;
/**
- * TODO close the SSLEngine when told to, and modify {@link #input(byte[], int, int)} and {@link #output(byte[], int, int)}
+ * TODO close the SSLEngine when told to, and modify {@link #wrapOutput()} and {@link #unwrapInput()}
* to respond appropriately thereafter.
*/
public class SimpleSslTransportWrapper implements SslTransportWrapper
{
private static final Logger _logger = Logger.getLogger(SimpleSslTransportWrapper.class.getName());
- /**
- * Our testing has shown that application buffers need to be a bit larger
- * than that provided by {@link SSLSession#getApplicationBufferSize()} otherwise
- * {@link Status#BUFFER_OVERFLOW} will result on {@link SSLEngine#unwrap()}.
- * Sun's own example uses 50, so we use the same.
- */
- private static final int APPLICATION_BUFFER_EXTRA = 50;
+ private final ProtonSslEngine _sslEngine;
private final TransportInput _underlyingInput;
private final TransportOutput _underlyingOutput;
- private ProtonSslEngine _sslEngine;
+ private final ByteBuffer _inputBuffer;
- /** Used by {@link #output(byte[], int, int)}. Acts as a buffer for the output from underlyingOutput */
- private ByteHolder _clearOutputHolder;
+ private final ByteBuffer _outputBuffer;
- /** Used by {@link #output(byte[], int, int)}. Caches the output from SSLEngine.wrap */
- private ByteHolder _encodedOutputHolder;
-
- /** Used by {@link #input(byte[], int, int)}. A buffer for the decoded bytes that will be passed to _underlyingInput */
- private ByteHolder _decodedInputHolder;
+ /**
+ * A buffer for the decoded bytes that will be passed to _underlyingInput.
+ * This extra layer of buffering is necessary in case the underlying input's buffer
+ * is too small for SSLEngine to ever unwrap into.
+ */
+ private final ByteBuffer _decodedInputBuffer;
- /** used by {@link #input(byte[], int, int)} to cache the left over bytes not yet decoded due to buffer underflow. */
- private final BytePipeline _encodedLeftoversPipeline = new BytePipeline();
+ private ByteBuffer _readOnlyOutputBufferView;
/** could change during the lifetime of the ssl connection owing to renegotiation. */
private String _cipherName;
@@ -74,48 +72,56 @@ public class SimpleSslTransportWrapper i
/** could change during the lifetime of the ssl connection owing to renegotiation. */
private String _protocolName;
+
+ /**
+ * We store the last input result so that we know not to process any more input
+ * if it failed.
+ */
+ private TransportResult _lastInputResult = TransportResultFactory.ok();
+
+
SimpleSslTransportWrapper(ProtonSslEngine sslEngine, TransportInput underlyingInput, TransportOutput underlyingOutput)
{
_underlyingInput = underlyingInput;
_underlyingOutput = underlyingOutput;
_sslEngine = sslEngine;
- createByteHolders();
- }
- private void createByteHolders()
- {
- int appBufferMax = _sslEngine.getApplicationBufferSize();
- int packetBufferMax = _sslEngine.getPacketBufferSize();
-
- _clearOutputHolder = new ByteHolder(appBufferMax + APPLICATION_BUFFER_EXTRA);
- _decodedInputHolder = new ByteHolder(appBufferMax + APPLICATION_BUFFER_EXTRA);
- _decodedInputHolder.prepareToRead();
+ int effectiveAppBufferMax = _sslEngine.getEffectiveApplicationBufferSize();
+ int packetSize = _sslEngine.getPacketBufferSize();
- _encodedOutputHolder = new ByteHolder(packetBufferMax);
- _encodedOutputHolder.prepareToRead();
- }
+ // Input and output buffers need to be large enough to contain one SSL packet,
+ // as stated in SSLEngine JavaDoc.
+ _inputBuffer = newWriteableBuffer(packetSize);
+ _outputBuffer = newWriteableBuffer(packetSize);
- @Override
- public int input(byte[] encodedBytes, int offset, int size)
- {
- final int initialSizeOfEncodedLeftovers = _encodedLeftoversPipeline.getSize();
- final ByteBuffer oldPlusNewEncodedByteBuffer = _encodedLeftoversPipeline.appendAndClear(encodedBytes, offset, size);
+ _decodedInputBuffer = newReadableBuffer(effectiveAppBufferMax);
- if(!_decodedInputHolder.readInto(_underlyingInput))
+ if(_logger.isLoggable(Level.FINE))
{
- // underlying input couldn't accept all existing leftovers so we return without trying to decode any new data
- return 0;
+ _logger.fine("Constructed " + this);
}
+ }
+
+ /**
+ * Unwraps the contents of {@link #_inputBuffer} and passes it to {@link #_underlyingInput}.
+ *
+ * Regarding the state of {@link #_inputBuffer}:
+ * - On entry, it is assumed to be readable.
+ * - On exit, it is still readable and its "remaining" bytes are those that we were unable
+ * to unwrap (e.g. if they don't form a whole packet).
+ */
+ private TransportResult unwrapInput()
+ {
try
{
+ TransportResult underlyingResult = TransportResultFactory.ok();
boolean keepLooping = true;
- int totalBytesDecoded = 0;
-
do
{
- SSLEngineResult result = _sslEngine.unwrap(oldPlusNewEncodedByteBuffer, _decodedInputHolder.prepareToWrite());
- _decodedInputHolder.prepareToRead();
+ _decodedInputBuffer.compact();
+ SSLEngineResult result = _sslEngine.unwrap(_inputBuffer, _decodedInputBuffer);
+ _decodedInputBuffer.flip();
runDelegatedTasks(result);
updateCipherAndProtocolName(result);
@@ -127,20 +133,18 @@ public class SimpleSslTransportWrapper i
if(sslResultStatus == SSLEngineResult.Status.OK)
{
- totalBytesDecoded += result.bytesConsumed();
+ // continue
}
else if(sslResultStatus == SSLEngineResult.Status.BUFFER_UNDERFLOW)
{
- // Not an error. We store the not-yet-decoded bytes which will hopefully be augmented by some more
+ // Not an error. The not-yet-decoded bytes remain in _inputBuffer and will hopefully be augmented by some more
// in a subsequent invocation of this method, allowing decoding to be done.
- _encodedLeftoversPipeline.set(oldPlusNewEncodedByteBuffer, totalBytesDecoded);
}
else
{
throw new IllegalStateException("Unexpected SSL Engine state " + sslResultStatus);
}
- boolean allAcceptedByUnderlyingInput = true;
if(result.bytesProduced() > 0)
{
if(handshakeStatus != HandshakeStatus.NOT_HANDSHAKING)
@@ -148,108 +152,82 @@ public class SimpleSslTransportWrapper i
_logger.warning("WARN unexpectedly produced bytes for the underlying input when handshaking");
}
- allAcceptedByUnderlyingInput = _decodedInputHolder.readInto(_underlyingInput);
+ underlyingResult = pourAll(_decodedInputBuffer, _underlyingInput);
}
keepLooping = handshakeStatus == HandshakeStatus.NOT_HANDSHAKING
&& sslResultStatus != Status.BUFFER_UNDERFLOW
- && allAcceptedByUnderlyingInput;
+ && underlyingResult.isOk();
}
while(keepLooping);
- int newEncodedLeftovers = _encodedLeftoversPipeline.getSize();
- int sizeToReportConsumed = totalBytesDecoded + newEncodedLeftovers - initialSizeOfEncodedLeftovers;
- return sizeToReportConsumed;
+ return underlyingResult;
}
catch(SSLException e)
{
- throw new TransportException("Problem during input. useClientMode: " + _sslEngine.getUseClientMode(), e);
+ return TransportResultFactory.error(new TransportException(
+ "Problem during input. useClientMode: " + _sslEngine.getUseClientMode(), e));
}
}
/**
- * Write encoded output to the supplied destination.
+ * Wrap the underlying transport's output, passing it to the output buffer.
*
- * The following conditions hold before and after this method call:
- * {@link #_encodedOutputHolder} is readable.
- * {@link #_clearOutputHolder} is writeable.
+ * {@link #_outputBuffer} is assumed to be writeable on entry and is guaranteed to
+ * be still writeable on exit.
*/
- @Override
- public int output(byte[] destination, int offset, final int size)
+ private void wrapOutput()
{
- try
+ boolean keepWrapping = hasSpaceForSslPacket(_outputBuffer);
+ while(keepWrapping)
{
- int totalBytesWrittenToDestination = _encodedOutputHolder.readInto(destination, offset, size);
- if(totalBytesWrittenToDestination == size)
- {
- return size;
- }
-
- boolean keepLooping = true;
- do
+ ByteBuffer clearOutputBuffer = _underlyingOutput.getOutputBuffer();
+ try
{
- if (_clearOutputHolder.hasSpace())
+ if(clearOutputBuffer.hasRemaining())
{
- int numberOfBytesInClearOutputHolder = _clearOutputHolder.writeOutputFrom(_underlyingOutput);
- if (numberOfBytesInClearOutputHolder == 0)
- {
- break; // no output to wrap
- }
- }
+ SSLEngineResult result = _sslEngine.wrap(clearOutputBuffer, _outputBuffer);
- int availableDestinationSize = size - totalBytesWrittenToDestination;
- // SSLEngine will wrap directly into the provided destination byte[] if there is space, or into _encodedOutputHolder otherwise.
- boolean wrappingIntoEncodedOutputHolder = availableDestinationSize < _sslEngine.getPacketBufferSize();
- final ByteBuffer sslWrapDst;
- if (wrappingIntoEncodedOutputHolder)
- {
- sslWrapDst = _encodedOutputHolder.prepareToWrite();
- }
- else
- {
- sslWrapDst = ByteBuffer.wrap(destination, offset+totalBytesWrittenToDestination, availableDestinationSize);
- }
+ logEngineClientModeAndResult(result, "output");
- SSLEngineResult result = _sslEngine.wrap(_clearOutputHolder.prepareToRead(), sslWrapDst);
- logEngineClientModeAndResult(result, "output");
+ Status sslResultStatus = result.getStatus();
+ if(sslResultStatus == SSLEngineResult.Status.BUFFER_OVERFLOW)
+ {
+ throw new IllegalStateException("Insufficient space to perform wrap into encoded output buffer. Buffer: " + _outputBuffer);
+ }
+ else if(sslResultStatus != SSLEngineResult.Status.OK)
+ {
+ throw new RuntimeException("Unexpected SSLEngineResult status " + sslResultStatus);
+ }
- _clearOutputHolder.prepareToWrite();
+ runDelegatedTasks(result);
+ updateCipherAndProtocolName(result);
- Status sslResultStatus = result.getStatus();
- if(sslResultStatus == SSLEngineResult.Status.BUFFER_OVERFLOW)
- {
- throw new IllegalStateException("Insufficient space to perform wrap into encoded output buffer that has " + sslWrapDst.remaining() + " remaining bytes. wrappingIntoEncodedOutputHolder=" + wrappingIntoEncodedOutputHolder);
- }
- else if(sslResultStatus != SSLEngineResult.Status.OK)
- {
- throw new RuntimeException("Unexpected SSLEngineResult status " + sslResultStatus);
+ keepWrapping = result.getHandshakeStatus() == HandshakeStatus.NOT_HANDSHAKING
+ && hasSpaceForSslPacket(_outputBuffer);
}
-
- runDelegatedTasks(result);
- updateCipherAndProtocolName(result);
-
- final int numberOfNewlyEncodedBytesWritten = Math.min(availableDestinationSize, result.bytesProduced());
- if(wrappingIntoEncodedOutputHolder)
+ else
{
- _encodedOutputHolder.prepareToRead();
- _encodedOutputHolder.readInto(destination, offset+totalBytesWrittenToDestination, numberOfNewlyEncodedBytesWritten);
+ // no output to wrap
+ keepWrapping = false;
}
-
- totalBytesWrittenToDestination += numberOfNewlyEncodedBytesWritten;
-
- keepLooping = result.getHandshakeStatus() == HandshakeStatus.NOT_HANDSHAKING
- && totalBytesWrittenToDestination < size;
}
- while(keepLooping);
-
- return totalBytesWrittenToDestination;
- }
- catch(SSLException e)
- {
- throw new TransportException("Problem during output. useClientMode: " + _sslEngine.getUseClientMode(), e);
+ catch(SSLException e)
+ {
+ throw new TransportException("Problem during output. useClientMode: " + _sslEngine.getUseClientMode(), e);
+ }
+ finally
+ {
+ _underlyingOutput.outputConsumed();
+ }
}
}
+ private boolean hasSpaceForSslPacket(ByteBuffer byteBuffer)
+ {
+ return byteBuffer.remaining() >= _sslEngine.getPacketBufferSize();
+ }
+
/** @return the cipher name, which is null until the SSL handshaking is completed */
@Override
public String getCipherName()
@@ -308,4 +286,69 @@ public class SimpleSslTransportWrapper i
.append(" bytesProduced = ").append(result.bytesProduced())
.append("]").toString();
}
+
+ @Override
+ public ByteBuffer getInputBuffer()
+ {
+ _lastInputResult.checkIsOk();
+ return _inputBuffer;
+ }
+
+ @Override
+ public TransportResult processInput()
+ {
+ _inputBuffer.flip();
+
+ try
+ {
+ _lastInputResult = unwrapInput();
+ return _lastInputResult;
+ }
+ finally
+ {
+ _inputBuffer.compact();
+ }
+ }
+
+ /*
+ * Pre-condition of {@link #_outputBuffer}: it's writeable
+ * Post-condition of {@link #_outputBuffer}: it's readable
+ */
+ @Override
+ public ByteBuffer getOutputBuffer()
+ {
+ wrapOutput();
+ _outputBuffer.flip();
+
+ _readOnlyOutputBufferView = _outputBuffer.asReadOnlyBuffer();
+ return _readOnlyOutputBufferView;
+ }
+
+ @Override
+ public void outputConsumed()
+ {
+ if(_readOnlyOutputBufferView == null)
+ {
+ throw new IllegalStateException("Illegal invocation with previously calling getOutputBuffer");
+ }
+
+ _outputBuffer.position(_readOnlyOutputBufferView.position());
+ _readOnlyOutputBufferView = null;
+ _outputBuffer.compact();
+ }
+
+
+ @Override
+ public String toString()
+ {
+ StringBuilder builder = new StringBuilder();
+ builder.append("SimpleSslTransportWrapper [sslEngine=").append(_sslEngine)
+ .append(", inputBuffer=").append(_inputBuffer)
+ .append(", outputBuffer=").append(_outputBuffer)
+ .append(", decodedInputBuffer=").append(_decodedInputBuffer)
+ .append(", cipherName=").append(_cipherName)
+ .append(", protocolName=").append(_protocolName)
+ .append("]");
+ return builder.toString();
+ }
}
Modified: qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SslEngineFacadeFactory.java
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SslEngineFacadeFactory.java?rev=1484842&r1=1484841&r2=1484842&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SslEngineFacadeFactory.java (original)
+++ qpid/proton/trunk/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/ssl/SslEngineFacadeFactory.java Tue May 21 15:49:52 2013
@@ -350,10 +350,12 @@ public class SslEngineFacadeFactory
}
Reader reader = null;
+ PEMReader pemReader = null;
+
try
{
reader = new FileReader(pemFile);
- PEMReader pemReader = new PEMReader(reader, passwordFinder);
+ pemReader = new PEMReader(reader, passwordFinder);
Object pemObject = pemReader.readObject();
if (!checkPemObjectIsOfAllowedTypes(pemObject, expectedInterfaces))
{
@@ -374,6 +376,17 @@ public class SslEngineFacadeFactory
}
finally
{
+ if(pemReader != null)
+ {
+ try
+ {
+ pemReader.close();
+ }
+ catch(IOException e)
+ {
+ _logger.log(Level.SEVERE, "Couldn't close PEM reader", e);
+ }
+ }
if (reader != null)
{
try
@@ -382,7 +395,7 @@ public class SslEngineFacadeFactory
}
catch (IOException e)
{
- // Ignore
+ _logger.log(Level.SEVERE, "Couldn't close PEM file reader", e);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org