Commit d90b6ad0 authored by justcoding121's avatar justcoding121

refactor retry

parent 0d313e0e
using Polly;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Titanium.Web.Proxy.Network.Tcp;
namespace Titanium.Web.Proxy.Network
{
internal class RetryPolicy<T> where T : Exception
{
private readonly int retries;
private readonly TcpConnectionFactory tcpConnectionFactory;
internal RetryPolicy(int retries, TcpConnectionFactory tcpConnectionFactory)
{
this.retries = retries;
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>
/// Execute and retry the given action until retry number of times.
/// </summary>
/// <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="initialConnection">Initial Tcp connection to use.</param>
/// <returns></returns>
internal Task ExecuteAsync(Func<TcpServerConnection, Task> action,
Func<Task<TcpServerConnection>> generator, ref TcpServerConnection initialConnection)
{
var outerContext = new Dictionary<string, object> { { "connection", initialConnection } };
Task result;
try
{
result = getRetryPolicy().ExecuteAsync(async (context) =>
{
//setup connection
var connection = context["connection"] as TcpServerConnection ??
await generator();
context["connection"] = connection;
//retry
await action(connection);
}, outerContext);
}
//all retries failed
catch
{
//update the original connection to last used connection
initialConnection = outerContext["connection"] as TcpServerConnection;
throw;
}
return result;
}
}
}
......@@ -7,7 +7,6 @@ using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using Polly;
using StreamExtended;
using StreamExtended.Network;
using Titanium.Web.Proxy.EventArguments;
......@@ -125,7 +124,7 @@ namespace Titanium.Web.Proxy
private SystemProxyManager systemProxySettingsManager { get; }
//Number of exception retries when connection pool is enabled.
private int retries => EnableConnectionPool ? MaxCachedConnections + 1 : 0;
private int retries => EnableConnectionPool ? MaxCachedConnections : 0;
/// <summary>
/// Is the proxy currently running?
......@@ -809,22 +808,9 @@ namespace Titanium.Web.Proxy
/// <summary>
/// Connection retry policy when using connection pool.
/// </summary>
private Policy retryPolicy<T>() where T : Exception
private RetryPolicy<T> retryPolicy<T>() where T : Exception
{
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;
}
});
return new RetryPolicy<T>(retries, tcpConnectionFactory);
}
}
}
......@@ -204,30 +204,28 @@ namespace Titanium.Web.Proxy
connection = null;
}
//for connection pool retry fails until cache is exhausted
await retryPolicy<ServerConnectionException>().ExecuteAsync(async (context) =>
{
connection = context["connection"] as TcpServerConnection ??
await getServerConnection(args, false,
//a connection generator task with captured parameters via closure.
Func<Task<TcpServerConnection>> generator = () => getServerConnection(args, false,
clientConnection.NegotiatedApplicationProtocol,
false, cancellationToken);
context["connection"] = connection;
//for connection pool, retry fails until cache is exhausted.
await retryPolicy<ServerConnectionException>().ExecuteAsync(async (serverConnection) =>
{
// if upgrading to websocket then relay the request without reading the contents
if (request.UpgradeToWebSocket)
{
await handleWebSocketUpgrade(httpCmd, args, request,
response, clientStream, clientStreamWriter,
connection, cancellationTokenSource, cancellationToken);
serverConnection, cancellationTokenSource, cancellationToken);
closeServerConnection = true;
return;
}
// construct the web request that we are going to issue on behalf of the client.
await handleHttpSessionRequestInternal(connection, args);
await handleHttpSessionRequestInternal(serverConnection, args);
}, new Dictionary<string, object> { { "connection", connection } });
}, generator, ref connection);
//user requested
if (args.WebSession.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