Websocket 扩展 azure 服务总线答案

【问题标题】：Websocket to extend a azure service busWebsocket 扩展 azure 服务总线
【发布时间】：2017-05-17 04:14:39
【问题描述】：

我正在.net 中通过 rest 和 web 套接字编写一个高度可用的 api。我正在寻找以下策略：

跟踪每个打开的 Web 套接字到我的 api 应用程序
如何在 asp.net webapi 应用程序中使用队列消息或来自 azure 服务总线的主题
我应该研究什么模式或策略

在这一点上，我的关键词搜索并没有为我提供很多关于如何实现这一点的见解，但如果有人有任何想法，我将不胜感激！

TL;DR：有人在 webapi 项目中创建了队列服务来将消息推送到 websocket 中吗？

【问题讨论】：

我找到了一个关于Building Real-time Web Apps with ASP.NET WebAPI and WebSockets的教程，你可以通过在WebSocketCollection中访问WebSocketHandler.AspNetWebSocketContext.WebSocket找到WebSocket实例。我无法更好地理解您关于使用 Websocket 扩展 azure 服务总线的场景，您可以提供有关您的场景的更多详细信息。
太棒了，谢谢布鲁斯！现在阅读这篇文章！
嘿布鲁斯，这是在 api 项目中实现 websockets 的一个很好的资源，但我的意图是连接套接字，然后等待来自 azure 服务总线队列的消息然后下推当它进来时给客户。

标签： .net asp.net-web-api websocket azureservicebus

【解决方案1】：

有人在 webapi 项目中创建了队列服务来将消息推送到 websocket 吗？

根据您的要求，您可以创建一个后台任务来接收来自服务总线队列的消息，如下所示：

Global.asax.cs

public class WebApiApplication : System.Web.HttpApplication
{
    private static ManualResetEvent _maunalResetEvent = new ManualResetEvent(false);
    protected void Application_Start()
    {
        AreaRegistration.RegisterAllAreas();
        GlobalConfiguration.Configure(WebApiConfig.Register);
        FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
        RouteConfig.RegisterRoutes(RouteTable.Routes);
        BundleConfig.RegisterBundles(BundleTable.Bundles);

        ThreadPool.QueueUserWorkItem(_ =>
        {
            var connectionString = "";
            var queueName = "samplequeue";
            var client = QueueClient.CreateFromConnectionString(connectionString, queueName);
            client.OnMessage(message =>
            {
                var messageBody = message.GetBody<String>();
                System.Diagnostics.Trace.TraceInformation(String.Format("Message body: {0}", messageBody));

               //TODO: send message to the specific client (user)
            });
            _maunalResetEvent.WaitOne();
        });
    }

    protected void Application_End()
    {
        _maunalResetEvent.Set();
    }
}

注意：更多关于从队列接收消息的细节，可以参考这个官方tutorial。

为了将消息推送到您的客户端，如果您基于webapi-and-websockets，我假设您需要选择WebSocketCollection 实例，然后选择特定的WebSocketHandler 实例与_username 匹配的属性来自接收到的队列消息或向所有客户端广播消息如下：

当接收到队列消息时，使用以下代码向所有客户端广播：

ChatWebSocketHandler.ChatClients.Broadcast("{queue-message}");

此外，您可以利用 SignalR 并将您的客户端（用户）映射到 signalR 连接，然后当您收到队列消息时，您可以将其发送给您的客户端（用户）。更多详情可以参考Mapping SignalR Users to Connections。另外，你可以关注SignalR Hubs API Guide 和这个git sample。

【讨论】：

我正在编写的实现是利用 websockets 作为集成点。需要一个 signalR 客户端不是我的意图，而是一个基本的字符串字符串套接字连接，用于将通知推送到高度受限的（防火墙后面）网络，除了 Web 流量之外不允许任何其他内容。

【解决方案2】：

我的要求有点复杂。我能够通过主题和订阅模式解决问题。

我从队列中发布到主题，并通过在 websocket 请求上生成的过滤订阅广播到所有 websocket 连接。很漂亮！

 private async Task ProcessWebsocketSession(AspNetWebSocketContext context)
    {
        var details = ((APIIdentity)User.Identity).Details;
        //SubscriptionHandler creates the subscription for the given system, and disposes of it safely when leaving scope
        using (var ws = context.WebSocket)
        using (var subHander = new SubscriptionHandler(details))
        {
            var client = subHander.Client;
            client.OnMessage((brokerMsg) => 
            {
                var payload = brokerMsg.GetBody<string>();
                var dataToSend = JsonConvert.DeserializeObject<Data>(brokerMsg.GetBody<string>());
                //refreshing from db, to check if data has been updated in process
                dataToSend = Repo.GetData(BusinessID, dataToSend .ID).FirstOrDefault();
                if (dataToSend != null && !dataToSend Send.IsConfirmed.Value)
                {
                    byte[] binaryData = Encoding.ASCII.GetBytes(payload);
                    var segment = new ArraySegment<byte>(binaryData);
                    var task = ws.SendAsync(segment, WebSocketMessageType.Text,
                        true, CancellationToken.None);
                    task.Wait();
                }
                brokerMsg.Complete();
            });
            var receivedBytes = new ArraySegment<byte>(new byte[1024]);
            while (ws.State == WebSocketState.Open)
            {
                // ping/pong response to verify socket is connected without changing data
                var msg = await ws.ReceiveAsync(receivedBytes, CancellationToken.None);
                if (msg.MessageType == WebSocketMessageType.Text)
                {
                    var text = Encoding.ASCII.GetString(receivedBytes.ToArray());
                    if (text.ToLower() == "ping")
                    {
                        await ws.SendAsync(new ArraySegment<byte>(Encoding.ASCII.GetBytes(text.ToCharArray())), WebSocketMessageType.Text, true, CancellationToken.None);
                    }
                }

            } 

        }

【讨论】：