Я написал базовую реализацию c, используя TcpClient.
Переменные:
- Хост: 127.0.0.1
- Порт: 5000
- Метод: GET
- URL: / home / message
Listener
private static async Task ConnectEventStreamAsync(CancellationToken token)
{
var client = new TcpClient();
try
{
await client.ConnectAsync("127.0.0.1", 5000);
if (!client.Connected)
throw new Exception("Unable to connect the host");
var encoding = Encoding.UTF8;
var stream = client.GetStream();
var connectBytes = encoding.GetBytes(
"GET /home/message HTTP/1.1\r\n" +
"Host: 127.0.0.1\r\n" +
"Content-Length: 0\r\n\r\n"
);
await stream.WriteAsync(connectBytes, 0, connectBytes.Length, token);
var buffer = new byte[4096];
while (!token.IsCancellationRequested)
{
var readCount = await stream.ReadAsync(buffer, 0, buffer.Length, token);
if (readCount > 0)
{
var message = encoding.GetString(buffer, 0, readCount);
using (var stringReader = new StringReader(message))
{
string line;
while ((line = await stringReader.ReadLineAsync()) != null)
{
//--try to read the next chunk
if (!int.TryParse(line, NumberStyles.HexNumber, null, out var bytes))
continue;
var data = await stringReader.ReadLineAsync();
Console.WriteLine($">>New Event ({bytes} bytes)\r\n{data}");
}
}
}
await Task.Delay(500, token);
}
}
finally
{
client.Dispose();
}
}
РЕДАКТИРОВАТЬ : вот другое решение с использованием HttpClient. Я добавил дополнительный код для анализа сообщений text-eventstream. Вы можете прочитать больше здесь , чтобы внести улучшения.
private static async Task ConnectEventStreamAsync(CancellationToken token)
{
var client = new HttpClient
{
Timeout = Timeout.InfiniteTimeSpan
};
try
{
var response = await client.GetAsync(
"http://127.0.0.1:5000/home/message",
HttpCompletionOption.ResponseHeadersRead,
token
);
if (!response.IsSuccessStatusCode)
throw new Exception("Unable to connect the stream");
var isTextEventStream = response.Content.Headers.ContentType.MediaType == "text/event-stream";
if (!isTextEventStream)
throw new InvalidOperationException("Invalid resource content type");
var stream = await response.Content.ReadAsStreamAsync();
var buffer = new byte[4096];
while (!token.IsCancellationRequested)
{
var readCount = await stream.ReadAsync(buffer, 0, buffer.Length, token);
if (readCount > 0)
{
var data = Encoding.UTF8.GetString(buffer, 0, readCount);
await ParseDataAsync(data);
}
await Task.Delay(500, token);
}
}
finally
{
client.Dispose();
}
async Task ParseDataAsync(string data)
{
using (var stringReader = new StringReader(data))
{
string line;
while ((line = await stringReader.ReadLineAsync()) != null)
{
if (line.StartsWith("event:"))
{
var eventText = line.Substring("event:".Length);
Console.WriteLine($">>Event ({eventText.Length} bytes)\r\n{eventText}");
continue;
}
if (line.StartsWith("data:"))
{
var dataText = line.Substring("data:".Length);
Console.WriteLine($">>Data ({dataText.Length} bytes)\r\n{dataText}");
continue;
}
if (line.StartsWith("id:"))
{
var eventId = line.Substring("id:".Length);
Console.WriteLine($">>Event ID ({eventId.Length} bytes)\r\n{eventId}");
continue;
}
if (line.StartsWith("retry:"))
{
var retryValue = line.Substring("retry:".Length);
Console.WriteLine($">>Retry ({retryValue.Length} bytes)\r\n{retryValue}");
continue;
}
if (line.StartsWith(":"))
{
Console.WriteLine($">>Comment ({line.Length - 1} bytes)\r\n{line.Substring(1)}");
}
}
}
}
}
Серверная часть, использующая Asp. Net Core (очень похожа на ту, что отображается пользователем, который спрашивает).
...
private static ConcurrentQueue<PingData> pings = new ConcurrentQueue<PingData>();
[HttpGet]
public void Ping(int userID)
{
pings.Enqueue(new PingData { UserID = userID });
}
[HttpGet]
public void Message()
{
Response.ContentType = "text/event-stream";
Response.WriteAsync($":Hello {Request.Host}\n");
const int intervalMs = 1000;
do
{
if (pings.TryDequeue(out var nextPing))
{
Response.WriteAsync($"event:Ping\n");
Response.WriteAsync($"retry:{intervalMs}\n");
Response.WriteAsync($"id:{DateTime.Now.Ticks}\n");
Response.WriteAsync($"data:{JsonConvert.SerializeObject(nextPing)}\n\n");
}
Thread.Sleep(intervalMs);
} while (Response.Body.CanWrite);
}
public class PingData
{
public int UserID { get; set; }
public DateTime Date { get; set; } = DateTime.Now;
}
...