BeetleX实现MessagePack和Protobuf消息控制器调用websocket服务详解

最近有用户问如何使用BeetleX封装一个基于Protobuf格式的websocket服务并支持控制器调用；其实BeetleX.FastHttpApi是支持Websocket服务和自定义数据格式的，但需要对组件有一定了解的情况才能进行扩展；接下来通过封装一个支持Protobuf和MessagePack的websocket服务来介绍BeetleX.FastHttpApi的相关功能扩展和应用。

格式封装

首先需要一个特性来标记消息类型在数据中用怎样的数据来标记

[AttributeUsage(AttributeTargets.Class)]public class BinaryTypeAttribute : Attribute{public BinaryTypeAttribute(int id){this.ID = id;}public int ID { get; set; }}

可以通过一个Int类型来定义数值与消息类的关系。然后再定义一个特性用于描述消息对应的方法控制器类

[AttributeUsage(AttributeTargets.Class)]public class MessageControllerAttribute : Attribute{}

接下来就可以封装对应Protobuf的数据转换类，为了转换方便就直接使用Protobuf.Net这个组件了

public class BinaryDataFactory{//数值与类型的关系private Dictionary<int, Type> mMessageTypes = new Dictionary<int, Type>();//类型与数值的关系private Dictionary<Type, int> mMessageIDs = new Dictionary<Type, int>();//定义消息与方法的关系private Dictionary<Type, ActionHandler> mActions = new Dictionary<Type, ActionHandler>();public void RegisterComponent<T>(){RegisterComponent(typeof(T));}public void RegisterComponent(Type type){foreach (var item in type.Assembly.GetTypes()){BinaryTypeAttribute[] bta = (BinaryTypeAttribute[])item.GetCustomAttributes(typeof(BinaryTypeAttribute), false);if (bta != null && bta.Length > 0){mMessageTypes[bta[0].ID] = item;mMessageIDs[item] = bta[0].ID;}
#if PROTOBUF_SERVERvar mca = item.GetCustomAttribute<MessageControllerAttribute>(false);if (mca != null){var controller = Activator.CreateInstance(item);foreach (MethodInfo method in item.GetMethods(BindingFlags.Public | BindingFlags.Instance)){var parameters = method.GetParameters();if (parameters.Length == 2 && parameters[0].ParameterType == typeof(WebSocketReceiveArgs)) {ActionHandler handler = new ActionHandler(controller, method);mActions[parameters[1].ParameterType] = handler;}}}
#endif}}//序列化对象public ArraySegment<byte> Serializable(object data){MemoryStream memory = new MemoryStream();var type = GetTypeID(data.GetType(), true);memory.Write(type);Serializer.Serialize(memory, data);return new ArraySegment<byte>(memory.GetBuffer(), 0, (int)memory.Length);}//反序列化对象public object Deserialize(byte[] data){return Deserialize(new ArraySegment<byte>(data, 0, data.Length));}//反序列化对象public object Deserialize(ArraySegment<byte> data){MemoryStream memory = new MemoryStream(data.Array, data.Offset, data.Count);byte[] id = new byte[4];memory.Read(id, 0, 4);Type type = GetMessageType(id, true);return Serializer.Deserialize(type, memory);}//获消息对应数值的存储数据public byte[] GetTypeID(Type type, bool littleEndian){if (mMessageIDs.TryGetValue(type, out int value)){if (!littleEndian){value = BeetleX.Buffers.BitHelper.SwapInt32(value);}return BitConverter.GetBytes(value);}throw new Exception($"binary websocket {type} id not found!");}//根据数值获取类型public Type GetMessageType(int id){mMessageTypes.TryGetValue(id, out Type result);return result;}//根据存储数获取类型public Type GetMessageType(byte[] data, bool littleEndian){int value = BitConverter.ToInt32(data, 0);if (!littleEndian)value = BeetleX.Buffers.BitHelper.SwapInt32(value);return GetMessageType(value);}//根据消息获处理方法public ActionHandler GetHandler(object message){mActions.TryGetValue(message.GetType(), out ActionHandler result);return result;}}public class ActionHandler{public ActionHandler(object controller, MethodInfo method){Method = method;Controller = controller;IsVoid = method.ReturnType == typeof(void);IsTaskResult = method.ReturnType.BaseType == typeof(Task);if (IsTaskResult && method.ReturnType.IsGenericType){HasTaskResultData = true;mGetPropertyInfo = method.ReturnType.GetProperty("Result", BindingFlags.Public | BindingFlags.Instance);}}private PropertyInfo mGetPropertyInfo;public MethodInfo Method { get; private set; }public bool IsTaskResult { get; set; }public bool HasTaskResultData { get; set; }public bool IsVoid { get; private set; }public object Controller { get; private set; }public object GetTaskResult(Task task){return mGetPropertyInfo.GetValue(task);}public object Execute(params object[] data){return Method.Invoke(Controller, data);}}

协议转换

Protobuf的数据格式处理完成后就要针对BeetleX.FastHttpApi构建一个相应数据转换器，这个转换器实现也是非常简单的。

public class ProtobufFrameSerializer : IDataFrameSerializer{public static BinaryDataFactory BinaryDataFactory { get; set; } = new BinaryDataFactory();public object FrameDeserialize(DataFrame data, PipeStream stream){var buffers = new byte[data.Length];stream.Read(buffers, 0, buffers.Length);return BinaryDataFactory.Deserialize(buffers);}public void FrameRecovery(byte[] buffer){}public ArraySegment<byte> FrameSerialize(DataFrame packet, object body){return BinaryDataFactory.Serializable(body);}}

实现IDataFrameSerializer接口的相关方法即可，接下来就针对HttpApiServer封装一些扩展方法用于注册对象和绑定Websocket数据处理事件。

public static class ProtobufExtensions{public static HttpApiServer RegisterProtobuf<T>(this HttpApiServer server){ProtobufFrameSerializer.BinaryDataFactory.RegisterComponent<T>();return server;}public static HttpApiServer UseProtobufController(this HttpApiServer server, Action<WebSocketReceiveArgs> handler = null){server.WebSocketReceive = async (o, e) =>{try{var msg = e.Frame.Body;var action = ProtobufFrameSerializer.BinaryDataFactory.GetHandler(msg);if (action != null){if (!action.IsVoid){if (action.IsTaskResult){Task task = (Task)action.Execute(e, msg);await task;if (action.HasTaskResultData){var result = action.GetTaskResult(task);e.ResponseBinary(result);}}else{var result = action.Execute(e, msg);e.ResponseBinary(result);}}}else{handler?.Invoke(e);}}catch (Exception e_){e.Request.Server.GetLog(BeetleX.EventArgs.LogType.Warring)?.Log(BeetleX.EventArgs.LogType.Error, e.Request.Session, $"Websocket packet process error {e_.Message}");}};return server;}}

服务使用

所有相关功能都封装后就可以启动HttpApiServer并把相关功能设置使用，在使用之前需要引用BeetleX.FastHttpApi.Hosting和Protobuf.Net组件

[BinaryType(1)][ProtoContract]public class User{[ProtoMember(1)]public string Name { get; set; }[ProtoMember(2)]public string Email { get; set; }[ProtoMember(3)]public DateTime ResultTime { get; set; }}[MessageController]public class Controller{public User Login(WebSocketReceiveArgs e, User user){user.ResultTime = DateTime.Now;return user;}}class Program{static void Main(string[] args){BeetleX.FastHttpApi.Hosting.HttpServer server = new BeetleX.FastHttpApi.Hosting.HttpServer(80);server.Setting((service, options) =>{options.LogLevel = BeetleX.EventArgs.LogType.Trace;options.LogToConsole = true;options.ManageApiEnabled = false;options.WebSocketFrameSerializer = new ProtobufFrameSerializer();});server.Completed(http =>{http.RegisterProtobuf<User>();http.UseProtobufController();});server.Run();}}

客户端

如果你希望在.Net中使用Websocket客户端，BeetleX同样也提供一个扩展组件BeetleX.Http.Clients,通过这组件可以轻易封装一个Protobuf的Websocket客户端。

public class ProtobufClient : BinaryClient{public ProtobufClient(string host) : base(host) { }public static BinaryDataFactory BinaryDataFactory { get; private set; } = new BinaryDataFactory();protected override object OnDeserializeObject(ArraySegment<byte> data){return BinaryDataFactory.Deserialize(data);}protected override ArraySegment<byte> OnSerializeObject(object data){return BinaryDataFactory.Serializable(data);}}

封装完成后就可以使用了

class Program{static async Task Main(string[] args){ProtobufClient.BinaryDataFactory.RegisterComponent<User>();var client = new ProtobufClient("ws://localhost");User user = new User { Email="henryfan@msn.com", Name="henryfan" };var result = await client.ReceiveFrom<User>(user);Console.WriteLine(result.Name);}}

完整示例代码可以访问

https://github.com/beetlex-io/BeetleX-Samples/tree/master/Websocket.ProtobufPacket

https://github.com/beetlex-io/BeetleX-Samples/tree/master/Websocket.MessagePackPacket