/
WorkerReceiver.cs
78 lines (71 loc) · 2.74 KB
/
WorkerReceiver.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
using System;
using System.Diagnostics;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http.Connections;
using Newtonsoft.Json;
namespace Microsoft.AspNetCore.SignalR.Crankier
{
public class WorkerReceiver
{
private readonly StreamReader _reader;
private readonly IWorker _worker;
private CancellationTokenSource _receiveMessageCts;
public WorkerReceiver(StreamReader reader, IWorker worker)
{
_reader = reader;
_worker = worker;
}
public void Start()
{
if (_receiveMessageCts != null)
{
_receiveMessageCts.Cancel();
}
_receiveMessageCts = new CancellationTokenSource();
Task.Run(async () =>
{
while (!_receiveMessageCts.Token.IsCancellationRequested)
{
var messageString = await _reader.ReadLineAsync();
try
{
var message = JsonConvert.DeserializeObject<Message>(messageString);
switch (message.Command.ToLowerInvariant())
{
case "ping":
await _worker.PingAsync(
message.Value["Value"].ToObject<int>());
break;
case "connect":
await _worker.ConnectAsync(
message.Value["TargetAddress"].ToObject<string>(),
message.Value["TransportType"].ToObject<HttpTransportType>(),
message.Value["NumberOfConnections"].ToObject<int>());
break;
case "starttest":
await _worker.StartTestAsync(
TimeSpan.FromMilliseconds(message.Value.Value<double>("SendInterval")),
message.Value["SendBytes"].ToObject<int>());
break;
case "stop":
await _worker.StopAsync();
break;
}
}
catch (Exception ex)
{
Trace.WriteLine(ex.Message);
}
}
});
}
public void Stop()
{
_receiveMessageCts.Cancel();
}
}
}