In one of my applications I make requests to a server. However, that server may at some point start making requests to my application as well. The server also allows polling, this means that I am able to make a request to a URL on the server and it only returns after some time if it does not have any content available at the moment (the polling does not guarantee that there will be content).
So I implemented the class TwoWayHttpClient
that is able to abstract all of this for me. It makes use of a HttpClient
, to make requests to the server. It creates a HttpListener
if needed, and also supports the polling.
This class provides the method RequestAndWaitForReply
that makes a request to the server and waits for an expected message from server (that can arrive several minutes after). Remember that that message can arrive both via the listener, or via polling.
The class also provides the method WaitForMessage
in the case the server sends the message without me having to make any request.
BaseIncomingMessage
provides an abstraction to the kind of the response that is sent by the server. Classes that derive from it can override IsExpectedMessageImpl
to determinate if the server sent the expected message.
One thing that I don't like a lot is that this class is used in two different circumstances. The first is when ever I receive a request from the server and put it on the queue. The second is when I create an instance of it to be able to compare to the received message.
FilterQueue
is already up to review on my other post.
public class TwoWayHttpClient : IDisposable
{
private readonly HttpClientHandler _handler;
private readonly HttpClient _client;
protected HttpListener _listener;
protected readonly FilterQueue<IIncomingMessage> _queue = new FilterQueue<IIncomingMessage>();
public string ListeningAddress { get; private set; }
private TwoWayHttpClient()
{
_handler = new HttpClientHandler()
{
CookieContainer = new CookieContainer()
};
_client = new HttpClient(_handler);
}
protected TwoWayHttpClient(string server)
: this()
{
_client.BaseAddress = new Uri("http://" + server);
_client.Timeout = TimeSpan.FromMinutes(10);
}
public void StartListener(int listenerPort, string baseAddress = "/")
{
if (_listener != null)
{
return;
}
//the listening address contaning local ip adress, so there aren't any issues about who the server needs to send the request to
ListeningAddress = "http://" + GetLocalIPAddress() + ":" + listenerPort + baseAddress;
_listener = new HttpListener();
//Add the prefix for the listener on the specified port, the + sign is important so it can recieve requests
_listener.Prefixes.Add("http://+:" + listenerPort + baseAddress);
_listener.Start();
//starts listening to requests
Task.Run(async () =>
{
try
{
while (_listener.IsListening)
{
var context = await _listener.GetContextAsync();
//for every request recieved put it on a queue.
_queue.Add(GetIncommingMessageFromContext(context));
//answer to the server with 200. This is important, so the server can keep sending other requests
using (context.Response) { }
}
}
catch (ObjectDisposedException)
{
//listener was disposed, ignore it
}
});
}
CancellationTokenSource _cancelSource = new CancellationTokenSource();
public void InitiatePolling(string url)
{
_cancelSource = new CancellationTokenSource();
var token = _cancelSource.Token;
Task.Run(async () =>
{
while (true)
{
_queue.WaitForWaiters();
token.ThrowIfCancellationRequested();
var response = await Request(url);
if (!IsValidResponse(response))
{
throw new InvalidOperationException(response.ToString());
}
if (response.StatusCode != HttpStatusCode.NoContent)
{
_queue.Add(GetIncommingMessageFromResponse(response));
}
}
}, token);
}
protected virtual IIncomingMessage GetIncommingMessageFromResponse(HttpResponseMessage response)
{
return new BaseIncomingMessage(response);
}
protected virtual IIncomingMessage GetIncommingMessageFromContext(HttpListenerContext context)
{
return new BaseIncomingMessage(context);
}
private static string GetLocalIPAddress()
{
var host = Dns.GetHostEntry(Dns.GetHostName());
foreach (var ip in host.AddressList)
{
if (ip.AddressFamily == AddressFamily.InterNetwork)
{
return ip.ToString();
}
}
throw new Exception("Local IP Address Not Found!");
}
private Task<string> GetQueryStringFromDictionary(IDictionary<string, object> parameters)
{
var values = parameters.Where(p => p.Value != null)
.Select(p => new KeyValuePair<string, string>(p.Key, p.Value.ToString()));
using (var query = new FormUrlEncodedContent(values))
{
return query.ReadAsStringAsync();
}
}
private Cookie GetSetCookie(HttpResponseMessage response)
{
IEnumerable<string> authValues;
if (!response.Headers.TryGetValues("Set-Cookie", out authValues))
{
return null;
}
var cookieValue = authValues.FirstOrDefault();
var idxEquals = cookieValue.IndexOf('=');
var name = cookieValue.Substring(0, idxEquals);
var value = cookieValue.Substring(idxEquals + 1, cookieValue.IndexOf(';') - idxEquals - 1);
return new Cookie(name, value)
{
Domain = _client.BaseAddress.Host
};
}
public async Task<HttpResponseMessage> Request(string url, IDictionary<string, object> parameters = null, string name = "")
{
if (parameters != null && parameters.Count != 0)
{
var query = await GetQueryStringFromDictionary(parameters);
url = url + "?" + query;
}
var response = await _client.GetAsync(url);
var cookie = GetSetCookie(response);
if (cookie != null)
{
_handler.CookieContainer.Add(cookie);
}
return response;
}
protected async Task<TwoWayHttpResponse> RequestAndWaitForReply(string url, object parameters, IIncomingMessage expectedMessage)
{
if (parameters == null)
{
parameters = new object();
}
var response = await Request(url, parameters.ToDictionary());
if (IsValidResponse(response))
{
if (expectedMessage == null)
{
return new TwoWayHttpResponse
{
Response = response
};
}
var result = new TwoWayHttpResponse
{
Response = response,
ServerRequest = await WaitForMessage(expectedMessage)
};
return result;
}
return new TwoWayHttpResponse
{
Response = response
};
}
public virtual Task<IIncomingMessage> WaitForMessage(IIncomingMessage expectedMessage)
{
var message = _queue.Take(expectedMessage.IsExpectedMessage, Timeout.InfiniteTimeSpan);
return Task.FromResult(message);
}
public virtual void ClearQueue()
{
_queue.Clear();
}
protected virtual bool IsValidResponse(HttpResponseMessage response)
{
return (int)response.StatusCode >= 200 && (int)response.StatusCode < 400;
}
public void Dispose()
{
if (_listener != null)
{
_listener.Stop();
(_listener as IDisposable).Dispose();
}
if (_cancelSource != null)
{
_cancelSource.Cancel();
}
_client.Dispose();
}
}
public class BaseIncomingMessage : IIncomingMessage
{
#if DEBUG
private readonly HttpListenerContext _context;
private readonly HttpResponseMessage _response;
#endif
private readonly Task<string> _content;
public BaseIncomingMessage(HttpListenerContext context)
{
_context = context;
using (var reader = new StreamReader(_context.Request.InputStream))
{
_content = reader.ReadToEndAsync();
}
}
public BaseIncomingMessage(HttpResponseMessage response)
{
_response = response;
_content = response.Content.ReadAsStringAsync();
}
public BaseIncomingMessage()
{
}
public bool IsExpectedMessage(IIncomingMessage message)
{
return IsExpectedMessageImpl(message, _content.Result);
}
public virtual bool IsExpectedMessageImpl(IIncomingMessage originalMessage, string content)
{
return true;
}
}
public interface IIncomingMessage
{
bool IsExpectedMessage(IIncomingMessage compareTo);
}
Some utilities:
public static class ObjectUtilities
{
private static bool HasValue(object value)
{
if (value != null)
{
if (value is bool)
{
return (bool)value;
}
return true;
}
return false;
}
public static Dictionary<string, object> ToDictionary(this object myObj)
{
if (myObj == null)
{
return new Dictionary<string, object>();
}
return myObj.GetType()
.GetProperties()
.Where(pi => HasValue(pi.GetValue(myObj, null)))
.Select(pi =>
{
var dataMemberAttribute = pi.GetCustomAttribute<DataMemberAttribute>();
var name = dataMemberAttribute == null ? pi.Name : dataMemberAttribute.Name;
return new { Name = name, Value = pi.GetValue(myObj, null) };
})
.AsParallel()
.ToDictionary(ks => ks.Name, vs => vs.Value);
}
}