[CONF] Apache Qpid > Port server to new interface

[confluence]

Port server to new interface Page edited by Aidan Skinner Overview Transport layer selection in o.a.q.server.Main Separation of AMQP processing from networking Implementation of NetworkDriver Implementation of ProtocolEngine Overview This component element is comprised of the following sub-tasks: Existing code needs to be refactored to remove dependence on MINA Existing code ported to use Current and proposed network interfaces Configuration to select a transport QpidByteBuffer These tasks are described in more detail below. Transport layer selection in o.a.q.server.Main Currently the broker's Main method contains setup code specific to MINA. Main needs to be modified to remove the MINA specific option (NIO, MultiIO, executor pool etc) processing from there. This should be replaced with creation of a Network Driver instance with configuration being picked up & applied in the same way as other subsystems are configured ie. using ServerConfiguration. Class Method Change Main bind Moved to NetworkDriver Main startup Network configuration moved to NetworkDriver, replaced with construction of NetworkDriver ServerConfiguration getNetworkConfiguration needs to parse config file and return networking configuration information Separation of AMQP processing from networking Class Changes AMQMinaProtocolSession decouple the AMQP semantics from the underlying networking. In particular, the threading model and idle timer related code should be moved to the NetworkDriver implementation. As a result, it should also be renamed to AMQProtocolEngine, with _8_0 and _0_9 delegates for handling protocol specific functionality. This class will implement the ProtocolEngine interface and become the central point for processing AMQP frames. AMQProtocolSessionMBean Needs updated to use AMQProtocolSession without relying on the underlying implementation AMQPFastProtocolHandler AMQP related functionality combined in the ProtocolEngine and networking related functionality moved to the NetworkDriver implementation. AMQPProtocolProvider AMQP related functionality combined in the ProtocolEngine and networking related functionality moved to the NetworkDriver implementation. QpidByteBuffer At a number of points in the code we rely on behaviour particular to MINA ByteBuffers, in particular the auto-expanding functionality. To avoid having to rewrite those parts at this stage we should import the MINA ByteBuffer class as QpidByteBuffer and continue to use it as is. Note that this functionality is not related to the network buffers, the QpidByteBuffers are the outputted data structure. The existing implementations will have their MINA or java.nio.ByteBuffers converted to QpidByteBuffers by the ProtocolEngine. AMQDecoder Should no longer extend CumulativeProtocolDecoder Event, Job Created by AMQProtocolEngine.recieved rather than AsynchPoolingFilter MinaNetworkDriver New class which implements the NetworkDriver interface using MINA HeartbeatConfiguration New class which provides configuration for Heartbeats Implementation of NetworkDriver public class MINANetworkDriver implements Sender<java.nio.ByteBuffer> { /** * Creates a concrete NetworkDriver which opens a connection to the specified address on the port * and starts the given ProtocolEngine instance using MINA */ NetworkDriver (int port, InetAddress destination, ProtocolEngine enginer, NetworkDriverConfiguration config); /** * Listens for incoming connections on the specified port and IPaddresses. Creates a new concrete * NetworkDriver for each incoming connection and creates a ProtocolEngine with the given factory */ NetworkDriver (int port, InetAddress[] addresses, ProtocolEngineFactory factory, NetworkDriverConfiguration config); void SocketAddress getRemoteAddress() void setMaxReadIdle(int idleTime) void setMaxWriteIdle(int idleTime) /** * Adds the data to an Event through AsynchWritePoolingFilter for writing to MINA */ void send(java.nio.ByteBuffer data) } Implementation of ProtocolEngine public class AMQProtocolEngine implements ProtocolEngine { /** * Processes data. If no concrete AMQProtocolEngine_0_N delegate exists at the time, it uses AMQDecoder to * decode the AMQ protocol initiation and creates the appropriate delegate. Otherwise it passes the data * onto the delegate for processing. */ void recieved(java.nio.ByteBuffer data) void setNetworkDriver (NetworkDriver driver) void SocketAddress getRemoteAddress() long getWrittenBytes() long getReadBytes() long close() void writerIdle() void readerIdle() } public class AMQProtocolEngine_0_N implements ProtocolEngine { /** * Decodes the data using AMQBlockDecoder_0_N and adds the Event to the current Job for processing. * When the Job is processed, the RecievedEvent calls frameRecieved which either closes the session if it's * ChannelCloseOk or calls the frame bodys handle() method */ void recieved(java.nio.ByteBuffer data) void setNetworkDriver (NetworkDriver driver) void SocketAddress getRemoteAddress() long getWrittenBytes() long getReadBytes() long close() void writerIdle() void readerIdle() } public class AMQProtocolEngineFactory implements ProtocolEngineFactory { // Returns a new instance of an AMQProtocolEngine ProtocolEngine newProtocolEngine() } Change Notification Preferences View Online | View Change | Add Comment --------------------------------------------------------------------- Apache Qpid - AMQP Messaging Implementation Project: http://qpid.apache.org Use/Interact: mailto:commits-subscr...@qpid.apache.org