Commit dc12d02e authored by justcoding121's avatar justcoding121

retry fix

parent c5b87959
...@@ -338,8 +338,7 @@ namespace Titanium.Web.Proxy ...@@ -338,8 +338,7 @@ namespace Titanium.Web.Proxy
} }
finally finally
{ {
if (!calledRequestHandler if (!calledRequestHandler)
&& prefetchConnectionTask != null)
{ {
await tcpConnectionFactory.Release(prefetchConnectionTask, closeServerConnection); await tcpConnectionFactory.Release(prefetchConnectionTask, closeServerConnection);
} }
......
...@@ -11,66 +11,73 @@ namespace Titanium.Web.Proxy.Network ...@@ -11,66 +11,73 @@ namespace Titanium.Web.Proxy.Network
private readonly int retries; private readonly int retries;
private readonly TcpConnectionFactory tcpConnectionFactory; private readonly TcpConnectionFactory tcpConnectionFactory;
private TcpServerConnection currentConnection;
private Exception exception;
internal RetryPolicy(int retries, TcpConnectionFactory tcpConnectionFactory) internal RetryPolicy(int retries, TcpConnectionFactory tcpConnectionFactory)
{ {
this.retries = retries; this.retries = retries;
this.tcpConnectionFactory = tcpConnectionFactory; this.tcpConnectionFactory = tcpConnectionFactory;
} }
//get the policy
private Policy getRetryPolicy()
{
return Policy.Handle<T>()
.RetryAsync(retries,
onRetryAsync: async (ex, i, context) =>
{
if (context["connection"] != null)
{
//close connection on error
var connection = (TcpServerConnection)context["connection"];
await tcpConnectionFactory.Release(connection, true);
context["connection"] = null;
}
});
}
/// <summary> /// <summary>
/// Execute and retry the given action until retry number of times. /// Execute and retry the given action until retry number of times.
/// </summary> /// </summary>
/// <param name="action">The action to retry.</param> /// <param name="action">The action to retry.</param>
/// <param name="generator">The Tcp connection generator to be invoked to get new connection for retry.</param> /// <param name="generator">The Tcp connection generator to be invoked to get new connection for retry.</param>
/// <param name="initialConnection">Initial Tcp connection to use.</param> /// <param name="initialConnection">Initial Tcp connection to use.</param>
/// <returns></returns> /// <returns>Returns the latest connection used and the latest exception if any.</returns>
internal Task ExecuteAsync(Func<TcpServerConnection, Task> action, internal async Task<RetryResult> ExecuteAsync(Func<TcpServerConnection, Task> action,
Func<Task<TcpServerConnection>> generator, ref TcpServerConnection initialConnection) Func<Task<TcpServerConnection>> generator, TcpServerConnection initialConnection)
{ {
var outerContext = new Dictionary<string, object> { { "connection", initialConnection } }; currentConnection = initialConnection;
Task result;
try try
{ {
result = getRetryPolicy().ExecuteAsync(async (context) => //retry on error with polly policy
//do not use polly context to store connection; it does not save states b/w attempts
await getRetryPolicy().ExecuteAsync(async () =>
{ {
//setup connection //setup connection
var connection = context["connection"] as TcpServerConnection ?? currentConnection = currentConnection as TcpServerConnection ??
await generator(); await generator();
//try
await action(currentConnection);
context["connection"] = connection; });
}
catch (Exception e) { exception = e; }
//retry return new RetryResult(currentConnection, exception);
await action(connection); }
}, outerContext); //get the policy
} private Policy getRetryPolicy()
//all retries failed {
finally return Policy.Handle<T>()
{ .RetryAsync(retries,
//update the original connection to last used connection onRetryAsync: async (ex, i, context) =>
initialConnection = outerContext["connection"] as TcpServerConnection; {
} if (currentConnection != null)
{
//close connection on error
await tcpConnectionFactory.Release(currentConnection, true);
currentConnection = null;
}
return result; });
}
}
internal class RetryResult
{
internal bool IsSuccess => Exception == null;
internal TcpServerConnection LatestConnection { get; }
internal Exception Exception { get; }
internal RetryResult(TcpServerConnection lastConnection, Exception exception)
{
LatestConnection = lastConnection;
Exception = exception;
} }
} }
} }
...@@ -209,9 +209,9 @@ namespace Titanium.Web.Proxy ...@@ -209,9 +209,9 @@ namespace Titanium.Web.Proxy
tcpConnectionFactory.GetServerConnection(this, args, isConnect: false, tcpConnectionFactory.GetServerConnection(this, args, isConnect: false,
applicationProtocol:clientConnection.NegotiatedApplicationProtocol, applicationProtocol:clientConnection.NegotiatedApplicationProtocol,
noCache: false, cancellationToken: cancellationToken); noCache: false, cancellationToken: cancellationToken);
//for connection pool, retry fails until cache is exhausted. //for connection pool, retry fails until cache is exhausted.
await retryPolicy<ServerConnectionException>().ExecuteAsync(async (serverConnection) => var result = await retryPolicy<ServerConnectionException>().ExecuteAsync(async (serverConnection) =>
{ {
// if upgrading to websocket then relay the request without reading the contents // if upgrading to websocket then relay the request without reading the contents
if (request.UpgradeToWebSocket) if (request.UpgradeToWebSocket)
...@@ -226,7 +226,16 @@ namespace Titanium.Web.Proxy ...@@ -226,7 +226,16 @@ namespace Titanium.Web.Proxy
// construct the web request that we are going to issue on behalf of the client. // construct the web request that we are going to issue on behalf of the client.
await handleHttpSessionRequestInternal(serverConnection, args); await handleHttpSessionRequestInternal(serverConnection, args);
}, generator, ref connection); }, generator, connection);
//update connection to latest used
connection = result.LatestConnection;
//throw if exception happened
if(!result.IsSuccess)
{
throw result.Exception;
}
//user requested //user requested
if (args.WebSession.CloseServerConnection) if (args.WebSession.CloseServerConnection)
......
...@@ -164,8 +164,7 @@ namespace Titanium.Web.Proxy ...@@ -164,8 +164,7 @@ namespace Titanium.Web.Proxy
} }
finally finally
{ {
if (!calledRequestHandler if (!calledRequestHandler)
&& prefetchConnectionTask != null)
{ {
await tcpConnectionFactory.Release(prefetchConnectionTask, closeServerConnection); await tcpConnectionFactory.Release(prefetchConnectionTask, closeServerConnection);
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment