Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use non-deprecated output methods for proton to allow for faster bulk 

sends of outbound amqp frames.
This commit is contained in:
Timothy Bish 2014-02-04 12:47:42 -05:00
parent 9eb7fb9062
commit e102e64e9d
2 changed files with 27 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
View File

@ -18,6 +18,7 @@ package org.apache.activemq.transport.amqp;
import java.io.IOException; import java.io.IOException;
import java.io.UnsupportedEncodingException; import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.EnumSet; import java.util.EnumSet;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
@ -159,16 +160,13 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
void pumpProtonToSocket() { void pumpProtonToSocket() {
try { try {
int size = 1024 * 64;
byte data[] = new byte[size];
boolean done = false; boolean done = false;
while (!done) { while (!done) {
int count = protonTransport.output(data, 0, size); ByteBuffer toWrite = protonTransport.getOutputBuffer();
if (count > 0) { if (toWrite != null && toWrite.hasRemaining()) {
final Buffer buffer; // // System.out.println("writing: " + buffer.toString().substring(5).replaceAll("(..)", "$1 "));
buffer = new Buffer(data, 0, count); amqpTransport.sendToAmqp(toWrite);
// System.out.println("writing: " + buffer.toString().substring(5).replaceAll("(..)", "$1 ")); protonTransport.outputConsumed();
amqpTransport.sendToAmqp(buffer);
} else { } else {
done = true; done = true;
} }
@ -248,10 +246,12 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
sasl.done(Sasl.SaslOutcome.PN_SASL_OK); sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
amqpTransport.getWireFormat().magicRead = false; amqpTransport.getWireFormat().magicRead = false;
sasl = null; sasl = null;
LOG.debug("SASL [PLAIN] Handshake complete.");
} else if ("ANONYMOUS".equals(sasl.getRemoteMechanisms()[0])) { } else if ("ANONYMOUS".equals(sasl.getRemoteMechanisms()[0])) {
sasl.done(Sasl.SaslOutcome.PN_SASL_OK); sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
amqpTransport.getWireFormat().magicRead = false; amqpTransport.getWireFormat().magicRead = false;
sasl = null; sasl = null;
LOG.debug("SASL [ANONYMOUS] Handshake complete.");
} }
} }
} }

21
activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java
View File

@ -21,6 +21,10 @@ import java.io.DataInputStream;
import java.io.DataOutput; import java.io.DataOutput;
import java.io.DataOutputStream; import java.io.DataOutputStream;
import java.io.IOException; import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import org.apache.activemq.util.ByteArrayInputStream; import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream; import org.apache.activemq.util.ByteArrayOutputStream;
@ -53,8 +57,21 @@ public class AmqpWireFormat implements WireFormat {
@Override @Override
public void marshal(Object command, DataOutput dataOut) throws IOException { public void marshal(Object command, DataOutput dataOut) throws IOException {
Buffer frame = (Buffer) command; if (command instanceof ByteBuffer) {
frame.writeTo(dataOut); ByteBuffer buffer = (ByteBuffer) command;
if (dataOut instanceof OutputStream) {
WritableByteChannel channel = Channels.newChannel((OutputStream) dataOut);
channel.write(buffer);
} else {
while (buffer.hasRemaining()) {
dataOut.writeByte(buffer.get());
}
}
} else {
Buffer frame = (Buffer) command;
frame.writeTo(dataOut);
}
} }
boolean magicRead = false; boolean magicRead = false;