src/org/amqp/LifecycleEventHandler.as @@ -19,7 +19,7 @@ <!-- Compile code generator --> - <target name="codeGenerator" depends="properties"> + <target name="codegen" depends="properties"> <mkdir dir="${build.dir}"/> <copy todir="${build.dir}"> <fileset dir="codegen/specs"/> @@ -47,7 +47,7 @@ <!-- Compile all of the classes under the "src" tree into a .swc file --> - <target name="lib" depends="codeGenerator"> + <target name="lib" depends="codegen"> <java jar="${compc}" fork="true" failonerror="true"> <classpath> <fileset dir="${flexsdk.dir}/lib"> build.xml @@ -295,10 +295,13 @@ public class CodeGenerator { if (synchronous) { String response = ap4.evalXPathToString(); Method r = bindMethod(nav, response, clazz); + r.setBottomHalf(true); mangleResponseMethodName(r); method.setResponse(r); } + method.setBottomHalf(synchronous && methodName.contains("-")); + methods.add(method); } codegen/src/org/amqp/as3/codegen/CodeGenerator.java @@ -16,12 +16,22 @@ public class Method { private boolean hasAltResponse = false; + private boolean bottomHalf = false; + private boolean synchronous; private List<Field> fields; private Method response; + public boolean isBottomHalf() { + return bottomHalf; + } + + public void setBottomHalf(boolean bottomHalf) { + this.bottomHalf = bottomHalf; + } + public boolean isHasAltResponse() { return hasAltResponse; } codegen/src/org/amqp/as3/codegen/model/Method.java @@ -120,6 +120,11 @@ override public function getAltResponse():Method { } <endif> + +override public function isBottomHalf():Boolean { + return <meth.bottomHalf>; +} + override public function getClassId():int{ return <meth.amqpClass.index>; } @@ -135,8 +140,14 @@ override public function writeArgumentsTo(writer:MethodArgumentWriter):void { override public function readArgumentsFrom(reader:MethodArgumentReader):void { <meth.fields:readArgs(); separator="\n"> } + +public function dump():void { + trace("-------- <method.amqpClass.name>.<method.name> --------"); + <meth.fields:dump(); separator="\n"> +} >> +dump() ::= "trace(\"<it.name>: {\" + _<it.name> + \"}\");" readArgs() ::= "_<it.name> = reader.<readMap.(it.type)>();" codegen/template/AMQP_Method.as.stg @@ -45,4 +45,4 @@ package org.amqp dispatcher.dispatchEvent(new ProtocolEvent(cmd)); } } -} \ No newline at end of file +} src/org/amqp/BaseCommandReceiver.as @@ -47,4 +47,4 @@ package org.amqp function removeEventListener(method:Method, callback:Function):void; } -} \ No newline at end of file +} src/org/amqp/CommandReceiver.as @@ -48,7 +48,6 @@ package org.amqp var stateHandler:ConnectionStateHandler = new ConnectionStateHandler(state); session0 = new SessionImpl(this, 0, stateHandler); - session0.addEventListener(new CloseOk(), afterGracefulClose); stateHandler.registerWithSession(session0); sessionManager = new SessionManager(this); src/org/amqp/Connection.as @@ -35,6 +35,10 @@ package org.amqp return false; } + public function isBottomHalf():Boolean { + return false; + } + public function getResponse():Method { return null; } src/org/amqp/Method.as @@ -20,16 +20,16 @@ package org.amqp public interface Session { function closeGracefully():void; - function forceClose():void; - - function sendCommand(c:Command):void; - + /** + * I think sendCommand/2 can be refactored down to sendCommand/1 + * and hence be made more intuitive + */ + function sendCommand(c:Command, fun:Function = null):void; function handleFrame(frame:Frame):void; - - function addEventListener(method:Method, fun:Function):void; - - function removeEventListener(method:Method, fun:Function):void; + function rpc(c:Command, fun:Function):void; + function registerLifecycleHandler(handler:LifecycleEventHandler):void + function emitLifecyleEvent():void; } -} \ No newline at end of file +} src/org/amqp/Session.as @@ -75,4 +75,4 @@ package org.amqp sessions.clear(); } } -} \ No newline at end of file +} src/org/amqp/SessionManager.as @@ -25,6 +25,7 @@ package org.amqp.impl import org.amqp.BaseCommandReceiver; import org.amqp.Command; import org.amqp.ConnectionParameters; + import org.amqp.LifecycleEventHandler; import org.amqp.ProtocolEvent; import org.amqp.methods.connection.Close; import org.amqp.methods.connection.CloseOk; @@ -61,8 +62,6 @@ package org.amqp.impl public function ConnectionStateHandler(params:ConnectionParameters){ connectionParams = params; - addEventListener(new OpenOk(), onOpenOk); - addEventListener(new CloseOk(), onCloseOk); addEventListener(new Start(), onStart); addEventListener(new Tune(), onTune); } @@ -82,7 +81,6 @@ package org.amqp.impl public function onCloseOk(cmd:Command):void { var closeOk:CloseOk = cmd.method as CloseOk; - //dispatchAfterCloseEvent(); state = STATE_CLOSED; } @@ -112,7 +110,7 @@ package org.amqp.impl startOk.response = new ByteArrayLongString(buf); startOk.locale = "en_US"; - send(new Command(startOk)); + session.sendCommand(new Command(startOk)); } public function onTune(event:ProtocolEvent):void { @@ -121,12 +119,12 @@ package org.amqp.impl tuneOk.channelmax = tune.channelmax; tuneOk.framemax = tune.framemax; tuneOk.heartbeat = tune.heartbeat; - send(new Command(tuneOk)); + session.sendCommand(new Command(tuneOk)); var open:Open = new Open(); open.virtualhost = connectionParams.vhostpath; open.capabilities = ""; open.insist = false; - send(new Command(open)); + session.rpc(new Command(open), onOpenOk); } public function onOpenOk(event:ProtocolEvent):void { @@ -140,6 +138,9 @@ package org.amqp.impl state = STATE_OPEN; //dispatchAfterOpenEvent(); } + + // Call the lifecycle event handlers + session.emitLifecyleEvent(); } private function close():void { @@ -148,11 +149,8 @@ package org.amqp.impl close.replytext = "Goodbye"; close.classid = 0; close.methodid = 0; - send(new Command(close)); + session.rpc(new Command(close), onCloseOk); } - private function send(cmd:Command):void { - session.sendCommand(cmd); - } } } src/org/amqp/impl/ConnectionStateHandler.as @@ -17,20 +17,40 @@ **/ package org.amqp.impl { + import de.polygonal.ds.ArrayedQueue; + import de.polygonal.ds.PriorityQueue; + + import flash.events.EventDispatcher; + import org.amqp.Command; import org.amqp.CommandReceiver; import org.amqp.Connection; import org.amqp.Frame; + import org.amqp.LifecycleEventHandler; import org.amqp.Method; import org.amqp.Session; public class SessionImpl implements Session { + protected var QUEUE_SIZE:int = 100; + private var connection:Connection; private var channel:int; private var commandReceiver:CommandReceiver; private var currentCommand:Command; + 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(); + public function SessionImpl(con:Connection, ch:int, receiver:CommandReceiver) { connection = con; channel = ch; @@ -43,15 +63,72 @@ package org.amqp.impl } currentCommand.handleFrame(frame); if (currentCommand.isComplete()) { + /** + * The idea is that this callback will always be invoked when a command is + * to be processed by the session handler, so it can kick off the dequeuing + * of any pending RPCs in addition to dispatching to the target callback handler, + * which is implemented in the super class. + */ commandReceiver.receive(currentCommand); + if (currentCommand.method.isBottomHalf()) { + rpcBottomHalf(); + } currentCommand = new Command(); } } - public function sendCommand(cmd:Command):void { + public function registerLifecycleHandler(handler:LifecycleEventHandler):void { + lifecycleHandlers.push(handler); + } + + public function emitLifecyleEvent():void { + for (var i:uint = 0; i < lifecycleHandlers.length; i++) { + (lifecycleHandlers[i] as LifecycleEventHandler).afterOpen(); + } + } + + /** + * The logic behind this is that a non-null fun signifies an RPC, + * if the fun is null, then it is an asynchronous command. + */ + public function sendCommand(cmd:Command, fun:Function = null):void { + + if (null != fun) { + if (rpcQueue.isEmpty()) { + send(cmd); + } + rpcQueue.enqueue({command:cmd,callback:fun}); + } + else { + send(cmd); + } + } + + private function send(cmd:Command):void { cmd.transmit(channel, connection); } + public function rpc(cmd:Command, fun:Function):void { + var method:Method = cmd.method; + commandReceiver.addEventListener(method.getResponse(), fun); + if (null != method.getAltResponse()) { + commandReceiver.addEventListener(method.getAltResponse(), fun); + } + sendCommand(cmd, fun); + //trace("RPC top half: " + cmd.method); + } + + private function rpcBottomHalf():void { + if (!rpcQueue.isEmpty()) { + rpcQueue.dequeue(); + if (!rpcQueue.isEmpty()) { + var o:Object = rpcQueue.peek(); + //trace("RPC bottom half: " + o.command.method); + send(o.command); + } + } + } + public function closeGracefully():void { commandReceiver.closeGracefully(); } @@ -60,12 +137,5 @@ package org.amqp.impl commandReceiver.forceClose(); } - public function addEventListener(method:Method, fun:Function):void { - commandReceiver.addEventListener(method, fun); - } - - public function removeEventListener(method:Method, fun:Function):void { - commandReceiver.removeEventListener(method, fun); - } } -} \ No newline at end of file +} src/org/amqp/impl/SessionImpl.as @@ -51,13 +51,12 @@ package org.amqp.impl protected var state:int = STATE_CONNECTION; protected var QUEUE_SIZE:int = 100; - protected var commandQueue:PriorityQueue = new PriorityQueue(QUEUE_SIZE); + protected var pendingConsumers:ArrayedQueue = new ArrayedQueue(QUEUE_SIZE); protected var consumers:HashMap = new HashMap(); public function SessionStateHandler(){ - addEventListener(new OpenOk(), onOpenOk); - addEventListener(new CloseOk(), onCloseOk); + // TODO Look into whether this is really necessary addEventListener(new Deliver(), onDeliver); } @@ -72,12 +71,7 @@ package org.amqp.impl } public function rpc(cmd:Command, fun:Function):void { - var method:Method = cmd.method; - addEventListener(method.getResponse(), fun); - if (null != method.getAltResponse()) { - addEventListener(method.getAltResponse(), fun); - } - dispatch(cmd); + session.rpc(cmd, fun); } private function cancelRpcHandler(method:Method, fun:Function):void { @@ -85,22 +79,7 @@ package org.amqp.impl } public function dispatch(cmd:Command):void { - switch(state) { - case STATE_CLOSED: { stateError(cmd.method.getClassId()); } - case STATE_CONNECTION: { - if (cmd.method.getClassId() > STATE_CHANNEL) { - enqueueCommand(cmd); - } - else { - sendCommand(cmd); - } - break; - } - default: { - flushQueue(); - sendCommand(cmd); - } - } + session.sendCommand(cmd, null); } public function register(consume:Consume, consumer:BasicConsumer):void{ @@ -144,7 +123,6 @@ package org.amqp.impl public function onOpenOk(event:ProtocolEvent):void { transition(STATE_OPEN); - flushQueue(-1); } /** @@ -155,37 +133,6 @@ package org.amqp.impl } /** - * Enqueues a command in order of ascending class id. - **/ - private function enqueueCommand(cmd:Command):void { - commandQueue.enqueue(cmd); - } - - private function flushQueue(limit:int = -1):void { - var cmd:Command = commandQueue.dequeue() as Command; - if (null == cmd) return; - var classId:int = cmd.method.getClassId(); - - if (limit > -1) { - if (classId > limit) { - commandQueue.enqueue(cmd); - } - else { - sendCommand(cmd); - flushQueue(limit); - } - } - else { - sendCommand(cmd); - flushQueue(); - } - } - - private function sendCommand(cmd:Command):void { - session.sendCommand(cmd); - } - - /** * Cheap hack of an FSM **/ private function transition(newState:int):void { src/org/amqp/impl/SessionStateHandler.as @@ -21,23 +21,22 @@ package org.amqp.patterns.impl import org.amqp.Command; import org.amqp.Connection; + import org.amqp.LifecycleEventHandler; import org.amqp.ProtocolEvent; import org.amqp.headers.BasicProperties; import org.amqp.impl.SessionStateHandler; import org.amqp.methods.basic.Publish; import org.amqp.methods.channel.Open; - import org.amqp.methods.connection.OpenOk; import org.amqp.methods.exchange.Declare; import org.amqp.methods.queue.Bind; import org.amqp.methods.queue.DeclareOk; import org.amqp.patterns.Serializer; import org.amqp.util.Properties; - public class AbstractDelegate + public class AbstractDelegate implements LifecycleEventHandler { public var exchange:String; public var exchangeType:String; - public var realm:String; public var connection:Connection; public var serializer:Serializer; @@ -47,14 +46,16 @@ package org.amqp.patterns.impl public function AbstractDelegate(c:Connection) { connection = c; connection.start(); - connection.baseSession.addEventListener(new OpenOk(), openChannel); + connection.baseSession.registerLifecycleHandler(this); } - protected function openChannel(event:ProtocolEvent):void { + public function afterOpen():void { + openChannel(); + } + + protected function openChannel():void { sessionHandler = connection.sessionManager.create(); - var open:Open = new Open(); - sessionHandler.dispatch(new Command(open)); - sessionHandler.addEventListener(new org.amqp.methods.channel.OpenOk(), onChannelOpenOk); + sessionHandler.rpc(new Command(new Open()), onChannelOpenOk); } protected function publish(x:String, routing_key:String, data:ByteArray, properties:BasicProperties = null):void { @@ -70,13 +71,13 @@ package org.amqp.patterns.impl var exchange:org.amqp.methods.exchange.Declare = new org.amqp.methods.exchange.Declare(); exchange.exchange = x; exchange.type = type; - sessionHandler.dispatch(new Command(exchange)); + sessionHandler.rpc(new Command(exchange), onExchangeDeclareOk); } protected function declareQueue(q:String):void { var queue:org.amqp.methods.queue.Declare = new org.amqp.methods.queue.Declare(); queue.queue = q; - sessionHandler.dispatch(new Command(queue)); + sessionHandler.rpc(new Command(queue), onQueueDeclareOk); } protected function bindQueue(x:String,q:String, key:String):void { @@ -84,12 +85,12 @@ package org.amqp.patterns.impl bind.exchange = x; bind.queue = q; bind.routingkey = key; - sessionHandler.dispatch(new Command(bind)); + + sessionHandler.rpc(new Command(bind), onQueueBindOk); } protected function setupReplyQueue():void { declareQueue(""); - sessionHandler.addEventListener(new DeclareOk(),onQueueDeclareOk); } protected function getReplyQueue(event:ProtocolEvent):String { @@ -108,5 +109,15 @@ package org.amqp.patterns.impl **/ protected function onQueueDeclareOk(event:ProtocolEvent):void {} + /** + * This should be overriden by specializing classes + **/ + protected function onExchangeDeclareOk(event:ProtocolEvent):void {} + + /** + * This should be overriden by specializing classes + **/ + protected function onQueueBindOk(event:ProtocolEvent):void {} + } } src/org/amqp/patterns/impl/AbstractDelegate.as @@ -47,7 +47,6 @@ package org.amqp.patterns.impl override protected function onChannelOpenOk(event:ProtocolEvent):void { declareExchange(exchange,exchangeType); declareQueue(""); - sessionHandler.addEventListener(new DeclareOk(),onQueueDeclareOk); } override protected function onQueueDeclareOk(event:ProtocolEvent):void { @@ -88,4 +87,4 @@ package org.amqp.patterns.impl } } -} \ No newline at end of file +} src/org/amqp/patterns/impl/RpcServer.as @@ -18,10 +18,10 @@ package org.amqp.patterns.impl { import com.ericfeminella.utils.HashMap; - + import flash.events.EventDispatcher; import flash.utils.ByteArray; - + import org.amqp.BasicConsumer; import org.amqp.Command; import org.amqp.Connection; @@ -65,12 +65,7 @@ package org.amqp.patterns.impl public function unsubscribe(key:String):void { var cancel:Cancel = new Cancel(); var topic:* = topics.getValue(key); - - cancel.consumertag = topic.consumerTag; - sessionHandler.dispatch(new Command(cancel)); - sessionHandler.addEventListener(new org.amqp.methods.basic.CancelOk, onCancelOk); - - dispatcher.removeEventListener(key, topic.callback); + sessionHandler.unregister(topic.consumerTag); topics.remove(key); } @@ -80,7 +75,7 @@ package org.amqp.patterns.impl consume.noack = true; consume.consumertag = replyQueue + ":" + o; sessionHandler.register(consume, this); - + bindQueue(exchange, replyQueue, o); } @@ -102,13 +97,13 @@ package org.amqp.patterns.impl } public function onConsumeOk(tag:String):void { - var key:String = tag.split(":")[1]; - var topic:* = topics.getValue(key); - - topic.consumerTag = tag; - topics.put(key, topic); - - dispatcher.addEventListener(key, topic.callback); + var key:String = tag.split(":")[1]; + var topic:* = topics.getValue(key); + + topic.consumerTag = tag; + topics.put(key, topic); + + dispatcher.addEventListener(key, topic.callback); } public function onCancelOk(tag:String):void {} @@ -120,4 +115,4 @@ package org.amqp.patterns.impl dispatcher.dispatchEvent(new CorrelatedMessageEvent(properties.correlationid, result)); } } -} \ No newline at end of file +} src/org/amqp/patterns/impl/SubscribeClientImpl.as @@ -34,7 +34,6 @@ package org.amqp.test import org.amqp.methods.channel.Open; import org.amqp.methods.exchange.Declare; import org.amqp.methods.queue.Bind; - import org.amqp.methods.queue.BindOk; import org.amqp.util.Properties; public class AbstractTest extends TestCase @@ -83,7 +82,7 @@ package org.amqp.test timer.start(); } - protected function openChannel():void { + protected function openChannel(callback:Function):void { sessionHandler = sessionManager.create(); var open:Open = new Open(); @@ -100,16 +99,14 @@ package org.amqp.test bind.queue = q; bind.routingkey = bind_key; - var onBindOk:Function = function(event:ProtocolEvent):void{ - trace("onBindOk called"); + var devNull:Function = function(event:ProtocolEvent):void{ + trace("devNull called for " + event.command.method); }; - sessionHandler.dispatch(new Command(open)); - sessionHandler.dispatch(new Command(exchange)); - sessionHandler.dispatch(new Command(queue)); - sessionHandler.dispatch(new Command(bind)); - - sessionHandler.addEventListener(new BindOk(), addAsync(onBindOk, TIMEOUT) ); + sessionHandler.rpc(new Command(open), devNull); + sessionHandler.rpc(new Command(exchange), devNull); + sessionHandler.rpc(new Command(queue), devNull); + sessionHandler.rpc(new Command(bind), callback);//addAsync(callback, TIMEOUT)); } } } test/src/org/amqp/test/AbstractTest.as @@ -1,17 +1,19 @@ package org.amqp.test { - import flash.events.Event; import flash.utils.ByteArray; import flexunit.framework.TestSuite; import org.amqp.Command; + import org.amqp.LifecycleEventHandler; import org.amqp.ProtocolEvent; import org.amqp.methods.basic.Get; - import org.amqp.methods.connection.OpenOk; - public class GetTest extends AbstractTest + public class GetTest extends AbstractTest implements LifecycleEventHandler { + + private const testString:String = "hello, world"; + public function GetTest(methodName:String=null) { super(methodName); @@ -23,15 +25,18 @@ package org.amqp.test return myTS; } + public function afterOpen():void { + openChannel(runGetTest); + } + public function testGet():void { connection.start(); - baseSession.addEventListener(new OpenOk(), addAsync(runGetTest, TIMEOUT) ); + connection.baseSession.registerLifecycleHandler(this); } - public function runGetTest(event:Event):void { - openChannel(); + public function runGetTest(event:ProtocolEvent):void { var data:ByteArray = new ByteArray(); - data.writeUTF("hello, world"); + data.writeUTF(testString); publish(data); var _get:Get = new Get(); _get.queue = q; @@ -43,8 +48,10 @@ package org.amqp.test var data:ByteArray = event.command.content; data.position = 0; if (data.bytesAvailable > 0) { - trace("onGetOk ---> " + data.readUTF()); + var msg:String = data.readUTF(); + assertEquals(testString, msg); + trace("onGetOk ---> " + msg); } } } -} \ No newline at end of file +} test/src/org/amqp/test/GetTest.as @@ -24,18 +24,14 @@ package org.amqp.test import flexunit.framework.TestSuite; import org.amqp.Command; + import org.amqp.LifecycleEventHandler; import org.amqp.ProtocolEvent; import org.amqp.methods.channel.Close; - import org.amqp.methods.channel.Open; - import org.amqp.methods.connection.OpenOk; - import org.amqp.methods.exchange.Declare; + import org.amqp.methods.connection.Close; import org.amqp.methods.exchange.Delete; - import org.amqp.methods.queue.Bind; - import org.amqp.methods.queue.BindOk; - import org.amqp.methods.queue.Declare; import org.amqp.methods.queue.Delete; - public class LifecycleTest extends AbstractTest + public class LifecycleTest extends AbstractTest implements LifecycleEventHandler { public function LifecycleTest(methodName:String){ @@ -52,18 +48,20 @@ package org.amqp.test connection.start(); // Deliberately call this twice to test the state handling of the connection connection.start(); - baseSession.addEventListener(new OpenOk(), addAsync(afterBaseSessionOpened, TIMEOUT) ); + connection.baseSession.registerLifecycleHandler(this); } - public function afterBaseSessionOpened(event:Event):void { - openChannel(); + public function afterOpen():void { + openChannel(teardownExchange); + /* var timer:Timer = new Timer(DELAY, 1); timer.addEventListener(TimerEvent.TIMER, teardownExchange); timer.start(); - + */ } - public function teardownExchange(event:Event):void { + public function teardownExchange(event:ProtocolEvent):void { + trace("teardownExchange"); var queueDelete:org.amqp.methods.queue.Delete = new org.amqp.methods.queue.Delete(); queueDelete.queue = q; @@ -73,18 +71,12 @@ package org.amqp.test exchangeDelete.exchange = x; exchangeDelete.ifunused = false; - var onXDeleteOk:Function = function(event:ProtocolEvent):void{ - trace("onXDeleteOk called"); - }; - var onQDeleteOk:Function = function(event:ProtocolEvent):void{ - trace("onQDeleteOk called"); + var whoCares:Function = function(event:ProtocolEvent):void{ + trace("whoCares called"); }; - sessionHandler.addEventListener( new org.amqp.methods.queue.DeleteOk(), addAsync(onQDeleteOk, TIMEOUT)); - sessionHandler.addEventListener( new org.amqp.methods.exchange.DeleteOk(), addAsync(onXDeleteOk, TIMEOUT)); - - sessionHandler.dispatch(new Command(queueDelete)); - sessionHandler.dispatch(new Command(exchangeDelete)); + sessionHandler.rpc(new Command(queueDelete), whoCares);//addAsync(whoCares, TIMEOUT)); + sessionHandler.rpc(new Command(exchangeDelete), whoCares);//addAsync(whoCares, TIMEOUT)); var timer:Timer = new Timer(DELAY, 1); timer.addEventListener(TimerEvent.TIMER, closeSession); @@ -99,8 +91,7 @@ package org.amqp.test trace("Channel closed"); closeConnection(); }; - sessionHandler.addEventListener(new org.amqp.methods.channel.CloseOk(), addAsync(fun, TIMEOUT) ); - sessionHandler.dispatch(new Command(close)); + sessionHandler.rpc(new Command(close), fun);//addAsync(fun, TIMEOUT)); } public function closeConnection():void { @@ -110,8 +101,7 @@ package org.amqp.test var fun:Function = function(event:ProtocolEvent):void { trace("Connection closed"); }; - sessionHandler.addEventListener(new org.amqp.methods.connection.CloseOk(), addAsync(fun, TIMEOUT) ); - sessionHandler.dispatch(new Command(close)); + sessionHandler.rpc(new Command(close), fun);//addAsync(fun, TIMEOUT)); } } test/src/org/amqp/test/LifecycleTest.as @@ -17,7 +17,6 @@ **/ package org.amqp.test { - import flash.events.Event; import flash.events.TimerEvent; import flash.utils.ByteArray; import flash.utils.Timer; @@ -26,16 +25,15 @@ package org.amqp.test import org.amqp.BasicConsumer; import org.amqp.Command; + import org.amqp.LifecycleEventHandler; import org.amqp.ProtocolEvent; import org.amqp.headers.BasicProperties; import org.amqp.methods.basic.Consume; import org.amqp.methods.basic.Deliver; import org.amqp.methods.channel.Open; - import org.amqp.methods.connection.OpenOk; import org.amqp.methods.queue.Declare; - import org.amqp.methods.queue.DeclareOk; - public class PublishSubscribeTest extends AbstractTest implements BasicConsumer + public class PublishSubscribeTest extends AbstractTest implements BasicConsumer, LifecycleEventHandler { protected var consumerTag:String; @@ -55,26 +53,28 @@ package org.amqp.test public function testPublish():void { connection.start(); - baseSession.addEventListener(new OpenOk(), addAsync(runPublishTest, TIMEOUT) ); + connection.baseSession.registerLifecycleHandler(this); } - override protected function openChannel():void { + public function afterOpen():void { + openChannel(runPublishTest); + } + + override protected function openChannel(callback:Function):void { + + var whoCares:Function = function(event:ProtocolEvent):void{ + trace("whoCares called"); + }; + sessionHandler = sessionManager.create(); var open:Open = new Open(); var queue:org.amqp.methods.queue.Declare = new org.amqp.methods.queue.Declare(); queue.queue = q; - sessionHandler.dispatch(new Command(open)); - sessionHandler.dispatch(new Command(queue)); - - var onQueueDeclareOk:Function = function(event:ProtocolEvent):void{ - trace("onQueueDeclareOk called"); - }; - - sessionHandler.addEventListener(new DeclareOk(), addAsync(onQueueDeclareOk, TIMEOUT) ); + sessionHandler.rpc(new Command(open), whoCares); + sessionHandler.rpc(new Command(queue), callback); } - public function runPublishTest(event:Event):void { - openChannel(); + public function runPublishTest(event:ProtocolEvent):void { var consume:Consume = new Consume(); consume.queue = q; consume.noack = true; test/src/org/amqp/test/PublishSubscribeTest.as 53a18fcfa91a9b24bdcced67ca4eb172536c6d20 Ben Hood 0x6e6562@gmail.com http://github.com/0x6e6562/as3-amqp/commit/f15408036611dee724f051e8d3c05ad5aa35102d f15408036611dee724f051e8d3c05ad5aa35102d 2008-10-02T13:44:03-07:00 2008-10-02T13:44:03-07:00 Refactored RPC queueing 6844a1f5c91b6097d942d982bef771da711a7f45 Ben Hood 0x6e6562@gmail.com