Asp.net Core中如何使用中间件来管理websocket

介绍

我喜欢.net core 这个东西，其实不仅仅源于它性能很高，可以跨平台，还因为它的设计模式确实令人着迷。以前没.net core 的时候，.net用websocket必须跑在windows server 2012上，但我一般不会这么干，都把websocket架在nodejs的服务器上。这么分出来，麻烦肯定是麻烦的，而且js这东西，写复杂和几年后再看都是头疼的问题。那么，如果.net core是以kestrel运行的，那么就不再需要考虑服务器的版本运行，任何一个地方都可以用websocket

asp.net core signalr是一个有用的库，可以简化web应用程序中实时通信的管理。但是，我宁愿使用websockets，因为我想要更灵活，并且与任何websocket客户端兼容。

在microsoft的文档中，我找到了一个很好的websockets工作示例。它仍然是管理连接，以便能够从一个连接向其他连接广播消息，这是signalr开箱即用的功能。期望这个逻辑非常复杂，我想从startup类中删除它。

背景

要阅读asp.net core中的websockets支持，可以在此处。如果您想了解中间件以及如何在asp.net core中编写它，请阅读此。

代码使用

首先，你必须添加 microsoft.aspnetcore.websockets 包到你的项目。

现在，您可以创建一个扩展方法和类来管理websockets：

public static class websocketextensions
{
 public static iapplicationbuilder usecustomwebsocketmanager(this iapplicationbuilder app)
 {
  return app.usemiddleware<customwebsocketmanager>();
 }
}

public class customwebsocketmanager
{
 private readonly requestdelegate _next;

 public customwebsocketmanager(requestdelegate next)
 {
  _next = next;
 }

 public async task invoke(httpcontext context, icustomwebsocketfactory wsfactory, icustomwebsocketmessagehandler wsmhandler)
 {
  if (context.request.path == "/ws")
  {
   if (context.websockets.iswebsocketrequest)
   {
    string username = context.request.query["u"];
    if (!string.isnullorempty(username))
    {
     websocket websocket = await context.websockets.acceptwebsocketasync();
     customwebsocket userwebsocket = new customwebsocket()
     {
      websocket = websocket,
      username = username
     };
     wsfactory.add(userwebsocket);
     await wsmhandler.sendinitialmessages(userwebsocket);
     await listen(context, userwebsocket, wsfactory, wsmhandler);
    }
   }
   else
   {
     context.response.statuscode = 400;
   }
  }
  await _next(context);
 }

 private async task listen(httpcontext context, customwebsocket userwebsocket, icustomwebsocketfactory wsfactory, icustomwebsocketmessagehandler wsmhandler)
 {
  websocket websocket = userwebsocket.websocket;
  var buffer = new byte[1024 * 4];
  websocketreceiveresult result = await websocket.receiveasync(new arraysegment<byte>(buffer), cancellationtoken.none);
  while (!result.closestatus.hasvalue)
  {
    await wsmhandler.handlemessage(result, buffer, userwebsocket, wsfactory);
    buffer = new byte[1024 * 4];
    result = await websocket.receiveasync(new arraysegment<byte>(buffer), cancellationtoken.none);
  } 
  wsfactory.remove(userwebsocket.username);
  await websocket.closeasync(result.closestatus.value, result.closestatusdescription, cancellationtoken.none);
 }
}

在这种情况下，websockets请求在url中始终包含“/ ws”。查询字符串包含用于将websocket与登录用户相关联的用户名的参数u。

customwebsocket是一个包含websocket和用户名的类：

public class customwebsocket
{
 public websocket websocket { get; set; }
 public string username { get; set; }
}

我也创建了自定义websocket消息:

class customwebsocketmessage
{
 public string text { get; set; }
 public datetime messagdatetime { get; set; }
 public string username { get; set; }
 public wsmessagetype type { get; set; }
}

其中type是您可能拥有的不同类型消息的枚举。

在startup类中，您必须注册以下服务：

services.addsingleton<icustomwebsocketfactory, customwebsocketfactory>();
services.addsingleton<icustomwebsocketmessagehandler, customwebsocketmessagehandler>();

customwebsocketfactory负责收集连接的websockets列表：

public interface icustomwebsocketfactory
{
 void add(customwebsocket uws);
 void remove(string username);
 list<customwebsocket> all();
 list<customwebsocket> others(customwebsocket client);
 customwebsocket client(string username);
}

public class customwebsocketfactory : icustomwebsocketfactory
{
 list<customwebsocket> list;

 public customwebsocketfactory()
 {
  list = new list<customwebsocket>();
 }

 public void add(customwebsocket uws)
 {
  list.add(uws);
 }

 //when disconnect
 public void remove(string username) 
 {
  list.remove(client(username));
 }

 public list<customwebsocket> all()
 {
  return list;
 }
 
 public list<customwebsocket> others(customwebsocket client)
 {
  return list.where(c => c.username != client.username).tolist();
 }
 
 public customwebsocket client(string username)
 {
  return list.first(c=>c.username == username);
 }
}

customwebsocketmessagehandler包含有关消息的逻辑（即在连接时需要发送任何消息以及如何对传入消息作出反应）

public interface icustomwebsocketmessagehandler
{
 task sendinitialmessages(customwebsocket userwebsocket);
 task handlemessage(websocketreceiveresult result, byte[] buffer, customwebsocket userwebsocket, icustomwebsocketfactory wsfactory);
 task broadcastothers(byte[] buffer, customwebsocket userwebsocket, icustomwebsocketfactory wsfactory);
 task broadcastall(byte[] buffer, customwebsocket userwebsocket, icustomwebsocketfactory wsfactory);
}

public class customwebsocketmessagehandler : icustomwebsocketmessagehandler
{
 public async task sendinitialmessages(customwebsocket userwebsocket)
 {
  websocket websocket = userwebsocket.websocket;
  var msg = new customwebsocketmessage
  {
   messagdatetime = datetime.now,
   type = wsmessagetype.anytype,
   text = anytext,
   username = "system"
  };

  string serialisedmessage = jsonconvert.serializeobject(msg);
  byte[] bytes = encoding.ascii.getbytes(serialisedmessage);
  await websocket.sendasync(new arraysegment<byte>(bytes, 0, bytes.length), websocketmessagetype.text, true, cancellationtoken.none);
 }

 public async task handlemessage(websocketreceiveresult result, byte[] buffer, customwebsocket userwebsocket, icustomwebsocketfactory wsfactory)
 {
  string msg = encoding.ascii.getstring(buffer);
  try
  {
   var message = jsonconvert.deserializeobject<customwebsocketmessage>(msg);
   if (message.type == wsmessagetype.anytype)
   {
   await broadcastothers(buffer, userwebsocket, wsfactory);
   }
  }
  catch (exception e)
  {
   await userwebsocket.websocket.sendasync(new arraysegment<byte>(buffer, 0, result.count), result.messagetype, result.endofmessage, cancellationtoken.none);
  }
 }

 public async task broadcastothers(byte[] buffer, customwebsocket userwebsocket, icustomwebsocketfactory wsfactory)
 {
  var others = wsfactory.others(userwebsocket);
  foreach (var uws in others)
  {
   await uws.websocket.sendasync(new arraysegment<byte>(buffer, 0, buffer.length), websocketmessagetype.text, true, cancellationtoken.none);
  }
 }

 public async task broadcastall(byte[] buffer, customwebsocket userwebsocket, icustomwebsocketfactory wsfactory)
 {
  var all = wsfactory.all();
  foreach (var uws in all)
  {
   await uws.websocket.sendasync(new arraysegment<byte>(buffer, 0, buffer.length), websocketmessagetype.text, true, cancellationtoken.none);
  }
 }
}

最后，在configure方法的startup类中添加以下内容：

var websocketoptions = new websocketoptions()
{
 keepaliveinterval = timespan.fromseconds(120),
 receivebuffersize = 4 * 1024
};

app.usewebsockets(websocketoptions);
app.usecustomwebsocketmanager();

通过这种方式，starup类保持干净，管理websockets的逻辑可以扩展，使您可以根据自己的喜好灵活地组织它。

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对的支持。