You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ai...@apache.org on 2008/04/18 18:09:23 UTC
svn commit: r649585 [5/5] - in /incubator/qpid/branches/thegreatmerge: ./
qpid/cpp/ qpid/cpp/etc/ qpid/cpp/rubygen/ qpid/cpp/rubygen/0-10/
qpid/cpp/src/ qpid/cpp/src/qpid/ qpid/cpp/src/qpid/amqp_0_10/
qpid/cpp/src/qpid/broker/ qpid/cpp/src/qpid/client/...
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ConnectionDelegate.java Fri Apr 18 09:09:03 2008
@@ -26,7 +26,10 @@
import org.apache.qpidity.QpidException;
import java.io.UnsupportedEncodingException;
+
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
@@ -79,17 +82,27 @@
public void init(Channel ch, ProtocolHeader hdr)
{
- // XXX: hardcoded version
- if (hdr.getMajor() != 0 && hdr.getMinor() != 10)
+ ch.getConnection().send(new ConnectionEvent(0, new ProtocolHeader
+ (1,
+ TransportConstants.getVersionMajor(),
+ TransportConstants.getVersionMinor())));
+ if (hdr.getMajor() != TransportConstants.getVersionMajor() &&
+ hdr.getMinor() != TransportConstants.getVersionMinor())
{
// XXX
- ch.getConnection().send(new ConnectionEvent(0, new ProtocolHeader(1, TransportConstants.getVersionMajor(),
- TransportConstants.getVersionMinor())));
+ ch.getConnection().send(new ConnectionEvent(0, new ProtocolHeader
+ (1,
+ TransportConstants.getVersionMajor(),
+ TransportConstants.getVersionMinor())));
ch.getConnection().close();
}
else
{
- ch.connectionStart(hdr.getMajor(), hdr.getMinor(), null, "PLAIN", "utf8");
+ List<Object> plain = new ArrayList<Object>();
+ plain.add("PLAIN");
+ List<Object> utf8 = new ArrayList<Object>();
+ utf8.add("utf8");
+ ch.connectionStart(null, plain, utf8);
}
}
@@ -99,13 +112,13 @@
@Override public void connectionStart(Channel context, ConnectionStart struct)
{
String mechanism = null;
- String response = null;
+ byte[] response = null;
try
{
mechanism = SecurityHelper.chooseMechanism(struct.getMechanisms());
saslClient = Sasl.createSaslClient(new String[]{ mechanism },null, "AMQP", "localhost", null,
SecurityHelper.createCallbackHandler(mechanism,_username,_password ));
- response = new String(saslClient.evaluateChallenge(new byte[0]),_locale);
+ response = saslClient.evaluateChallenge(new byte[0]);
}
catch (UnsupportedEncodingException e)
{
@@ -128,13 +141,9 @@
{
try
{
- String response = new String(saslClient.evaluateChallenge(struct.getChallenge().getBytes()),_locale);
+ byte[] response = saslClient.evaluateChallenge(struct.getChallenge());
context.connectionSecureOk(response);
}
- catch (UnsupportedEncodingException e)
- {
- // need error handling
- }
catch (SaslException e)
{
// need error handling
@@ -144,14 +153,14 @@
@Override public void connectionTune(Channel context, ConnectionTune struct)
{
// should update the channel max given by the broker.
- context.connectionTuneOk(struct.getChannelMax(), struct.getFrameMax(), struct.getHeartbeat());
+ context.connectionTuneOk(struct.getChannelMax(), struct.getMaxFrameSize(), struct.getHeartbeatMax());
context.connectionOpen(_virtualHost, null, Option.INSIST);
}
@Override public void connectionOpenOk(Channel context, ConnectionOpenOk struct)
{
- String knownHosts = struct.getKnownHosts();
+ List<Object> knownHosts = struct.getKnownHosts();
if(_negotiationCompleteLock != null)
{
_negotiationCompleteLock.lock();
@@ -187,13 +196,13 @@
byte[] challenge = null;
if ( challenge == null)
{
- context.connectionTune(Integer.MAX_VALUE,maxFrame, 0);
+ context.connectionTune(Integer.MAX_VALUE, maxFrame, 0, Integer.MAX_VALUE);
}
else
{
try
{
- context.connectionSecure(new String(challenge,_locale));
+ context.connectionSecure(challenge);
}
catch(Exception e)
{
@@ -218,16 +227,16 @@
try
{
saslServer = Sasl.createSaslServer(_mechanism, "AMQP", "ABC",new HashMap(),SecurityHelper.createCallbackHandler(_mechanism,_username,_password));
- byte[] challenge = saslServer.evaluateResponse(struct.getResponse().getBytes());
+ byte[] challenge = saslServer.evaluateResponse(struct.getResponse());
if ( challenge == null)
{
- context.connectionTune(Integer.MAX_VALUE,maxFrame, 0);
+ context.connectionTune(Integer.MAX_VALUE, maxFrame, 0, Integer.MAX_VALUE);
}
else
{
try
{
- context.connectionSecure(new String(challenge,_locale));
+ context.connectionSecure(challenge);
}
catch(Exception e)
{
@@ -250,8 +259,9 @@
@Override public void connectionOpen(Channel context, ConnectionOpen struct)
{
- String hosts = "amqp:1223243232325";
- context.connectionOpenOk(hosts);
+ List<Object> hosts = new ArrayList<Object>();
+ hosts.add("amqp:1223243232325");
+ context.connectionOpenOk(hosts);
}
public String getPassword()
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Data.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Data.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Data.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Data.java Fri Apr 18 09:09:03 2008
@@ -93,7 +93,7 @@
{
str.append(" | ");
}
- str.append(str(buf, 20));
+ str.append(str(buf));
}
str.append(")");
return str.toString();
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Method.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Method.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Method.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Method.java Fri Apr 18 09:09:03 2008
@@ -20,6 +20,7 @@
*/
package org.apache.qpidity.transport;
+import org.apache.qpidity.transport.network.Frame;
/**
* Method
@@ -34,11 +35,12 @@
{
// XXX: should generate separate factories for separate
// namespaces
- return (Method) Struct.create(type);
+ return (Method) StructFactory.createInstruction(type);
}
// XXX: command subclass?
private long id;
+ private boolean sync = false;
public final long getId()
{
@@ -50,6 +52,16 @@
this.id = id;
}
+ public final boolean isSync()
+ {
+ return sync;
+ }
+
+ void setSync(boolean value)
+ {
+ this.sync = value;
+ }
+
public abstract boolean hasPayload();
public abstract byte getEncodedTrack();
@@ -58,7 +70,14 @@
public <C> void delegate(C context, ProtocolDelegate<C> delegate)
{
- delegate.method(context, this);
+ if (getEncodedTrack() == Frame.L4)
+ {
+ delegate.command(context, this);
+ }
+ else
+ {
+ delegate.control(context, this);
+ }
}
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ProtocolDelegate.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ProtocolDelegate.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ProtocolDelegate.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/ProtocolDelegate.java Fri Apr 18 09:09:03 2008
@@ -31,7 +31,9 @@
void init(C context, ProtocolHeader header);
- void method(C context, Method method);
+ void control(C context, Method control);
+
+ void command(C context, Method command);
void header(C context, Header header);
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/RangeSet.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/RangeSet.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/RangeSet.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/RangeSet.java Fri Apr 18 09:09:03 2008
@@ -98,6 +98,13 @@
ranges.clear();
}
+ public RangeSet copy()
+ {
+ RangeSet copy = new RangeSet();
+ copy.ranges.addAll(ranges);
+ return copy;
+ }
+
public String toString()
{
StringBuffer str = new StringBuffer();
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Session.java Fri Apr 18 09:09:03 2008
@@ -32,6 +32,7 @@
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.apache.qpidity.transport.Option.*;
/**
* Session
@@ -56,23 +57,34 @@
private static boolean ENABLE_REPLAY = false;
private static final Logger log = Logger.get(Session.class);
+ private byte[] name;
+ private long timeout = 60000;
+
// channel may be null
Channel channel;
// incoming command count
- private long commandsIn = 0;
+ long commandsIn = 0;
// completed incoming commands
private final RangeSet processed = new RangeSet();
- private long processedMark = -1;
private Range syncPoint = null;
// outgoing command count
private long commandsOut = 0;
private Map<Long,Method> commands = new HashMap<Long,Method>();
- private long mark = 0;
+ private long maxComplete = -1;
private AtomicBoolean closed = new AtomicBoolean(false);
+ public Session(byte[] name)
+ {
+ this.name = name;
+ }
+
+ public byte[] getName()
+ {
+ return name;
+ }
public Map<Long,Method> getOutstandingCommands()
{
@@ -133,25 +145,12 @@
public void flushProcessed()
{
- boolean first = true;
- RangeSet rest = new RangeSet();
+ RangeSet copy;
synchronized (processed)
{
- for (Range r: processed)
- {
- if (first && r.includes(processedMark))
- {
- processedMark = r.getUpper();
- }
- else
- {
- rest.add(r);
- }
-
- first = false;
- }
+ copy = processed.copy();
}
- executionComplete(processedMark, rest);
+ sessionCompleted(copy);
}
void syncPoint()
@@ -193,34 +192,34 @@
log.debug("%s complete(%d, %d)", this, lower, upper);
synchronized (commands)
{
- for (long id = lower; id <= upper; id++)
+ for (long id = maxComplete; id <= upper; id++)
{
commands.remove(id);
}
+ if (lower <= maxComplete + 1)
+ {
+ maxComplete = Math.max(maxComplete, upper);
+ }
commands.notifyAll();
log.debug("%s commands remaining: %s", this, commands);
}
}
- void complete(long mark)
- {
- synchronized (commands)
- {
- complete(this.mark, mark);
- this.mark = mark;
- commands.notifyAll();
- }
- }
-
protected void invoke(Method m)
{
if (m.getEncodedTrack() == Frame.L4)
{
synchronized (commands)
{
- // You only need to keep the command if you need command level replay.
- // If not we only need to keep track of commands to make sync work
- commands.put(commandsOut++,(ENABLE_REPLAY?m:null));
+ long next = commandsOut++;
+ if (next == 0)
+ {
+ sessionCommandPoint(0, 0);
+ }
+ if (ENABLE_REPLAY)
+ {
+ commands.put(next, m);
+ }
channel.method(m);
}
}
@@ -230,7 +229,7 @@
}
}
- public void header(Header header)
+ public void header(Header header)
{
channel.header(header);
}
@@ -269,21 +268,32 @@
public void sync()
{
+ sync(timeout);
+ }
+
+ public void sync(long timeout)
+ {
log.debug("%s sync()", this);
synchronized (commands)
{
long point = commandsOut - 1;
- if (mark < point)
+ if (maxComplete < point)
{
- executionSync();
+ ExecutionSync sync = new ExecutionSync();
+ sync.setSync(true);
+ invoke(sync);
}
- while (!closed.get() && mark < point)
+ long start = System.currentTimeMillis();
+ long elapsed = 0;
+ while (!closed.get() && elapsed < timeout && maxComplete < point)
{
try {
- log.debug("%s waiting for[%d]: %s", this, point, commands);
- commands.wait();
+ log.debug("%s waiting for[%d]: %d, %s", this, point,
+ maxComplete, commands);
+ commands.wait(timeout - elapsed);
+ elapsed = System.currentTimeMillis() - start;
}
catch (InterruptedException e)
{
@@ -291,9 +301,16 @@
}
}
- if (mark < point)
+ if (maxComplete < point)
{
- throw new RuntimeException("session closed");
+ if (closed.get())
+ {
+ throw new RuntimeException("session closed");
+ }
+ else
+ {
+ throw new RuntimeException("timed out waiting for sync");
+ }
}
}
}
@@ -346,16 +363,19 @@
}
}
- public T get(long timeout, int nanos)
+ public T get(long timeout)
{
synchronized (this)
{
- while (!closed.get() && !isDone())
+ long start = System.currentTimeMillis();
+ long elapsed = 0;
+ while (!closed.get() && timeout - elapsed > 0 && !isDone())
{
try
{
log.debug("%s waiting for result: %s", Session.this, this);
- wait(timeout, nanos);
+ wait(timeout - elapsed);
+ elapsed = System.currentTimeMillis() - start;
}
catch (InterruptedException e)
{
@@ -364,22 +384,23 @@
}
}
- if (!isDone())
+ if (isDone())
+ {
+ return result;
+ }
+ else if (closed.get())
{
throw new RuntimeException("session closed");
}
-
- return result;
- }
-
- public T get(long timeout)
- {
- return get(timeout, 0);
+ else
+ {
+ return null;
+ }
}
public T get()
{
- return get(0);
+ return get(timeout);
}
public boolean isDone()
@@ -396,7 +417,8 @@
public void close()
{
- sessionClose();
+ sessionRequestTimeout(0);
+ sessionDetach(name);
// XXX: channel.close();
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/SessionDelegate.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/SessionDelegate.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/SessionDelegate.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/SessionDelegate.java Fri Apr 18 09:09:03 2008
@@ -35,20 +35,16 @@
{
public void init(Session ssn, ProtocolHeader hdr) { }
- public void method(Session ssn, Method method) {
- if (method.getEncodedTrack() == Frame.L4)
- {
- method.setId(ssn.nextCommandId());
- }
-
+ public void control(Session ssn, Method method) {
method.dispatch(ssn, this);
+ }
- if (method.getEncodedTrack() == Frame.L4)
+ public void command(Session ssn, Method method) {
+ method.setId(ssn.nextCommandId());
+ method.dispatch(ssn, this);
+ if (!method.hasPayload())
{
- if (!method.hasPayload())
- {
- ssn.processed(method);
- }
+ ssn.processed(method);
}
}
@@ -60,12 +56,12 @@
@Override public void executionResult(Session ssn, ExecutionResult result)
{
- ssn.result(result.getCommandId(), result.getData());
+ ssn.result(result.getCommandId(), result.getValue());
}
- @Override public void executionComplete(Session ssn, ExecutionComplete excmp)
+ @Override public void sessionCompleted(Session ssn, SessionCompleted cmp)
{
- RangeSet ranges = excmp.getRangedExecutionSet();
+ RangeSet ranges = cmp.getCommands();
if (ranges != null)
{
for (Range range : ranges)
@@ -73,12 +69,27 @@
ssn.complete(range.getLower(), range.getUpper());
}
}
- ssn.complete(excmp.getCumulativeExecutionMark());
}
- @Override public void executionFlush(Session ssn, ExecutionFlush flush)
+ @Override public void sessionFlush(Session ssn, SessionFlush flush)
+ {
+ if (flush.getCompleted())
+ {
+ ssn.flushProcessed();
+ }
+ if (flush.getConfirmed())
+ {
+ throw new Error("not implemented");
+ }
+ if (flush.getExpected())
+ {
+ throw new Error("not implemented");
+ }
+ }
+
+ @Override public void sessionCommandPoint(Session ssn, SessionCommandPoint scp)
{
- ssn.flushProcessed();
+ ssn.commandsIn = scp.getCommandId();
}
@Override public void executionSync(Session ssn, ExecutionSync sync)
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Struct.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Struct.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Struct.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/Struct.java Fri Apr 18 09:09:03 2008
@@ -71,8 +71,6 @@
return type;
}
- public abstract boolean hasTicket();
-
private final boolean isBit(Field<?,?> f)
{
return f.getType().equals(Boolean.class);
@@ -80,14 +78,7 @@
private final boolean packed()
{
- if (this instanceof Method)
- {
- return false;
- }
- else
- {
- return true;
- }
+ return getPackWidth() > 0;
}
private final boolean encoded(Field<?,?> f)
@@ -147,11 +138,6 @@
}
}
- if (hasTicket())
- {
- dec.readShort();
- }
-
for (Field<?,?> f : fields)
{
if (encoded(f))
@@ -185,11 +171,6 @@
{
enc.writeBit(false);
}
- }
-
- if (hasTicket())
- {
- enc.writeShort(0x0);
}
for (Field<?,?> f : fields)
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/TransportConstants.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/TransportConstants.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/TransportConstants.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/TransportConstants.java Fri Apr 18 09:09:03 2008
@@ -2,8 +2,9 @@
public class TransportConstants
{
- private static byte _protocol_version_minor = 0;
- private static byte _protocol_version_major = 99;
+
+ private static byte _protocol_version_minor = 10;
+ private static byte _protocol_version_major = 0;
public static void setVersionMajor(byte value)
{
@@ -24,4 +25,5 @@
{
return _protocol_version_minor;
}
+
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractDecoder.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractDecoder.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractDecoder.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractDecoder.java Fri Apr 18 09:09:03 2008
@@ -20,7 +20,10 @@
*/
package org.apache.qpidity.transport.codec;
+import java.io.UnsupportedEncodingException;
+
import java.util.ArrayList;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -99,19 +102,19 @@
nbits = 0;
}
- public short readOctet()
+ public short readUint8()
{
return uget();
}
- public int readShort()
+ public int readUint16()
{
int i = uget() << 8;
i |= uget();
return i;
}
- public long readLong()
+ public long readUint32()
{
long l = uget() << 24;
l |= uget() << 16;
@@ -120,7 +123,12 @@
return l;
}
- public long readLonglong()
+ public int readSequenceNo()
+ {
+ return (int) readUint32();
+ }
+
+ public long readUint64()
{
long l = 0;
for (int i = 0; i < 8; i++)
@@ -130,31 +138,67 @@
return l;
}
- public long readTimestamp()
+ public long readDatetime()
+ {
+ return readUint64();
+ }
+
+ private static final String decode(byte[] bytes, String charset)
{
- return readLonglong();
+ try
+ {
+ return new String(bytes, charset);
+ }
+ catch (UnsupportedEncodingException e)
+ {
+ throw new RuntimeException(e);
+ }
}
- public String readShortstr()
+ public String readStr8()
+ {
+ short size = readUint8();
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return decode(bytes, "UTF-8");
+ }
+
+ public String readStr16()
{
- short size = readOctet();
+ int size = readUint16();
byte[] bytes = new byte[size];
get(bytes);
- return new String(bytes);
+ return decode(bytes, "UTF-8");
}
- public String readLongstr()
+ public byte[] readVbin8()
{
- long size = readLong();
- byte[] bytes = new byte[(int) size];
+ int size = readUint8();
+ byte[] bytes = new byte[size];
get(bytes);
- return new String(bytes);
+ return bytes;
}
- public RangeSet readRfc1982LongSet()
+ public byte[] readVbin16()
{
- int count = readShort()/8;
+ int size = readUint16();
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return bytes;
+ }
+
+ public byte[] readVbin32()
+ {
+ int size = (int) readUint32();
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return bytes;
+ }
+
+ public RangeSet readSequenceSet()
+ {
+ int count = readUint16()/8;
if (count == 0)
{
return null;
@@ -164,16 +208,21 @@
RangeSet ranges = new RangeSet();
for (int i = 0; i < count; i++)
{
- ranges.add(readLong(), readLong());
+ ranges.add(readUint32(), readUint32());
}
return ranges;
}
}
+ public RangeSet readByteRanges()
+ {
+ throw new Error("not implemented");
+ }
+
public UUID readUuid()
{
- long msb = readLonglong();
- long lsb = readLonglong();
+ long msb = readUint64();
+ long lsb = readUint64();
return new UUID(msb, lsb);
}
@@ -194,34 +243,39 @@
return null;
}
}
+ if (type > 0)
+ {
+ int code = readUint16();
+ assert code == type;
+ }
st.read(this);
return st;
}
- public Struct readLongStruct()
+ public Struct readStruct32()
{
- long size = readLong();
+ long size = readUint32();
if (size == 0)
{
return null;
}
else
{
- int type = readShort();
+ int type = readUint16();
Struct result = Struct.create(type);
result.read(this);
return result;
}
}
- public Map<String,Object> readTable()
+ public Map<String,Object> readMap()
{
- long size = readLong();
+ long size = readUint32();
int start = count;
Map<String,Object> result = new LinkedHashMap();
while (count < start + size)
{
- String key = readShortstr();
+ String key = readStr8();
byte code = get();
Type t = getType(code);
Object value = read(t);
@@ -230,9 +284,9 @@
return result;
}
- public List<Object> readSequence()
+ public List<Object> readList()
{
- long size = readLong();
+ long size = readUint32();
int start = count;
List<Object> result = new ArrayList();
while (count < start + size)
@@ -247,10 +301,15 @@
public List<Object> readArray()
{
- long size = readLong();
+ long size = readUint32();
+ if (size == 0)
+ {
+ return Collections.EMPTY_LIST;
+ }
+
byte code = get();
Type t = getType(code);
- long count = readLong();
+ long count = readUint32();
List<Object> result = new ArrayList<Object>();
for (int i = 0; i < count; i++)
@@ -291,11 +350,11 @@
switch (width)
{
case 1:
- return readOctet();
+ return readUint8();
case 2:
- return readShort();
+ return readUint16();
case 4:
- return readLong();
+ return readUint32();
default:
throw new IllegalStateException("illegal width: " + width);
}
@@ -313,81 +372,72 @@
{
switch (t)
{
- case OCTET:
- case UNSIGNED_BYTE:
- return readOctet();
- case SIGNED_BYTE:
+ case BIN8:
+ case UINT8:
+ return readUint8();
+ case INT8:
return get();
case CHAR:
return (char) get();
case BOOLEAN:
return get() > 0;
- case TWO_OCTETS:
- case UNSIGNED_SHORT:
- return readShort();
-
- case SIGNED_SHORT:
- return (short) readShort();
-
- case FOUR_OCTETS:
- case UNSIGNED_INT:
- return readLong();
-
- case UTF32_CHAR:
- case SIGNED_INT:
- return (int) readLong();
+ case BIN16:
+ case UINT16:
+ return readUint16();
+
+ case INT16:
+ return (short) readUint16();
+
+ case BIN32:
+ case UINT32:
+ return readUint32();
+
+ case CHAR_UTF32:
+ case INT32:
+ return (int) readUint32();
case FLOAT:
- return Float.intBitsToFloat((int) readLong());
+ return Float.intBitsToFloat((int) readUint32());
- case EIGHT_OCTETS:
- case SIGNED_LONG:
- case UNSIGNED_LONG:
+ case BIN64:
+ case UINT64:
+ case INT64:
case DATETIME:
- return readLonglong();
+ return readUint64();
case DOUBLE:
- return Double.longBitsToDouble(readLonglong());
-
- case SIXTEEN_OCTETS:
- case THIRTY_TWO_OCTETS:
- case SIXTY_FOUR_OCTETS:
- case _128_OCTETS:
- case SHORT_BINARY:
- case BINARY:
- case LONG_BINARY:
- return readBytes(t);
+ return Double.longBitsToDouble(readUint64());
case UUID:
return readUuid();
- case SHORT_STRING:
- case SHORT_UTF8_STRING:
- case SHORT_UTF16_STRING:
- case SHORT_UTF32_STRING:
- case STRING:
- case UTF8_STRING:
- case UTF16_STRING:
- case UTF32_STRING:
- case LONG_STRING:
- case LONG_UTF8_STRING:
- case LONG_UTF16_STRING:
- case LONG_UTF32_STRING:
+ case STR8:
+ return readStr8();
+
+ case STR16:
+ return readStr16();
+
+ case STR8_LATIN:
+ case STR8_UTF16:
+ case STR16_LATIN:
+ case STR16_UTF16:
// XXX: need to do character conversion
return new String(readBytes(t));
- case TABLE:
- return readTable();
- case SEQUENCE:
- return readSequence();
+ case MAP:
+ return readMap();
+ case LIST:
+ return readList();
case ARRAY:
return readArray();
+ case STRUCT32:
+ return readStruct32();
- case FIVE_OCTETS:
- case DECIMAL:
- case NINE_OCTETS:
- case LONG_DECIMAL:
+ case BIN40:
+ case DEC32:
+ case BIN72:
+ case DEC64:
// XXX: what types are we supposed to use here?
return readBytes(t);
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractEncoder.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractEncoder.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractEncoder.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/AbstractEncoder.java Fri Apr 18 09:09:03 2008
@@ -20,8 +20,11 @@
*/
package org.apache.qpidity.transport.codec;
+import java.io.UnsupportedEncodingException;
+
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -48,17 +51,17 @@
static
{
ENCODINGS.put(Boolean.class, Type.BOOLEAN);
- ENCODINGS.put(String.class, Type.LONG_STRING);
- ENCODINGS.put(Long.class, Type.SIGNED_LONG);
- ENCODINGS.put(Integer.class, Type.SIGNED_INT);
- ENCODINGS.put(Short.class, Type.SIGNED_SHORT);
- ENCODINGS.put(Byte.class, Type.SIGNED_BYTE);
- ENCODINGS.put(Map.class, Type.TABLE);
- ENCODINGS.put(List.class, Type.SEQUENCE);
+ ENCODINGS.put(String.class, Type.STR16);
+ ENCODINGS.put(Long.class, Type.INT64);
+ ENCODINGS.put(Integer.class, Type.INT32);
+ ENCODINGS.put(Short.class, Type.INT16);
+ ENCODINGS.put(Byte.class, Type.INT8);
+ ENCODINGS.put(Map.class, Type.MAP);
+ ENCODINGS.put(List.class, Type.LIST);
ENCODINGS.put(Float.class, Type.FLOAT);
ENCODINGS.put(Double.class, Type.DOUBLE);
ENCODINGS.put(Character.class, Type.CHAR);
- ENCODINGS.put(byte[].class, Type.LONG_BINARY);
+ ENCODINGS.put(byte[].class, Type.VBIN32);
}
protected Sizer sizer()
@@ -120,14 +123,14 @@
flushBits();
}
- public void writeOctet(short b)
+ public void writeUint8(short b)
{
assert b < 0x100;
put((byte) b);
}
- public void writeShort(int s)
+ public void writeUint16(int s)
{
assert s < 0x10000;
@@ -135,7 +138,7 @@
put(lsb(s));
}
- public void writeLong(long i)
+ public void writeUint32(long i)
{
assert i < 0x100000000L;
@@ -145,7 +148,12 @@
put(lsb(i));
}
- public void writeLonglong(long l)
+ public void writeSequenceNo(int i)
+ {
+ writeUint32(i);
+ }
+
+ public void writeUint64(long l)
{
for (int i = 0; i < 8; i++)
{
@@ -154,47 +162,101 @@
}
- public void writeTimestamp(long l)
+ public void writeDatetime(long l)
{
- writeLonglong(l);
+ writeUint64(l);
}
+ private static final String checkLength(String s, int n)
+ {
+ if (s == null)
+ {
+ return "";
+ }
+
+ if (s.length() > n)
+ {
+ throw new IllegalArgumentException("string too long: " + s);
+ }
+ else
+ {
+ return s;
+ }
+ }
- public void writeShortstr(String s)
+ private static final byte[] encode(String s, String charset)
{
- if (s == null) { s = ""; }
- if (s.length() > 255) {
- throw new IllegalArgumentException(s);
+ try
+ {
+ return s.getBytes(charset);
+ }
+ catch (UnsupportedEncodingException e)
+ {
+ throw new RuntimeException(e);
}
- writeOctet((short) s.length());
- put(ByteBuffer.wrap(s.getBytes()));
}
- public void writeLongstr(String s)
+ public void writeStr8(String s)
+ {
+ s = checkLength(s, 255);
+ writeUint8((short) s.length());
+ put(ByteBuffer.wrap(encode(s, "UTF-8")));
+ }
+
+ public void writeStr16(String s)
{
- if (s == null) { s = ""; }
- writeLong(s.length());
- put(ByteBuffer.wrap(s.getBytes()));
+ s = checkLength(s, 65535);
+ writeUint16(s.length());
+ put(ByteBuffer.wrap(encode(s, "UTF-8")));
}
+ public void writeVbin8(byte[] bytes)
+ {
+ if (bytes == null) { bytes = new byte[0]; }
+ if (bytes.length > 255)
+ {
+ throw new IllegalArgumentException("array too long: " + bytes.length);
+ }
+ writeUint8((short) bytes.length);
+ put(ByteBuffer.wrap(bytes));
+ }
- public void writeRfc1982LongSet(RangeSet ranges)
+ public void writeVbin16(byte[] bytes)
+ {
+ if (bytes == null) { bytes = new byte[0]; }
+ writeUint16(bytes.length);
+ put(ByteBuffer.wrap(bytes));
+ }
+
+ public void writeVbin32(byte[] bytes)
+ {
+ if (bytes == null) { bytes = new byte[0]; }
+ writeUint32(bytes.length);
+ put(ByteBuffer.wrap(bytes));
+ }
+
+ public void writeSequenceSet(RangeSet ranges)
{
if (ranges == null)
{
- writeShort((short) 0);
+ writeUint16((short) 0);
}
else
{
- writeShort(ranges.size() * 8);
+ writeUint16(ranges.size() * 8);
for (Range range : ranges)
{
- writeLong(range.getLower());
- writeLong(range.getUpper());
+ writeUint32(range.getLower());
+ writeUint32(range.getUpper());
}
}
}
+ public void writeByteRanges(RangeSet ranges)
+ {
+ throw new Error("not implemented");
+ }
+
public void writeUuid(UUID uuid)
{
long msb = 0;
@@ -202,15 +264,10 @@
if (uuid != null)
{
msb = uuid.getMostSignificantBits();
- uuid.getLeastSignificantBits();
+ lsb = uuid.getLeastSignificantBits();
}
- writeLonglong(msb);
- writeLonglong(lsb);
- }
-
- public void writeContent(String c)
- {
- throw new Error("Deprecated");
+ writeUint64(msb);
+ writeUint64(lsb);
}
public void writeStruct(int type, Struct s)
@@ -237,22 +294,27 @@
}
}
+ if (type > 0)
+ {
+ writeUint16(type);
+ }
+
s.write(this);
}
- public void writeLongStruct(Struct s)
+ public void writeStruct32(Struct s)
{
if (s == null)
{
- writeLong(0);
+ writeUint32(0);
}
else
{
Sizer sizer = sizer();
- sizer.writeShort(s.getEncodedType());
+ sizer.writeUint16(s.getEncodedType());
s.write(sizer);
- writeLong(sizer.size());
- writeShort(s.getEncodedType());
+ writeUint32(sizer.size());
+ writeUint16(s.getEncodedType());
s.write(this);
}
}
@@ -309,46 +371,46 @@
return null;
}
- public void writeTable(Map<String,Object> table)
+ public void writeMap(Map<String,Object> map)
{
- if (table == null)
+ if (map == null)
{
- writeLong(0);
+ writeUint32(0);
return;
}
Sizer sizer = sizer();
- sizer.writeTable(table);
+ sizer.writeMap(map);
// XXX: - 4
- writeLong(sizer.size() - 4);
- writeTableEntries(table);
+ writeUint32(sizer.size() - 4);
+ writeMapEntries(map);
}
- protected void writeTableEntries(Map<String,Object> table)
+ protected void writeMapEntries(Map<String,Object> map)
{
- for (Map.Entry<String,Object> entry : table.entrySet())
+ for (Map.Entry<String,Object> entry : map.entrySet())
{
String key = entry.getKey();
Object value = entry.getValue();
Type type = encoding(value);
- writeShortstr(key);
+ writeStr8(key);
put(type.code);
write(type, value);
}
}
- public void writeSequence(List<Object> sequence)
+ public void writeList(List<Object> list)
{
Sizer sizer = sizer();
- sizer.writeSequence(sequence);
+ sizer.writeList(list);
// XXX: - 4
- writeLong(sizer.size() - 4);
- writeSequenceEntries(sequence);
+ writeUint32(sizer.size() - 4);
+ writeListEntries(list);
}
- protected void writeSequenceEntries(List<Object> sequence)
+ protected void writeListEntries(List<Object> list)
{
- for (Object value : sequence)
+ for (Object value : list)
{
Type type = encoding(value);
put(type.code);
@@ -358,10 +420,15 @@
public void writeArray(List<Object> array)
{
+ if (array == null)
+ {
+ array = Collections.EMPTY_LIST;
+ }
+
Sizer sizer = sizer();
sizer.writeArray(array);
// XXX: -4
- writeLong(sizer.size() - 4);
+ writeUint32(sizer.size() - 4);
writeArrayEntries(array);
}
@@ -371,7 +438,7 @@
if (array.isEmpty())
{
- type = Type.VOID;
+ return;
}
else
{
@@ -380,6 +447,8 @@
put(type.code);
+ writeUint32(array.size());
+
for (Object value : array)
{
write(type, value);
@@ -409,13 +478,13 @@
switch (width)
{
case 1:
- writeOctet((short) size);
+ writeUint8((short) size);
break;
case 2:
- writeShort(size);
+ writeUint16(size);
break;
case 4:
- writeLong(size);
+ writeUint32(size);
break;
default:
throw new IllegalStateException("illegal width: " + width);
@@ -444,11 +513,11 @@
{
switch (t)
{
- case OCTET:
- case UNSIGNED_BYTE:
- writeOctet(coerce(Short.class, value));
+ case BIN8:
+ case UINT8:
+ writeUint8(coerce(Short.class, value));
break;
- case SIGNED_BYTE:
+ case INT8:
put(coerce(Byte.class, value));
break;
case CHAR:
@@ -465,85 +534,78 @@
}
break;
- case TWO_OCTETS:
- case UNSIGNED_SHORT:
- writeShort(coerce(Integer.class, value));
+ case BIN16:
+ case UINT16:
+ writeUint16(coerce(Integer.class, value));
break;
- case SIGNED_SHORT:
- writeShort(coerce(Short.class, value));
+ case INT16:
+ writeUint16(coerce(Short.class, value));
break;
- case FOUR_OCTETS:
- case UNSIGNED_INT:
- writeLong(coerce(Long.class, value));
+ case BIN32:
+ case UINT32:
+ writeUint32(coerce(Long.class, value));
break;
- case UTF32_CHAR:
- case SIGNED_INT:
- writeLong(coerce(Integer.class, value));
+ case CHAR_UTF32:
+ case INT32:
+ writeUint32(coerce(Integer.class, value));
break;
case FLOAT:
- writeLong(Float.floatToIntBits(coerce(Float.class, value)));
+ writeUint32(Float.floatToIntBits(coerce(Float.class, value)));
break;
- case EIGHT_OCTETS:
- case SIGNED_LONG:
- case UNSIGNED_LONG:
+ case BIN64:
+ case UINT64:
+ case INT64:
case DATETIME:
- writeLonglong(coerce(Long.class, value));
+ writeUint64(coerce(Long.class, value));
break;
case DOUBLE:
long bits = Double.doubleToLongBits(coerce(Double.class, value));
- writeLonglong(bits);
- break;
-
- case SIXTEEN_OCTETS:
- case THIRTY_TWO_OCTETS:
- case SIXTY_FOUR_OCTETS:
- case _128_OCTETS:
- case SHORT_BINARY:
- case BINARY:
- case LONG_BINARY:
- writeBytes(t, coerce(byte[].class, value));
+ writeUint64(bits);
break;
case UUID:
writeUuid(coerce(UUID.class, value));
break;
- case SHORT_STRING:
- case SHORT_UTF8_STRING:
- case SHORT_UTF16_STRING:
- case SHORT_UTF32_STRING:
- case STRING:
- case UTF8_STRING:
- case UTF16_STRING:
- case UTF32_STRING:
- case LONG_STRING:
- case LONG_UTF8_STRING:
- case LONG_UTF16_STRING:
- case LONG_UTF32_STRING:
+ case STR8:
+ writeStr8(coerce(String.class, value));
+ break;
+
+ case STR16:
+ writeStr16(coerce(String.class, value));
+ break;
+
+ case STR8_LATIN:
+ case STR8_UTF16:
+ case STR16_LATIN:
+ case STR16_UTF16:
// XXX: need to do character conversion
writeBytes(t, coerce(String.class, value).getBytes());
break;
- case TABLE:
- writeTable((Map<String,Object>) coerce(Map.class, value));
+ case MAP:
+ writeMap((Map<String,Object>) coerce(Map.class, value));
break;
- case SEQUENCE:
- writeSequence(coerce(List.class, value));
+ case LIST:
+ writeList(coerce(List.class, value));
break;
case ARRAY:
writeArray(coerce(List.class, value));
break;
+ case STRUCT32:
+ writeStruct32(coerce(Struct.class, value));
+ break;
- case FIVE_OCTETS:
- case DECIMAL:
- case NINE_OCTETS:
- case LONG_DECIMAL:
+ case BIN40:
+ case DEC32:
+ case BIN72:
+ case DEC64:
// XXX: what types are we supposed to use here?
writeBytes(t, coerce(byte[].class, value));
break;
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Decoder.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Decoder.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Decoder.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Decoder.java Fri Apr 18 09:09:03 2008
@@ -38,26 +38,30 @@
{
boolean readBit();
- short readOctet();
- int readShort();
- long readLong();
- long readLonglong();
+ short readUint8();
+ int readUint16();
+ long readUint32();
+ long readUint64();
- long readTimestamp();
-
- String readShortstr();
- String readLongstr();
-
- RangeSet readRfc1982LongSet();
+ long readDatetime();
UUID readUuid();
- String readContent();
+ int readSequenceNo();
+ RangeSet readSequenceSet(); // XXX
+ RangeSet readByteRanges(); // XXX
+
+ String readStr8();
+ String readStr16();
+
+ byte[] readVbin8();
+ byte[] readVbin16();
+ byte[] readVbin32();
+
+ Struct readStruct32();
+ Map<String,Object> readMap();
+ List<Object> readList();
+ List<Object> readArray();
Struct readStruct(int type);
- Struct readLongStruct();
-
- Map<String,Object> readTable();
- List<Object> readSequence();
- List<Object> readArray();
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Encoder.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Encoder.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Encoder.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Encoder.java Fri Apr 18 09:09:03 2008
@@ -40,26 +40,30 @@
void flush();
void writeBit(boolean b);
- void writeOctet(short b);
- void writeShort(int s);
- void writeLong(long i);
- void writeLonglong(long l);
+ void writeUint8(short b);
+ void writeUint16(int s);
+ void writeUint32(long i);
+ void writeUint64(long l);
- void writeTimestamp(long l);
-
- void writeShortstr(String s);
- void writeLongstr(String s);
-
- void writeRfc1982LongSet(RangeSet ranges);
+ void writeDatetime(long l);
void writeUuid(UUID uuid);
- void writeContent(String c);
+ void writeSequenceNo(int s);
+ void writeSequenceSet(RangeSet ranges); // XXX
+ void writeByteRanges(RangeSet ranges); // XXX
+
+ void writeStr8(String s);
+ void writeStr16(String s);
+
+ void writeVbin8(byte[] bytes);
+ void writeVbin16(byte[] bytes);
+ void writeVbin32(byte[] bytes);
+
+ void writeStruct32(Struct s);
+ void writeMap(Map<String,Object> map);
+ void writeList(List<Object> list);
+ void writeArray(List<Object> array);
void writeStruct(int type, Struct s);
- void writeLongStruct(Struct s);
-
- void writeTable(Map<String,Object> table);
- void writeSequence(List<Object> sequence);
- void writeArray(List<Object> array);
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Sizer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Sizer.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Sizer.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/codec/Sizer.java Fri Apr 18 09:09:03 2008
@@ -38,30 +38,34 @@
public static final Sizer NULL = new Sizer()
{
- public void flush() {};
+ public void flush() {}
- public void writeBit(boolean b) {};
- public void writeOctet(short b) {};
- public void writeShort(int s) {};
- public void writeLong(long i) {};
- public void writeLonglong(long l) {};
+ public void writeBit(boolean b) {}
+ public void writeUint8(short b) {}
+ public void writeUint16(int s) {}
+ public void writeUint32(long i) {}
+ public void writeUint64(long l) {}
+
+ public void writeDatetime(long l) {}
+ public void writeUuid(UUID uuid) {}
+
+ public void writeSequenceNo(int s) {}
+ public void writeSequenceSet(RangeSet ranges) {} // XXX
+ public void writeByteRanges(RangeSet ranges) {} // XXX
+
+ public void writeStr8(String s) {}
+ public void writeStr16(String s) {}
+
+ public void writeVbin8(byte[] bytes) {}
+ public void writeVbin16(byte[] bytes) {}
+ public void writeVbin32(byte[] bytes) {}
+
+ public void writeStruct32(Struct s) {}
+ public void writeMap(Map<String,Object> map) {}
+ public void writeList(List<Object> list) {}
+ public void writeArray(List<Object> array) {}
- public void writeTimestamp(long l) {};
-
- public void writeShortstr(String s) {};
- public void writeLongstr(String s) {};
-
- public void writeRfc1982LongSet(RangeSet ranges) {};
- public void writeUuid(UUID uuid) {};
-
- public void writeContent(String c) {};
-
- public void writeStruct(int type, Struct s) {};
- public void writeLongStruct(Struct s) {};
-
- public void writeTable(Map<String,Object> table) {};
- public void writeSequence(List<Object> sequence) {};
- public void writeArray(List<Object> array) {};
+ public void writeStruct(int type, Struct s) {}
public int getSize() { return 0; }
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Assembler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Assembler.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Assembler.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Assembler.java Fri Apr 18 09:09:03 2008
@@ -37,6 +37,7 @@
import org.apache.qpidity.transport.ProtocolEvent;
import org.apache.qpidity.transport.ProtocolHeader;
import org.apache.qpidity.transport.Receiver;
+import org.apache.qpidity.transport.SegmentType;
import org.apache.qpidity.transport.Struct;
@@ -118,7 +119,7 @@
{
switch (frame.getType())
{
- case Frame.BODY:
+ case BODY:
emit(frame, new Data(frame, frame.isFirstFrame(),
frame.isLastFrame()));
break;
@@ -158,22 +159,29 @@
}
}
- private ProtocolEvent decode(Frame frame, byte type, List<ByteBuffer> segment)
+ private ProtocolEvent decode(Frame frame, SegmentType type, List<ByteBuffer> segment)
{
FragmentDecoder dec = new FragmentDecoder(segment.iterator());
switch (type)
{
- case Frame.METHOD:
- int methodType = dec.readShort();
- Method method = Method.create(methodType);
- method.read(dec);
- return method;
- case Frame.HEADER:
+ case CONTROL:
+ int controlType = dec.readUint16();
+ Method control = Method.create(controlType);
+ control.read(dec);
+ return control;
+ case COMMAND:
+ int commandType = dec.readUint16();
+ // read in the session header, right now we don't use it
+ dec.readUint16();
+ Method command = Method.create(commandType);
+ command.read(dec);
+ return command;
+ case HEADER:
List<Struct> structs = new ArrayList();
while (dec.hasRemaining())
{
- structs.add(dec.readLongStruct());
+ structs.add(dec.readStruct32());
}
return new Header(structs,frame.isLastFrame() && frame.isLastSegment());
default:
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Disassembler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Disassembler.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Disassembler.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Disassembler.java Fri Apr 18 09:09:03 2008
@@ -31,6 +31,7 @@
import org.apache.qpidity.transport.ProtocolError;
import org.apache.qpidity.transport.ProtocolEvent;
import org.apache.qpidity.transport.ProtocolHeader;
+import org.apache.qpidity.transport.SegmentType;
import org.apache.qpidity.transport.Sender;
import org.apache.qpidity.transport.Struct;
@@ -76,50 +77,50 @@
sender.close();
}
- private void fragment(byte flags, byte type, ConnectionEvent event,
+ private void fragment(byte flags, SegmentType type, ConnectionEvent event,
ByteBuffer buf, boolean first, boolean last)
{
if(!buf.hasRemaining())
{
//empty data
- byte nflags = flags;
+ byte nflags = flags;
if (first)
{
nflags |= FIRST_FRAME;
first = false;
}
- nflags |= LAST_FRAME;
- Frame frame = new Frame(nflags, type,
+ nflags |= LAST_FRAME;
+ Frame frame = new Frame(nflags, type,
event.getProtocolEvent().getEncodedTrack(),
event.getChannel());
- // frame.addFragment(buf);
+ // frame.addFragment(buf);
sender.send(frame);
}
else
{
- while (buf.hasRemaining())
- {
- ByteBuffer slice = buf.slice();
- slice.limit(min(maxPayload, slice.remaining()));
- buf.position(buf.position() + slice.remaining());
-
- byte newflags = flags;
- if (first)
+ while (buf.hasRemaining())
{
- newflags |= FIRST_FRAME;
- first = false;
+ ByteBuffer slice = buf.slice();
+ slice.limit(min(maxPayload, slice.remaining()));
+ buf.position(buf.position() + slice.remaining());
+
+ byte newflags = flags;
+ if (first)
+ {
+ newflags |= FIRST_FRAME;
+ first = false;
+ }
+ if (last && !buf.hasRemaining())
+ {
+ newflags |= LAST_FRAME;
+ }
+
+ Frame frame = new Frame(newflags, type,
+ event.getProtocolEvent().getEncodedTrack(),
+ event.getChannel());
+ frame.addFragment(slice);
+ sender.send(frame);
}
- if (last && !buf.hasRemaining())
- {
- newflags |= LAST_FRAME;
- }
-
- Frame frame = new Frame(newflags, type,
- event.getProtocolEvent().getEncodedTrack(),
- event.getChannel());
- frame.addFragment(slice);
- sender.send(frame);
- }
}
}
@@ -128,15 +129,40 @@
sender.send(header);
}
- public void method(ConnectionEvent event, Method method)
+ public void control(ConnectionEvent event, Method method)
+ {
+ method(event, method, SegmentType.CONTROL);
+ }
+
+ public void command(ConnectionEvent event, Method method)
+ {
+ method(event, method, SegmentType.COMMAND);
+ }
+
+ private void method(ConnectionEvent event, Method method, SegmentType type)
{
SizeEncoder sizer = new SizeEncoder();
- sizer.writeShort(method.getEncodedType());
+ sizer.writeUint16(method.getEncodedType());
+ if (type == SegmentType.COMMAND)
+ {
+ sizer.writeUint16(0);
+ }
method.write(sizer);
ByteBuffer buf = ByteBuffer.allocate(sizer.size());
BBEncoder enc = new BBEncoder(buf);
- enc.writeShort(method.getEncodedType());
+ enc.writeUint16(method.getEncodedType());
+ if (type == SegmentType.COMMAND)
+ {
+ if (method.isSync())
+ {
+ enc.writeUint16(0x0101);
+ }
+ else
+ {
+ enc.writeUint16(0x0100);
+ }
+ }
method.write(enc);
enc.flush();
buf.flip();
@@ -148,7 +174,7 @@
flags |= LAST_SEG;
}
- fragment(flags, METHOD, event, buf, true, true);
+ fragment(flags, type, event, buf, true, true);
}
public void header(ConnectionEvent event, Header header)
@@ -159,24 +185,24 @@
SizeEncoder sizer = new SizeEncoder();
for (Struct st : header.getStructs())
{
- sizer.writeLongStruct(st);
+ sizer.writeStruct32(st);
}
buf = ByteBuffer.allocate(sizer.size());
BBEncoder enc = new BBEncoder(buf);
for (Struct st : header.getStructs())
{
- enc.writeLongStruct(st);
+ enc.writeStruct32(st);
enc.flush();
}
header.setBuf(buf);
}
else
{
- buf = header.getBuf();
+ buf = header.getBuf();
}
buf.flip();
- fragment((byte) 0x0, HEADER, event, buf, true, true);
+ fragment((byte) 0x0, SegmentType.HEADER, event, buf, true, true);
}
public void data(ConnectionEvent event, Data data)
@@ -187,7 +213,7 @@
{
ByteBuffer buf = it.next();
boolean last = data.isLast() && !it.hasNext();
- fragment(LAST_SEG, BODY, event, buf, first, last);
+ fragment(LAST_SEG, SegmentType.BODY, event, buf, first, last);
first = false;
}
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Frame.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Frame.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Frame.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/Frame.java Fri Apr 18 09:09:03 2008
@@ -20,6 +20,7 @@
*/
package org.apache.qpidity.transport.network;
+import org.apache.qpidity.transport.SegmentType;
import org.apache.qpidity.transport.util.SliceIterator;
import java.nio.ByteBuffer;
@@ -48,10 +49,6 @@
public static final byte L3 = 2;
public static final byte L4 = 3;
- public static final byte METHOD = 1;
- public static final byte HEADER = 2;
- public static final byte BODY = 3;
-
public static final byte RESERVED = 0x0;
public static final byte VERSION = 0x0;
@@ -62,13 +59,13 @@
public static final byte LAST_FRAME = 0x1;
final private byte flags;
- final private byte type;
+ final private SegmentType type;
final private byte track;
final private int channel;
final private List<ByteBuffer> fragments;
private int size;
- public Frame(byte flags, byte type, byte track, int channel)
+ public Frame(byte flags, SegmentType type, byte track, int channel)
{
this.flags = flags;
this.type = type;
@@ -99,7 +96,7 @@
return size;
}
- public byte getType()
+ public SegmentType getType()
{
return type;
}
@@ -153,7 +150,7 @@
{
StringBuilder str = new StringBuilder();
str.append(String.format
- ("[%05d %05d %1d %1d %d%d%d%d]", getChannel(), getSize(),
+ ("[%05d %05d %1d %1d %d%d%d%d] ", getChannel(), getSize(),
getTrack(), getType(),
isFirstSegment() ? 1 : 0, isLastSegment() ? 1 : 0,
isFirstFrame() ? 1 : 0, isLastFrame() ? 1 : 0));
@@ -170,7 +167,7 @@
str.append(" | ");
}
- str.append(str(buf, 20));
+ str.append(str(buf));
}
return str.toString();
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java Fri Apr 18 09:09:03 2008
@@ -22,10 +22,10 @@
import java.nio.ByteBuffer;
-import org.apache.qpidity.transport.Constant;
import org.apache.qpidity.transport.ProtocolError;
import org.apache.qpidity.transport.ProtocolHeader;
import org.apache.qpidity.transport.Receiver;
+import org.apache.qpidity.transport.SegmentType;
import static org.apache.qpidity.transport.util.Functions.*;
@@ -77,7 +77,7 @@
private byte minor;
private byte flags;
- private byte type;
+ private SegmentType type;
private byte track;
private int channel;
private int size;
@@ -146,7 +146,7 @@
flags = buf.get();
return FRAME_HDR_TYPE;
case FRAME_HDR_TYPE:
- type = buf.get();
+ type = SegmentType.get(buf.get());
return FRAME_HDR_SIZE1;
case FRAME_HDR_SIZE1:
size = (0xFF & buf.get()) << 8;
@@ -218,7 +218,7 @@
return FRAME_END;
}
case FRAME_END:
- return expect(buf, Constant.FRAME_END, FRAME_HDR);
+ return expect(buf, OutputHandler.FRAME_END, FRAME_HDR);
default:
throw new IllegalStateException();
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/OutputHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/OutputHandler.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/OutputHandler.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/OutputHandler.java Fri Apr 18 09:09:03 2008
@@ -67,11 +67,13 @@
}
}
+ public static final int FRAME_END = 0xCE;
+
public void frame(Frame frame)
{
ByteBuffer hdr = ByteBuffer.allocate(HEADER_SIZE + frame.getSize() + 1);
hdr.put(frame.getFlags());
- hdr.put(frame.getType());
+ hdr.put((byte) frame.getType().getValue());
hdr.putShort((short) (frame.getSize() + HEADER_SIZE));
hdr.put(RESERVED);
hdr.put(frame.getTrack());
@@ -84,7 +86,7 @@
{
hdr.put(buf);
}
- hdr.put((byte) Constant.FRAME_END);
+ hdr.put((byte) FRAME_END);
hdr.flip();
synchronized (lock)
{
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/mina/MinaHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/mina/MinaHandler.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/mina/MinaHandler.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/network/mina/MinaHandler.java Fri Apr 18 09:09:03 2008
@@ -200,7 +200,7 @@
ConnectionDelegate delegate)
{
return connect(host, port, new ConnectionBinding
- (delegate, InputHandler.State.FRAME_HDR));
+ (delegate, InputHandler.State.PROTO_HDR));
}
private static class ConnectionBinding
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/util/Functions.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/util/Functions.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/util/Functions.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/main/java/org/apache/qpidity/transport/util/Functions.java Fri Apr 18 09:09:03 2008
@@ -52,14 +52,23 @@
public static final String str(ByteBuffer buf, int limit)
{
StringBuilder str = new StringBuilder();
+ str.append('"');
+
for (int i = 0; i < min(buf.remaining(), limit); i++)
{
- if (i > 0 && i % 2 == 0)
+ byte c = buf.get(buf.position() + i);
+
+ if (c > 31 && c < 127 && c != '\\')
{
- str.append(" ");
+ str.append((char)c);
+ }
+ else
+ {
+ str.append(String.format("\\x%02x", c));
}
- str.append(String.format("%02x", buf.get(buf.position() + i)));
}
+
+ str.append('"');
return str.toString();
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/common/src/test/java/org/apache/qpidity/transport/ConnectionTest.java Fri Apr 18 09:09:03 2008
@@ -100,12 +100,12 @@
}
Channel ch = conn.getChannel(0);
- Session ssn = new Session();
+ Session ssn = new Session("test".getBytes());
ssn.attach(ch);
try
{
- ssn.sessionOpen(1234);
+ ssn.sessionAttach(ssn.getName());
fail("writing to a closed socket succeeded");
}
catch (TransportException e)
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/cpp.async.testprofile
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/cpp.async.testprofile?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/cpp.async.testprofile (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/cpp.async.testprofile Fri Apr 18 09:09:03 2008
@@ -1,10 +1,10 @@
broker.version=0-10
-broker=${project.root}/../cpp/src/qpidd --data-dir ${build.data} -t --log-output ${build.data}/broker.log --load-module ${project.root}/../../cppStore/cpp/lib/.libs/libbdbstore.so --store-async yes
-broker.clean=${build.data}
+broker=${project.root}/../cpp/src/qpidd --data-dir ${build.data} -t --load-module ${project.root}/../../cppStore/cpp/lib/.libs/libbdbstore.so --store-async yes --auth no
+broker.clean=${project.root}/clean-dir ${build.data}
java.naming.provider.url=${project.root}/test-provider.properties
max_prefetch=1000
test.excludes=true
-test.excludesfile=${project.root}/010ExcludeList
+test.excludesfile=${project.root}/010ExcludeList-store
log=info
amqj.logging.level=$log
root.logging.level=$log
@@ -14,4 +14,4 @@
test=*Test
test1=*Tests
haltonfailure=no
-haltonerror=no
\ No newline at end of file
+haltonerror=no
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/cpp.sync.testprofile
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/cpp.sync.testprofile?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/cpp.sync.testprofile (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/cpp.sync.testprofile Fri Apr 18 09:09:03 2008
@@ -1,10 +1,10 @@
broker.version=0-10
-broker=${project.root}/../cpp/src/qpidd --data-dir ${build.data} -t --log-output ${build.data}/broker.log --load-module ${project.root}/../../cppStore/cpp/lib/.libs/libbdbstore.so --store-async no
-broker.clean=${build.data}
+broker=${project.root}/../cpp/src/qpidd --data-dir ${build.data} -t --load-module ${project.root}/../../cppStore/cpp/lib/.libs/libbdbstore.so --store-async no --auth no
+broker.clean=${project.root}/clean-dir ${build.data}
java.naming.provider.url=${project.root}/test-provider.properties
test.excludes=true
max_prefetch=1000
-test.excludesfile=${project.root}/010ExcludeList
+test.excludesfile=${project.root}/010ExcludeList-store
log=info
amqj.logging.level=$log
root.logging.level=$log
Modified: incubator/qpid/branches/thegreatmerge/qpid/java/default.testprofile
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/java/default.testprofile?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/java/default.testprofile (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/java/default.testprofile Fri Apr 18 09:09:03 2008
@@ -1,10 +1,11 @@
broker.version=0-8
broker=vm
-broker.clean=${project.root}/clean-dir
+broker.clean=${project.root}/clean-dir ${build.data}
java.naming.provider.url=${project.root}/test-provider.properties
test.excludes=true
test.excludesfile=${project.root}/08ExcludeList
log=info
+max_prefetch=1000
amqj.logging.level=$log
root.logging.level=$log
log4j.configuration=file://${project.root}/log4j-test.xml
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-config
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-config?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-config (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-config Fri Apr 18 09:09:03 2008
@@ -97,13 +97,17 @@
try:
self.spec = qpid.spec.load (_specpath)
self.client = Client (self.broker.host, self.broker.port, self.spec)
- self.client.start ({"LOGIN":"guest","PASSWORD":"guest"})
+ self.client.start (response='\x00' + "guest" + '\x00' + "guest",
+ mechanism="PLAIN")
self.channel = self.client.channel (1)
self.mclient = managementClient (self.spec)
self.mchannel = self.mclient.addChannel (self.channel)
except socket.error, e:
- print "Connect Error:", e
- exit (1)
+ print "Socket Error:", e
+ sys.exit (1)
+ except Closed, e:
+ print "Connect Failed:", e
+ sys.exit (1)
def Overview (self):
self.ConnectToBroker ()
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-route
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-route?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-route (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/commands/qpid-route Fri Apr 18 09:09:03 2008
@@ -79,7 +79,8 @@
try:
self.spec = qpid.spec.load (_specpath)
self.client = Client (broker.host, broker.port, self.spec)
- self.client.start ({"LOGIN":"guest","PASSWORD":"guest"})
+ self.client.start (response='\x00' + "guest" + '\x00' + "guest",
+ mechanism="PLAIN")
self.channel = self.client.channel (1)
self.mclient = managementClient (self.spec)
self.mch = self.mclient.addChannel (self.channel)
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/qpid/management.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/qpid/management.py?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/qpid/management.py (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/qpid/management.py Fri Apr 18 09:09:03 2008
@@ -83,11 +83,11 @@
class managementChannel:
""" This class represents a connection to an AMQP broker. """
- def __init__ (self, ch, topicCb, replyCb, cbContext):
+ def __init__ (self, ch, topicCb, replyCb, cbContext, _detlife=0):
""" Given a channel on an established AMQP broker connection, this method
opens a session and performs all of the declarations and bindings needed
to participate in the management protocol. """
- response = ch.session_open (detached_lifetime=300)
+ response = ch.session_open (detached_lifetime=_detlife)
self.sessionId = response.session_id
self.topicName = "mgmt-%08x-%04x-%04x-%04x-%04x%08x" % struct.unpack ("!LHHHHL", response.session_id)
self.replyName = "repl-%08x-%04x-%04x-%04x-%04x%08x" % struct.unpack ("!LHHHHL", response.session_id)
@@ -97,8 +97,8 @@
self.context = cbContext
self.reqsOutstanding = 0
- ch.queue_declare (queue=self.topicName, exclusive=1, auto_delete=1)
- ch.queue_declare (queue=self.replyName, exclusive=1, auto_delete=1)
+ ch.queue_declare (queue=self.topicName, exclusive=True, auto_delete=True)
+ ch.queue_declare (queue=self.replyName, exclusive=True, auto_delete=True)
ch.queue_bind (exchange="qpid.management",
queue=self.topicName, routing_key="mgmt.#")
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/qpid/managementdata.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/qpid/managementdata.py?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/qpid/managementdata.py (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/qpid/managementdata.py Fri Apr 18 09:09:03 2008
@@ -21,6 +21,7 @@
import qpid
import socket
+import struct
from qpid.management import managementChannel, managementClient
from threading import Lock
from disp import Display
@@ -63,19 +64,24 @@
#
def registerObjId (self, objId):
- if self.baseId == 0:
- if objId & 0x8000000000000000L == 0:
- self.baseId = objId - 1000
+ boot = objId & 0x7FFF000000000000L
+ if boot == 0:
+ return
+ self.bootSequence = boot
def displayObjId (self, objId):
- if objId & 0x8000000000000000L == 0:
- return objId - self.baseId
- return (objId & 0x7fffffffffffffffL) + 5000
+ bank = (objId & 0x0000FFFFFF000000L) >> 24
+ id = objId & 0x0000000000FFFFFFL
+ return bank * 1000 + id
def rawObjId (self, displayId):
- if displayId < 5000:
- return displayId + self.baseId
- return displayId - 5000 + 0x8000000000000000L
+ bank = displayId / 1000
+ id = displayId % 1000
+ if bank < 3:
+ objId = (bank << 24) + id
+ else:
+ objId = self.bootSequence + (bank << 24) + id
+ return objId
def displayClassName (self, cls):
(packageName, className, hash) = cls
@@ -158,7 +164,7 @@
self.lock = Lock ()
self.tables = {}
self.schema = {}
- self.baseId = 0
+ self.bootSequence = 0
self.disp = disp
self.lastUnit = None
self.methodSeq = 1
@@ -166,7 +172,8 @@
self.broker = Broker (host)
self.client = Client (self.broker.host, self.broker.port, self.spec)
- self.client.start ({"LOGIN": username, "PASSWORD": password})
+ self.client.start (response='\x00' + username + '\x00' + password,
+ mechanism="PLAIN")
self.channel = self.client.channel (1)
self.mclient = managementClient (self.spec, self.ctrlHandler, self.configHandler,
@@ -189,7 +196,7 @@
if item[0] == key:
typecode = item[1]
unit = item[2]
- if (typecode >= 1 and typecode <= 5) or typecode >= 12: # numerics
+ if (typecode >= 1 and typecode <= 5) or typecode == 12 or typecode == 13: # numerics
if unit == None or unit == self.lastUnit:
return str (value)
else:
@@ -214,7 +221,7 @@
else:
return "True"
elif typecode == 14:
- return str (UUID (bytes=value))
+ return "%08x-%04x-%04x-%04x-%04x%08x" % struct.unpack ("!LHHHHL", value)
elif typecode == 15:
return str (value)
return "*type-error*"
Modified: incubator/qpid/branches/thegreatmerge/qpid/specs/management-schema.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/specs/management-schema.xml?rev=649585&r1=649584&r2=649585&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/specs/management-schema.xml (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/specs/management-schema.xml Fri Apr 18 09:09:03 2008
@@ -45,7 +45,7 @@
===============================================================
-->
<class name="system">
- <configElement name="sysId" index="y" type="sstr" access="RC"/>
+ <configElement name="sysId" index="y" type="uuid" access="RC"/>
<configElement name="osName" type="sstr" access="RO" desc="Operating System Name"/>
<configElement name="nodeName" type="sstr" access="RO" desc="Node Name"/>