@@ -43,6 +43,8 @@ package org.amqp public var sessionManager:SessionManager; public var frameMax:int = 0; + private var frame:Frame = new Frame(); + public function Connection(state:ConnectionParameters) { connectionParams = state; var stateHandler:ConnectionStateHandler = new ConnectionStateHandler(state); @@ -146,26 +148,28 @@ package org.amqp **/ public function onSocketData(event:Event):void { while (delegate.isConnected() && delegate.bytesAvailable > 0) { - var frame:Frame = parseFrame(delegate); - maybeSendHeartbeat(); - if (frame != null) { - if (frame.type == AMQP.FRAME_HEARTBEAT) { - // just ignore this for now - } else if (frame.channel == 0) { - session0.handleFrame(frame); - } else { - var session:Session = sessionManager.lookup(frame.channel); - session.handleFrame(frame); - } - } else { - handleSocketTimeout(); - } + var f:Frame = parseFrame(delegate); + if (f == null) return; + if (f.type == AMQP.FRAME_HEARTBEAT) { + // just ignore this for now + } else if (f.channel == 0) { + session0.handleFrame(f); + } else { + var session:Session = sessionManager.lookup(f.channel); + session.handleFrame(f); + } } + maybeSendHeartbeat(); } private function parseFrame(delegate:IODelegate):Frame { - var frame:Frame = new Frame(); - return frame.readFrom(delegate) ? frame : null; + frame.readFrom(delegate); + if (frame.complete) { + var f:Frame = frame; + frame = new Frame(); + return f; + } + return null; } public function sendFrame(frame:Frame):void { src/org/amqp/Connection.as @@ -25,10 +25,13 @@ package org.amqp public class Frame { + public var complete:Boolean = false; + public var headerComplete:Boolean = false; public var type:uint; public var channel:int; protected var payload:ByteArray; protected var accumulator:ByteArray; + protected var payloadSize:int public function Frame() { this.payload = new ByteArray(); @@ -37,23 +40,36 @@ package org.amqp public function readFrom(input:IDataInput):Boolean { + if (!headerComplete) { + // wait till we've got enough data in the buffer to read the header + if (input.bytesAvailable < 8) return false; + + type = input.readUnsignedByte(); if (type == 'A' as uint) { /* Probably an AMQP.... header indicating a version mismatch. */ /* Otherwise meaningless, so try to read the version, and - * throw an exception, whether we read the version okay or + we * throw an exception, whether we read the version okay or * not. */ protocolVersionMismatch(input); } channel = input.readUnsignedShort(); - var payloadSize:int = input.readInt(); + payloadSize = input.readInt(); - if (payloadSize > 0) { - payload = new ByteArray(); - input.readBytes(payload, 0, payloadSize); - } + if (payloadSize > 0) { + payload = new ByteArray(); + } + + headerComplete = true; + } + + + if (payloadSize > 0 && payload.length < payloadSize) { + input.readBytes(payload, payload.length, Math.min(input.bytesAvailable,payloadSize - payload.length)); + if (payload.length < payloadSize) return false; + } accumulator = null; @@ -62,9 +78,9 @@ package org.amqp if (frameEndMarker != AMQP.FRAME_END) { throw new MalformedFrameError("Bad frame end marker: " + frameEndMarker); } - - return true; - } + complete = true; + return true; + } private function protocolVersionMismatch(input:IDataInput):void { src/org/amqp/Frame.as 339f87ec3305d54c23256dd6a0d0d9b7d902e99f Joost Diepenmaat joost@zeekat.nl http://github.com/0x6e6562/as3-amqp/commit/8a553b711f71b2a40a7ca52d939fbd491acb2b91 8a553b711f71b2a40a7ca52d939fbd491acb2b91 2009-08-28T05:06:33-07:00 2009-08-28T05:06:33-07:00 implemented correct buffering dc7c653e4eed9ad273adabd63ccfdcbc6a175a83 Joost Diepenmaat joost@zeekat.nl