Given Observable A representing incoming requests ...
interface IRequest
{
string Id { get; }
}
IObservable<IRequest> requests;
and Observable B representing responses ...
IObservable<string> responses;
I need to measure the time it takes to produce a response. My current solution looks like below. This works but I'm not sure if this could be simplified or implemented more streamlined. I'm also concerned about the deviation between actual response time ([GEN]) and observed response time ([OBS]) which becomes quiet significant if generatedResponseTime is set to sub 50ms.
class Program
{
public class Request
{
public string Id { get; set; }
}
static void Main(string[] args)
{
var generatedResponseTime = 100;
var responseSub = new Subject<string>();
IObservable<Request> requests = Observable.Interval(TimeSpan.FromMilliseconds(1000))
.Select(x=> new Request { Id = x.ToString("d") });
IObservable<string> responses = responseSub.AsObservable();
var responseTime = requests
.Where(x => !string.IsNullOrEmpty(x.Id))
.Select(req => new { Id = req.Id, Start = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() })
.SelectMany(req => responses
.Where(reqId => reqId == req.Id)
.Select(_ => new { RequestId = req.Id, ResponseTime = (int)(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - req.Start) })
.Take(1))
.Publish()
.RefCount();
requests.ObserveOn(TaskPoolScheduler.Default)
.Subscribe(req =>
{
Console.WriteLine($"[GEN] {req.Id} = {generatedResponseTime} ms");
Thread.Sleep(generatedResponseTime);
responseSub.OnNext(req.Id);
});
responseTime.Subscribe(x => Console.WriteLine($"[OBS] {x.RequestId} = {x.ResponseTime} ms"));
Console.ReadLine();
}
}