Commit 27085b0e authored by Honfika's avatar Honfika

throw less eexceptions in http tunnels with using APM (Begin-End methods)....

throw less eexceptions in http tunnels with using APM (Begin-End methods). (await ReadAsync always throws exception when socket is closed)
parent 461f788a
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
<s:Boolean x:Key="/Default/CodeInspection/CodeAnnotations/NamespacesWithAnnotations/=Titanium_002EWeb_002EProxy_002EExamples_002EWpf_002EAnnotations/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/KEEP_EXISTING_EMBEDDED_ARRANGEMENT/@EntryValue">False</s:Boolean>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/LINE_FEED_AT_FILE_END/@EntryValue">True</s:Boolean>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_ACCESSORHOLDER_ATTRIBUTE_ON_SAME_LINE_EX/@EntryValue">NEVER</s:String>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/PLACE_SIMPLE_EMBEDDED_STATEMENT_ON_SAME_LINE/@EntryValue">NEVER</s:String>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SIMPLE_CASE_STATEMENT_STYLE/@EntryValue">LINE_BREAK</s:String>
<s:String x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SIMPLE_EMBEDDED_STATEMENT_STYLE/@EntryValue">LINE_BREAK</s:String>
<s:Boolean x:Key="/Default/CodeStyle/CodeFormatting/CSharpFormat/SPACE_AFTER_TYPECAST_PARENTHESES/@EntryValue">False</s:Boolean>
......@@ -21,6 +24,11 @@
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/PredefinedNamingRules/=PrivateStaticFields/@EntryIndexedValue">&lt;Policy Inspect="True" Prefix="" Suffix="" Style="aaBb" /&gt;</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/PredefinedNamingRules/=PrivateStaticReadonly/@EntryIndexedValue">&lt;Policy Inspect="True" Prefix="" Suffix="" Style="aaBb" /&gt;</s:String>
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/UserRules/=dda2ffa1_002D435c_002D4111_002D88eb_002D1a7c93c382f0/@EntryIndexedValue">&lt;Policy&gt;&lt;Descriptor Staticness="Static, Instance" AccessRightKinds="Private" Description="Property (private)"&gt;&lt;ElementKinds&gt;&lt;Kind Name="PROPERTY" /&gt;&lt;/ElementKinds&gt;&lt;/Descriptor&gt;&lt;Policy Inspect="True" Prefix="" Suffix="" Style="aaBb" /&gt;&lt;/Policy&gt;</s:String>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpAttributeForSingleLineMethodUpgrade/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpRenamePlacementToArrangementMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EAddAccessorOwnerDeclarationBracesMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateThisQualifierSettings/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
\ No newline at end of file
......@@ -18,8 +18,6 @@ namespace Titanium.Web.Proxy.Helpers
private static readonly byte[] newLine = ProxyConstants.NewLine;
private static readonly Encoding encoding = Encoding.ASCII;
private static readonly Encoder encoder = Encoding.ASCII.GetEncoder();
internal HttpWriter(Stream stream, int bufferSize) : base(stream)
......@@ -35,10 +33,16 @@ namespace Titanium.Web.Proxy.Helpers
return WriteAsync(newLine);
}
public async Task WriteAsync(string value)
public Task WriteAsync(string value)
{
return WriteAsyncInternal(value, false);
}
private Task WriteAsyncInternal(string value, bool addNewLine)
{
int newLineChars = addNewLine ? newLine.Length : 0;
int charCount = value.Length;
if (charCount < BufferSize)
if (charCount < BufferSize - newLineChars)
{
value.CopyTo(0, charBuffer, 0, charCount);
......@@ -46,7 +50,13 @@ namespace Titanium.Web.Proxy.Helpers
try
{
int idx = encoder.GetBytes(charBuffer, 0, charCount, buffer, 0, true);
await WriteAsync(buffer, 0, idx);
if (newLineChars > 0)
{
Buffer.BlockCopy(newLine, 0, buffer, idx, newLineChars);
idx += newLineChars;
}
return WriteAsync(buffer, 0, idx);
}
finally
{
......@@ -55,15 +65,24 @@ namespace Titanium.Web.Proxy.Helpers
}
else
{
var buffer = encoding.GetBytes(value);
await WriteAsync(buffer);
var charBuffer = new char[charCount];
value.CopyTo(0, charBuffer, 0, charCount);
var buffer = new byte[charCount + newLineChars + 1];
int idx = encoder.GetBytes(charBuffer, 0, charCount, buffer, 0, true);
if (newLineChars > 0)
{
Buffer.BlockCopy(newLine, 0, buffer, idx, newLineChars);
idx += newLineChars;
}
return WriteAsync(buffer, 0, idx);
}
}
public async Task WriteLineAsync(string value)
public Task WriteLineAsync(string value)
{
await WriteAsync(value);
await WriteLineAsync();
return WriteAsyncInternal(value, true);
}
/// <summary>
......
......@@ -4,6 +4,7 @@ using System.IO;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using StreamExtended.Helpers;
using Titanium.Web.Proxy.Extensions;
using Titanium.Web.Proxy.Network.Tcp;
......@@ -109,15 +110,115 @@ namespace Titanium.Web.Proxy.Helpers
/// <summary>
/// relays the input clientStream to the server at the specified host name and port with the given httpCmd and headers as prefix
/// Usefull for websocket requests
/// Asynchronous Programming Model, which does not throw exceptions when the socket is closed
/// </summary>
/// <param name="clientStream"></param>
/// <param name="serverStream"></param>
/// <param name="bufferSize"></param>
/// <param name="onDataSend"></param>
/// <param name="onDataReceive"></param>
/// <param name="exceptionFunc"></param>
/// <returns></returns>
internal static async Task SendRaw(Stream clientStream, Stream serverStream, int bufferSize,
Action<byte[], int, int> onDataSend, Action<byte[], int, int> onDataReceive)
internal static async Task SendRawApm(Stream clientStream, Stream serverStream, int bufferSize,
Action<byte[], int, int> onDataSend, Action<byte[], int, int> onDataReceive, Action<Exception> exceptionFunc)
{
var tcs = new TaskCompletionSource<bool>();
var cts = new CancellationTokenSource();
cts.Token.Register(() => tcs.TrySetResult(true));
//Now async relay all server=>client & client=>server data
byte[] clientBuffer = BufferPool.GetBuffer(bufferSize);
byte[] serverBuffer = BufferPool.GetBuffer(bufferSize);
try
{
BeginRead(clientStream, serverStream, clientBuffer, cts, onDataSend, exceptionFunc);
BeginRead(serverStream, clientStream, serverBuffer, cts, onDataReceive, exceptionFunc);
await tcs.Task;
}
finally
{
BufferPool.ReturnBuffer(clientBuffer);
BufferPool.ReturnBuffer(serverBuffer);
}
}
private static void BeginRead(Stream inputStream, Stream outputStream, byte[] buffer, CancellationTokenSource cts, Action<byte[], int, int> onCopy, Action<Exception> exceptionFunc)
{
if (cts.IsCancellationRequested)
{
return;
}
bool readFlag = false;
var readCallback = (AsyncCallback)(ar =>
{
if (cts.IsCancellationRequested || readFlag)
{
return;
}
readFlag = true;
try
{
int read = inputStream.EndRead(ar);
if (read <= 0)
{
cts.Cancel();
return;
}
onCopy?.Invoke(buffer, 0, read);
var writeCallback = (AsyncCallback)(ar2 =>
{
if (cts.IsCancellationRequested)
{
return;
}
try
{
outputStream.EndWrite(ar2);
BeginRead(inputStream, outputStream, buffer, cts, onCopy, exceptionFunc);
}
catch (IOException ex)
{
cts.Cancel();
exceptionFunc(ex);
}
});
outputStream.BeginWrite(buffer, 0, read, writeCallback, null);
}
catch (IOException ex)
{
cts.Cancel();
exceptionFunc(ex);
}
});
var readResult = inputStream.BeginRead(buffer, 0, buffer.Length, readCallback, null);
if (readResult.CompletedSynchronously)
{
readCallback(readResult);
}
}
/// <summary>
/// relays the input clientStream to the server at the specified host name and port with the given httpCmd and headers as prefix
/// Usefull for websocket requests
/// Task-based Asynchronous Pattern
/// </summary>
/// <param name="clientStream"></param>
/// <param name="serverStream"></param>
/// <param name="bufferSize"></param>
/// <param name="onDataSend"></param>
/// <param name="onDataReceive"></param>
/// <param name="exceptionFunc"></param>
/// <returns></returns>
internal static async Task SendRawTap(Stream clientStream, Stream serverStream, int bufferSize,
Action<byte[], int, int> onDataSend, Action<byte[], int, int> onDataReceive, Action<Exception> exceptionFunc)
{
var cts = new CancellationTokenSource();
......
......@@ -194,9 +194,10 @@ namespace Titanium.Web.Proxy
((ConnectResponse)connectArgs.WebSession.Response).ServerHelloInfo = serverHelloInfo;
}
await TcpHelper.SendRaw(clientStream, connection.Stream, BufferSize,
await TcpHelper.SendRawApm(clientStream, connection.Stream, BufferSize,
(buffer, offset, count) => { connectArgs.OnDataSent(buffer, offset, count); },
(buffer, offset, count) => { connectArgs.OnDataReceived(buffer, offset, count); });
(buffer, offset, count) => { connectArgs.OnDataReceived(buffer, offset, count); },
ExceptionFunc);
}
return;
......@@ -441,9 +442,10 @@ namespace Titanium.Web.Proxy
await BeforeResponse.InvokeAsync(this, args, ExceptionFunc);
}
await TcpHelper.SendRaw(clientStream, connection.Stream, BufferSize,
await TcpHelper.SendRawApm(clientStream, connection.Stream, BufferSize,
(buffer, offset, count) => { args.OnDataSent(buffer, offset, count); },
(buffer, offset, count) => { args.OnDataReceived(buffer, offset, count); });
(buffer, offset, count) => { args.OnDataReceived(buffer, offset, count); },
ExceptionFunc);
args.Dispose();
break;
......
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