@@ -14,6 +14,9 @@ org.amqp.patterns.RpcServer TODO / Further exploration: --------------------------- +- The shutdown handling is probably not inline with the protocol. + (Look into the LifecycleTest to see how the connection is being closed) + - Optimize the file size of overall library to decrease its 304kb weight down to something reasonable. For one, I noticed the core library depends on flexunit.swc, which weighs in at 400kb itself. README @@ -149,7 +149,6 @@ package org.amqp var frame:Frame = parseFrame(delegate); maybeSendHeartbeat(); if (frame != null) { - // missedHeartbeats = 0; if (frame.type == AMQP.FRAME_HEARTBEAT) { // just ignore this for now } else if (frame.channel == 0) { @@ -173,7 +172,6 @@ package org.amqp if (delegate.isConnected()) { frame.writeTo(delegate); delegate.flush(); - //lastActivityTime = new Date().valueOf(); } else { throw new Error("Connection main loop not running"); } src/org/amqp/Connection.as @@ -95,7 +95,6 @@ package org.amqp if (accumulator != null) { payload.writeBytes(accumulator,0,accumulator.bytesAvailable); payload.position = 0; - accumulator = null; } } @@ -106,14 +105,9 @@ package org.amqp public function writeTo(os:IDataOutput):void{ finishWriting(); os.writeByte(type); - //os.writeByte(0); os.writeShort(channel); os.writeInt(payload.length); - //accumulator.position = 0; - //trace("ba = " + accumulator.length); - //os.writeInt(accumulator.length); os.writeBytes(payload); - //os.writeBytes(accumulator, 0, accumulator.bytesAvailable); os.writeByte(AMQP.FRAME_END); } @@ -143,4 +137,4 @@ package org.amqp return accumulator; } } -} \ No newline at end of file +} src/org/amqp/Frame.as @@ -24,13 +24,11 @@ package org.amqp { public static function shortStrSize(str:String):int{ return str.length + 1; - //str.getBytes("utf-8").length + 1; } /** Computes the AMQP wire-protocol length of a protocol-encoded long string. */ public static function longStrSize(str:String):int { return str.length + 4; - //str.getBytes("utf-8").length + 4; } public static function tableSize(table:Map):int{ @@ -52,11 +50,6 @@ package org.amqp else if(value is int) { acc += 4; } - /* - else if(value is BigDecimal) { - acc += 5; - } - */ else if(value is Date) { acc += 8; } @@ -71,4 +64,4 @@ package org.amqp return acc; } } -} \ No newline at end of file +} src/org/amqp/FrameHelper.as @@ -22,27 +22,8 @@ package org.amqp public interface LongString { - function length():int; - - /** - * Get the content stream. - * Repeated calls to this function return the same stream, - * which may not support rewind. - * @return An input stream the reads the content - * @throws IOException - */ function getStream():IDataInput; - - /** - * Get the content as a byte array. - * Repeated calls to this function return the same array. - * This function will fail if getContentLength() > Integer.MAX_VALUE - * throwing an IllegalStateException. - * @return the content as an array - * @throws IOException - */ function getBytes():ByteArray; - } -} \ No newline at end of file +} src/org/amqp/LongString.as @@ -52,12 +52,6 @@ package org.amqp.headers } } while (true); flags = acc; - /* - flags = new int[acc.size()]; - for (int i = 0; i < flags.length; i++) { - flags[i] = ((Integer) acc.get(i)).intValue(); - } - */ } /** src/org/amqp/headers/ContentHeaderPropertyReader.as @@ -110,10 +110,8 @@ package org.amqp.headers } public function ___writeLongstr(x:LongString):void { - outBytes.writeInt(x.length()); outBytes.writeBytes(x.getBytes(),0,0); - } /** Protected API - Writes a short integer value to the stream, if it's non-null */ @@ -125,9 +123,7 @@ package org.amqp.headers /** Protected API - Writes a short integer value to the stream, if it's non-null */ public function _writeShort(x:int):void { - outBytes.writeShort(x); - } /** Protected API - Writes an integer value to the stream, if it's non-null */ @@ -138,8 +134,7 @@ package org.amqp.headers } public function _writeLong(x:int):void { - - outBytes.writeInt(x); + outBytes.writeInt(x); } /** Protected API - Writes a long integer value to the stream, if it's non-null */ @@ -150,9 +145,7 @@ package org.amqp.headers } public function _writeLonglong(x:uint):void { - - outBytes.writeUnsignedInt(x); - + outBytes.writeUnsignedInt(x); } /** Protected API - Writes a table value to the stream, if it's non-null */ @@ -180,18 +173,6 @@ package org.amqp.headers _writeOctet(73); // 'I' _writeShort(value as int); } - /* - else if(value is BigDecimal) { - writeOctet(68); // 'D' - BigDecimal decimal = (BigDecimal)value; - writeOctet(decimal.scale()); - BigInteger unscaled = decimal.unscaledValue(); - if(unscaled.bitLength() > 32) //Integer.SIZE in Java 1.5 - throw new IllegalArgumentException - ("BigDecimal too large to be encoded"); - writeLong(decimal.unscaledValue().intValue()); - } - */ else if(value is Date) { _writeOctet(84);//'T' _writeTimestamp(value as Date); @@ -220,8 +201,7 @@ package org.amqp.headers } public function _writeOctet(x:int):void { - - outBytes.writeByte(x); + outBytes.writeByte(x); } /** Protected API - Writes a timestamp value to the stream, if it's non-null */ @@ -232,9 +212,7 @@ package org.amqp.headers } public function _writeTimestamp(x:Date):void { - - outBytes.writeInt(x.time / 1000); - + outBytes.writeInt(x.time / 1000); } } } src/org/amqp/headers/ContentHeaderPropertyWriter.as @@ -129,16 +129,12 @@ package org.amqp.impl public function onOpenOk(event:ProtocolEvent):void { var openOk:OpenOk = event.command.method as OpenOk; - // Maybe do something with the knownhosts? - //openOk.knownhosts; if (state == STATE_CLOSE_REQUESTED) { close(); } else { state = STATE_OPEN; - //dispatchAfterOpenEvent(); } - // Call the lifecycle event handlers session.emitLifecyleEvent(); } @@ -151,6 +147,5 @@ package org.amqp.impl close.methodid = 0; session.rpc(new Command(close), onCloseOk); } - } } src/org/amqp/impl/ConnectionStateHandler.as @@ -41,12 +41,6 @@ package org.amqp.impl private var dispatcher:EventDispatcher = new EventDispatcher(); - /** - * I'm not too happy about this new RPC queue - the whole queueing - * thing needs a complete refactoring so that RPCs are executed serially - * but also so that different intra-class RPC order is guaranteed. - */ - //protected var rpcQueue:PriorityQueue = new PriorityQueue(QUEUE_SIZE); protected var rpcQueue:ArrayedQueue = new ArrayedQueue(QUEUE_SIZE); private var lifecycleHandlers:Array = new Array(); @@ -115,7 +109,6 @@ package org.amqp.impl commandReceiver.addEventListener(method.getAltResponse(), fun); } sendCommand(cmd, fun); - //trace("RPC top half: " + cmd.method); } private function rpcBottomHalf():void { @@ -123,7 +116,6 @@ package org.amqp.impl rpcQueue.dequeue(); if (!rpcQueue.isEmpty()) { var o:Object = rpcQueue.peek(); - //trace("RPC bottom half: " + o.command.method); send(o.command); } } src/org/amqp/impl/SessionImpl.as @@ -55,15 +55,10 @@ package org.amqp.methods } public static function _readLongstr(input:IDataInput):LongString { - //final long contentLength = unsignedExtend(in.readInt()); var contentLength:int = input.readInt(); if(contentLength < int.MAX_VALUE) { - //final byte [] buffer = new byte[(int)contentLength]; - //in.readFully(buffer); - var buf:ByteArray = new ByteArray(); input.readBytes(buf, 0, contentLength); - return new ByteArrayLongString(buf); } else { @@ -73,8 +68,6 @@ package org.amqp.methods public static function _readShortstr(input:IDataInput):String { var length:int = input.readUnsignedByte(); - //var length3:uint = input.readUnsignedInt(); - //var length4:int = input.readShort(); return input.readUTFBytes(length); } @@ -114,7 +107,6 @@ package org.amqp.methods bits = input.readUnsignedByte(); bit = 0x01; } - var result:Boolean = (bits&bit) != 0; bit = bit << 1; return result; @@ -133,32 +125,22 @@ package org.amqp.methods public static function _readTable(input:IDataInput):Map { var table:Map = new HashMap(); - var tableLength:int = input.readInt();//unsignedExtend(in.readInt()); + var tableLength:int = input.readInt(); var tableIn:ByteArray = new ByteArray(); input.readBytes(tableIn, 0, tableLength); var value:Object = null; while(tableIn.bytesAvailable > 0) { - var name:String = _readShortstr(tableIn); var type:uint = tableIn.readUnsignedByte(); switch(type) { case 83 : //'S' value = _readLongstr(tableIn); - //value = _readShortstr(tableIn); break; case 73: //'I' value = tableIn.readInt(); break; - /* - case 68: //'D': - var scale:int = tableIn.readUnsignedByte(); - byte [] unscaled = new byte[4]; - tableIn.readFully(unscaled); - value = new BigDecimal(new BigInteger(unscaled), scale); - break; - */ case 84: //'T': value = _readTimestamp(tableIn); break; @@ -187,7 +169,6 @@ package org.amqp.methods var date:Date = new Date(); date.setTime(input.readInt() * 1000) return date; - //return new Date(in.readLong() * 1000); } /** Public API - reads an timestamp argument. */ @@ -195,7 +176,5 @@ package org.amqp.methods clearBits(); return _readTimestamp(input); } - - } } src/org/amqp/methods/MethodArgumentReader.as @@ -63,11 +63,8 @@ package org.amqp.methods /** Public API - encodes a short string argument. */ public final function writeShortstr(str:String):void { bitflush(); - //byte [] bytes = str.getBytes("utf-8"); - var buf:ByteArray = new ByteArray(); buf.writeUTFBytes(str); - output.writeByte(buf.length); output.writeBytes(buf, 0, 0); } @@ -82,7 +79,6 @@ package org.amqp.methods /** Public API - encodes a long string argument from a String. */ public final function writeString(str:String):void { bitflush(); - //byte [] bytes = str.getBytes("utf-8"); writeLong(str.length); output.writeUTFBytes(str); } @@ -96,18 +92,12 @@ package org.amqp.methods /** Public API - encodes an integer argument. */ public final function writeLong(l:int):void { bitflush(); - // java's arithmetic on this type is signed, however its - // reasonable to use ints to represent the unsigned long - // type - for values < Integer.MAX_VALUE everything works - // as expected output.writeInt(l); } /** Public API - encodes a long integer argument. */ public final function writeLonglong(ll:int):void { throw new Error("No longs in Actionscript"); - //bitflush(); - //output.writeInt(ll); } /** Public API - encodes a boolean/bit argument. */ @@ -127,10 +117,8 @@ package org.amqp.methods /** Public API - encodes a table argument. */ public final function writeTable(table:Map):void { - bitflush(); if (table == null) { - // Convenience. output.writeInt(0); } else { output.writeInt( FrameHelper.tableSize(table) ); @@ -138,7 +126,6 @@ package org.amqp.methods for (var key:String in table) { writeShortstr(key); var value:Object = table.getValue(key); - if(value is String) { writeOctet(83); // 'S' writeString(value as String); @@ -151,18 +138,6 @@ package org.amqp.methods writeOctet(73); // 'I' writeShort(value as int); } - /* - else if(value is BigDecimal) { - writeOctet(68); // 'D' - BigDecimal decimal = (BigDecimal)value; - writeOctet(decimal.scale()); - BigInteger unscaled = decimal.unscaledValue(); - if(unscaled.bitLength() > 32) //Integer.SIZE in Java 1.5 - throw new IllegalArgumentException - ("BigDecimal too large to be encoded"); - writeLong(decimal.unscaledValue().intValue()); - } - */ else if(value is Date) { writeOctet(84);//'T' writeTimestamp(value as Date); @@ -192,7 +167,6 @@ package org.amqp.methods /** Public API - encodes a timestamp argument. */ public final function writeTimestamp(timestamp:Date):void { - // AMQP uses POSIX time_t which is in seconds since the epoc writeLonglong( timestamp.valueOf() / 1000); } @@ -202,7 +176,6 @@ package org.amqp.methods */ public function flush():void { bitflush(); - //output.flush(); } } } src/org/amqp/methods/MethodArgumentWriter.as @@ -61,11 +61,8 @@ package org.amqp.util /** Public API - encodes a short string argument. */ public final function writeShortstr(str:String):void { bitflush(); - //byte [] bytes = str.getBytes("utf-8"); - var buf:ByteArray = new ByteArray(); buf.writeUTFBytes(str); - output.writeByte(buf.length); output.writeBytes(buf, 0, 0); } @@ -80,7 +77,6 @@ package org.amqp.util /** Public API - encodes a long string argument from a String. */ public final function writeString(str:String):void { bitflush(); - //byte [] bytes = str.getBytes("utf-8"); writeLong(str.length); output.writeUTFBytes(str); } @@ -94,18 +90,12 @@ package org.amqp.util /** Public API - encodes an integer argument. */ public final function writeLong(l:int):void { bitflush(); - // java's arithmetic on this type is signed, however its - // reasonable to use ints to represent the unsigned long - // type - for values < Integer.MAX_VALUE everything works - // as expected output.writeInt(l); } /** Public API - encodes a long integer argument. */ public final function writeLonglong(ll:int):void { throw new Error("No longs in Actionscript"); - //bitflush(); - //output.writeInt(ll); } /** Public API - encodes a boolean/bit argument. */ @@ -150,18 +140,6 @@ package org.amqp.util writeOctet(73); // 'I' writeShort(value as int); } - /* - else if(value is BigDecimal) { - writeOctet(68); // 'D' - BigDecimal decimal = (BigDecimal)value; - writeOctet(decimal.scale()); - BigInteger unscaled = decimal.unscaledValue(); - if(unscaled.bitLength() > 32) //Integer.SIZE in Java 1.5 - throw new IllegalArgumentException - ("BigDecimal too large to be encoded"); - writeLong(decimal.unscaledValue().intValue()); - } - */ else if(value is Date) { writeOctet(84);//'T' writeTimestamp(value as Date); @@ -191,7 +169,6 @@ package org.amqp.util /** Public API - encodes a timestamp argument. */ public final function writeTimestamp(timestamp:Date):void { - // AMQP uses POSIX time_t which is in seconds since the epoc writeLonglong( timestamp.valueOf() / 1000); } @@ -201,7 +178,6 @@ package org.amqp.util */ public function flush():void { bitflush(); - //output.flush(); } } } src/org/amqp/util/BinaryGenerator.as @@ -23,8 +23,7 @@ package org.amqp.util public class IOUtils { public static function copy(input:ByteArray, output:IDataOutput):void { - output.writeBytes(input); } } -} \ No newline at end of file +} src/org/amqp/util/IOUtils.as @@ -53,11 +53,6 @@ package org.amqp.test public function afterOpen():void { openChannel(teardownExchange); - /* - var timer:Timer = new Timer(DELAY, 1); - timer.addEventListener(TimerEvent.TIMER, teardownExchange); - timer.start(); - */ } public function teardownExchange(event:ProtocolEvent):void { @@ -75,8 +70,8 @@ package org.amqp.test trace("whoCares called"); }; - sessionHandler.rpc(new Command(queueDelete), whoCares);//addAsync(whoCares, TIMEOUT)); - sessionHandler.rpc(new Command(exchangeDelete), whoCares);//addAsync(whoCares, TIMEOUT)); + sessionHandler.rpc(new Command(queueDelete), whoCares); + sessionHandler.rpc(new Command(exchangeDelete), whoCares); var timer:Timer = new Timer(DELAY, 1); timer.addEventListener(TimerEvent.TIMER, closeSession); @@ -89,9 +84,10 @@ package org.amqp.test close.replytext = "Goodbye"; var fun:Function = function(event:ProtocolEvent):void { trace("Channel closed"); - closeConnection(); + // TODO Look into how the connection shutdown is being handled + //closeConnection(); }; - sessionHandler.rpc(new Command(close), fun);//addAsync(fun, TIMEOUT)); + sessionHandler.rpc(new Command(close), fun); } public function closeConnection():void { @@ -101,7 +97,7 @@ package org.amqp.test var fun:Function = function(event:ProtocolEvent):void { trace("Connection closed"); }; - sessionHandler.rpc(new Command(close), fun);//addAsync(fun, TIMEOUT)); + sessionHandler.rpc(new Command(close), fun); } } test/src/org/amqp/test/LifecycleTest.as 718487dc9fabd00263f33ee1af753f3096859f92 Ben Hood 0x6e6562@gmail.com http://github.com/0x6e6562/as3-amqp/commit/339f87ec3305d54c23256dd6a0d0d9b7d902e99f 339f87ec3305d54c23256dd6a0d0d9b7d902e99f 2008-11-15T09:07:57-08:00 2008-11-15T09:07:57-08:00 Removed all lot of dead comments 59be2448fc242cdaa80ff2746bb3af88abb33243 Ben Hood 0x6e6562@gmail.com