как .Net Task Parallel Library dataflow отправляет сообщение многим клиентам с не только одним кешем?

Question

Существует серверная задача, которая использует TPL Dataflow для отправки сообщений для многих клиентов. Задача.

• клиент подключается к серверу случайным образом
• Клиент может отправить сообщение на сервер и клиент может получить сообщение от сервера

сервер использовать BufferBlock<string> отправить сообщение клиенту, когда клиент подключается к серверу , он получает сообщение от этого BufferBlock<string>.

Но это BufferBlock<string> может кэшировать только одно сообщение, и клиент не может запрашивать более одного сообщения с сервера, а клиент не может установить условие для выбора того, какое сообщение получать.

Я хочу тип блока, который может кэшировать несколько сообщений; и клиент может читать не только одно сообщение из этого типа блока, и клиент может выбрать, какое сообщение получать;

Я пробовал другие TPL Dataflow типа блоков, но никто не имеет таких способностей, TPL Dataflow блок не подходит для таких требований?

Условие простое, каждое сообщение имеет метку времени, клиент просто отправляет временную метку на сервер, а затем возвращает серверные сообщения, отправленные после отметки времени.

using System; 
using System.Web; 
using System.Net; 
using System.Threading.Tasks; 
using System.Text; 
using SimpleJSON; 
using System.Collections.Generic; 
using System.Threading.Tasks.Dataflow; 

namespace TestHttp 
{ 
    public class HttpServer 
    { 
     private HttpListener httpListener; 
     public Task task; 
     public HttpServer() 
     { 
      var ta = Task.Factory.StartNew(RunHttp); 
      task = ta.Result; 
     } 

     private async Task RunHttp() 
     { 
      var httpPort = 9090; 
      httpListener = new HttpListener(); 
      httpListener.Prefixes.Add("http://*:"+httpPort+"/"); 
      httpListener.Start(); 
      while (httpListener.IsListening) 
      { 
       var context = await httpListener.GetContextAsync(); 
       var req = context.Request; 
       Handle(context, req); 
      } 

      httpListener.Stop(); 
      httpListener.Close(); 
     } 

     private async Task Handle(HttpListenerContext context, HttpListenerRequest req) 
     { 
      Console.WriteLine(req.RawUrl); 

      var resp = await HandleGet(req); 
      var buf = Encoding.UTF8.GetBytes(resp); 
      context.Response.AddHeader("Content-Encoding", "utf-8"); 
      context.Response.ContentEncoding = Encoding.UTF8; 
      context.Response.ContentLength64 = buf.Length; 
      try 
      { 
       context.Response.OutputStream.Write(buf, 0, buf.Length); 
      } 
      catch (Exception exp) 
      { 
       Console.WriteLine(exp.ToString()); 
      } 
      finally 
      { 
       context.Response.OutputStream.Close(); 
      } 

     } 

     private BufferBlock<string> messages = new BufferBlock<string>(); 

     private async Task<string> HandleGet(HttpListenerRequest req) 
     { 
      var r = req.RawUrl.Split('?'); 
      if (r[0] == "/send") 
      { 
       await messages.SendAsync(r[1]); 
       return "Suc"; 
      } 
      else if(r[0] == "/receive"){ 
       var timestamp = Convert.ToInt32(r[1]); 
       var ret = await messages.ReceiveAsync(); 
       return ret; 
      } 
      //Console.WriteLine(r[0]); 
      return "Error"; 

     } 
    } 
} 

Answer 1

Почему вы говорите, что BufferBlock может содержать только одно значение? Это неверно, он может содержать столько сообщений, сколько вы хотите, на основе вариантов создания блоков и, в частности, BoundedCapacity. Значение по умолчанию для этого параметра равно -1, что означает неограниченную емкость.

Итак, в настоящий момент клиент подключается, вы можете легко получить все сообщения, отфильтрованные по метке времени, и вернуть их клиенту. Возможно, это приведет к изменению сигнатуры результата для запроса клиента, поскольку вы должны предоставить параметр TimeStamp и предоставить возможность вернуть List<T> сообщений, а не только один. Без какого-либо кода мы не можем больше говорить об этой проблеме.

Answer 2

Я думаю, что блок потока данных TPL не может удовлетворить такое требование.

Я просто использовать список, чтобы держать все сообщения List<Message> messages;

struct Message { 
    int id; 
    string msg; 
} 

и мне нужно использовать замок или актер модель подобный почтовый ящик, для обработки моего клиента запрос на List<Message>.