之前的文章已经介绍了Beetle使用ProtoBuf.net进行对象序列化数据传输,这一章主要讲述Beetle如何使用FluorineFx和Flash进行AMF3通讯.其实现原理和使用ProtoBuf.net一样,扩展出一个MessageAdapter即可以.
MessageAdapter的实现如下:
public class MessageAdapter:IMessage { public object Message { get; set; } public static bool Send(TcpChannel channel,object message ) { MessageAdapter ma = new MessageAdapter(); ma.Message = message; return channel.Send(ma); } public void Load(BufferReader reader) { ByteArraySegment segment = ArrayPool.Pop(); int count = reader.ReadInt32(); reader.Read(count - 4, segment); using (System.IO.MemoryStream steram = new System.IO.MemoryStream(segment.Array, segment.Offset, segment.Count)) { FluorineFx.AMF3.ByteArray ba = new FluorineFx.AMF3.ByteArray(steram); ba.ObjectEncoding = FluorineFx.ObjectEncoding.AMF3; Message = ba.ReadObject(); } ArrayPool.Push(segment); } public void Save(BufferWriter writer) { ByteArraySegment segment = ArrayPool.Pop(); using (System.IO.MemoryStream steram = new System.IO.MemoryStream(segment.Array)) { FluorineFx.AMF3.ByteArray ba = new FluorineFx.AMF3.ByteArray(steram); ba.ObjectEncoding = FluorineFx.ObjectEncoding.AMF3; ba.WriteObject(Message); segment.SetInfo(0, (int)steram.Position); } writer.Write(segment.Count + 4); writer.Write(segment.Array,segment.Offset,segment.Count); ArrayPool.Push(segment); } public static ByteArrayPool ArrayPool = new ByteArrayPool(200, 1024 * 8); } 消息适配器实现比较简单在对象写入流的时候先把AMF3序列化对象流的长度+4写入头4个字节,然后再写入AMF3的数据流内容,从流中读取对象原来一样先把消息长度读取出来然后再读取AMF3数据流然后反序列化对象即可.
实现一个消息头描述长度的协议分析器:
public class HeadSizePackage:HeadSizeOfPackage { public HeadSizePackage() { } public HeadSizePackage(TcpChannel channel) : base(channel) { } protected override IMessage ReadMessageByType(BufferReader reader, out object typeTag) { typeTag = "MessageAdapter"; return new MessageAdapter(); } protected override void WriteMessageType(IMessage msg, BufferWriter writer) { } public override void MessageWrite(IMessage msg, BufferWriter writer) { msg.Save(writer); } public override IMessage MessageRead(BufferReader reader) { IMessage msg = null; object typeTag; msg = ReadMessageByType(reader, out typeTag); if (msg == null) throw NetTcpException.TypeNotFound(typeTag.ToString()); try { msg.Load(reader); } catch (Exception e) { NetTcpException err = NetTcpException.ObjectLoadError(typeTag.ToString(), e); throw err; } return msg; } } 这样一个消息扩展就完成具体生成的协议格式如下:
协议制定后就可似使用Beetle搭建基于AMF3的.net和flash数据传输。
首先是制定一个Tcp服务
TcpUtils.Setup(200, 1, 1); TcpServer server = new TcpServer(); server.ChannelConnected += OnConnected; server.ChannelDisposed += OnDisposed; server.Open(8340);
以上代码很简单初始化组件信息，构建一个TcpServer并绑定连接接入事件和连接断开事件；然后在所有IP的8340端绑定tcp服务。在连接接入的时候我们需要做些事情。
static void OnConnected(object sender, ChannelEventArgs e) { e.Channel.SetPackage ().ReceiveMessage = OnMessageReceive; e.Channel.ChannelError += OnError; e.Channel.BeginReceive(); Console.WriteLine("{0} connected!", e.Channel.EndPoint); } 在接入的事件里针对当前的Tcp通道设置一个协议分包器，并指定对应接收消息事件；Tcp通道相关信息设置完成后就可以调用BeginReceive()方法进入数据接收状态。接下来是消息处理事件的代码:
static void OnMessageReceive(PacketRecieveMessagerArgs e) { Beetle.FluorineFxAdapter.MessageAdapter adapter = (Beetle.FluorineFxAdapter.MessageAdapter)e.Message; if (adapter.Message is AMF3.Messages.Register) { OnRegister((AMF3.Messages.Register)adapter.Message,e.Channel); } else if (adapter.Message is AMF3.Messages.Get) { OnGet((AMF3.Messages.Get)adapter.Message, e.Channel); } else { } } static void OnRegister(AMF3.Messages.Register e, TcpChannel channel) { Console.WriteLine("{0} Register\t UserName:{1};PWD:{2};EMail:{3}", channel.EndPoint, e.UserName, e.PWD, e.EMail); } static void OnGet(AMF3.Messages.Get e, TcpChannel channel) { Console.WriteLine("{0} Get \t Customer:{1}", channel.EndPoint, e.CustomerID); AMF3.Messages.GetResponse response = new Messages.GetResponse(); for (int i = 0; i < 10; i++) { AMF3.Messages.Order order = new Messages.Order(); order.OrderID = 10248; order.CustomerID = "WILMK"; order.EmployeeID = 5; order.OrderDate = 629720352000000000; order.RequiredDate = 629744544000000000; order.ShipAddress = "59 rue de l'Abbaye"; order.ShipCity = "Reims"; order.ShipCountry = "France"; order.ShipName = "Vins et alcools Chevalier"; order.ShipPostalCode = "51100"; order.ShipRegion = "RJ"; response.Items.Add(order); } Beetle.FluorineFxAdapter.MessageAdapter.Send(channel, response); } 在这个例子中只处理了两种消息对象，分别是Register和Get;接收到Register只做了一个简单的输出，而在接收到Get则会返回一个Order列表。
Flash端实现
首先要实现协议分包器,由于Flash提供的Socket方法挺方便所以实现起来也是很容易的事情.
package{ import flash.net.Socket; import flash.utils.ByteArray; import flash.utils.Endian; import mx.graphics.shaderClasses.ExclusionShader; public class HeadSizeOfPackage { public function HeadSizeOfPackage() { } private var mMessageReceive:Function; //消息接收回调函数 public function get MessageReceive():Function { return mMessageReceive; } public function set MessageReceive(value:Function):void { mMessageReceive = value; } private var mReader:ByteArray = new ByteArray(); private var mWriter:ByteArray = new ByteArray(); private var mSize:int=0; //导入当前Socket接收的数据 public function Import(socket:Socket):void { socket.endian = Endian.LITTLE_ENDIAN; while(socket.bytesAvailable>0) { if(mSize==0) { mSize= socket.readInt()-4; mReader.clear(); } if(socket.bytesAvailable>= mSize) { socket.readBytes(mReader,mReader.length,mSize); var msg:Object = mReader.readObject(); if(MessageReceive!=null) MessageReceive(msg); mSize=0; } else{ mSize= mSize-socket.bytesAvailable; socket.readBytes(mReader,mReader.length,socket.bytesAvailable); } } } //发磅封装的协议数据 public function Send(message:Object,socket:Socket):void { socket.endian = Endian.LITTLE_ENDIAN; mWriter.clear(); mWriter.writeObject(message); socket.writeInt(mWriter.length+4); socket.writeBytes(mWriter,0,mWriter.length); socket.flush(); } }} 如果需要读取的数据大小为零则表明是一个新的消息,这个时候先把消息大小读取出来,注意由于c#是低字序,beetle的实现也没有处理.所以在这里需要把flash的socket设置成低字序处理.当读取一个完整的AMF3数据流后就直接读取相关对象并通过函数回调.在发送消息的方法原理一样,先写入消息总长度然后写入对应的AMF3数据流即可.这样一个flash端的分包和封包器就完成,下面就可以接入到.net的服务端进行数据交互了.
private var mPackage:HeadSizeOfPackage = new HeadSizeOfPackage(); protected function cmdConnect_clickHandler(event:MouseEvent):void { // TODO Auto-generated method stub mSocket = new Socket(); mSocket.connect(txtIPAddress.text,9860); mSocket.addEventListener(Event.CONNECT,onConnected); mSocket.addEventListener(ProgressEvent.SOCKET_DATA,socketDataHandler); mSocket.endian = Endian.LITTLE_ENDIAN; mPackage.MessageReceive=OnReceive; } private function OnReceive(msg:Object):void { if(msg is GetResponse) { var response:GetResponse= GetResponse(msg); lstData.dataProvider=new ArrayCollection(response.Items); } } private function socketDataHandler(event:ProgressEvent):void { trace("socketDataHandler: " + event); mPackage.Import(mSocket); } private function onConnected(event:Event):void { cmdRegister.enabled= true; cmdGet.enabled = true; } protected function cmdRegister_clickHandler(event:MouseEvent):void { // TODO Auto-generated method stub var reg:Register = new Register(); reg.EMail = txtEMail.text; reg.UserName=txtUserName.text; reg.PWD = txtPWD.text; mPackage.Send(reg,mSocket); } protected function cmdGet_clickHandler(event:MouseEvent):void { // TODO Auto-generated method stub var get:Get = new Get(); get.CustomerID = txtCustomerID.text; mPackage.Send(get,mSocket); } 具体运行效果
