Skip to content
Permalink
Browse files

Fixes #618

  • Loading branch information...
Splamy committed Aug 11, 2019
1 parent e5f0db8 commit d72e8f73e0fb4609603ac683ab38e05c62f74846
Showing with 15 additions and 19 deletions.
  1. +15 −19 TS3Client/Query/Ts3QueryClient.cs
@@ -69,11 +69,12 @@ public override void Connect(ConnectionData conData)
ConnectionData = conData;

tcpStream = tcpClient.GetStream();
// todo: remove reader and move logic into pipe read stream
tcpReader = new StreamReader(tcpStream, Util.Encoder);
tcpWriter = new StreamWriter(tcpStream, Util.Encoder) { NewLine = "\n" };

for (int i = 0; i < 3; i++)
if(tcpReader.ReadLine() != "TS3")
throw new Ts3Exception("Protocol violation. The stream must start with 'TS3'");
if (string.IsNullOrEmpty(tcpReader.ReadLine()))
tcpReader.ReadLine();
}
catch (SocketException ex) { throw new Ts3Exception("Could not connect.", ex); }
@@ -105,32 +106,30 @@ private async Task NetworkLoop(CancellationToken cancellationToken)
private async Task NetworkToPipeLoopAsync(NetworkStream stream, PipeWriter writer, CancellationToken cancellationToken = default)
{
const int minimumBufferSize = 4096;
var dataReadBuffer = new byte[4096];
#if !(NETCOREAPP2_2 || NETCOREAPP3_0)
var dataReadBuffer = new byte[minimumBufferSize];
#endif

while (!cancellationToken.IsCancellationRequested)
{
try
{
var mem = writer.GetMemory(minimumBufferSize);
#if NETCOREAPP2_2 || NETCOREAPP3_0
int bytesRead = await stream.ReadAsync(mem, cancellationToken).ConfigureAwait(false);
#else
int bytesRead = await stream.ReadAsync(dataReadBuffer, 0, dataReadBuffer.Length, cancellationToken).ConfigureAwait(false);
dataReadBuffer.CopyTo(mem);
#endif
if (bytesRead == 0)
{
break;
}

dataReadBuffer.CopyTo(mem);
//await writer.WriteAsync(dataReadBuffer.AsMemory(0, bytesRead));
//await writer.FlushAsync();
writer.Advance(bytesRead);
}
catch (IOException) { break; }

FlushResult result = await writer.FlushAsync(cancellationToken).ConfigureAwait(false);

if (result.IsCompleted)
{
var result = await writer.FlushAsync(cancellationToken).ConfigureAwait(false);
if (result.IsCompleted || result.IsCanceled)
break;
}
}
writer.Complete();
}
@@ -141,7 +140,7 @@ private async Task PipeProcessorAsync(PipeReader reader, CancellationToken cance
{
var result = await reader.ReadAsync(cancelationToken).ConfigureAwait(false);

ReadOnlySequence<byte> buffer = result.Buffer;
var buffer = result.Buffer;
SequencePosition? position;

do
@@ -162,11 +161,8 @@ private async Task PipeProcessorAsync(PipeReader reader, CancellationToken cance
} while (position != null);

reader.AdvanceTo(buffer.Start, buffer.End);

if (result.IsCompleted)
{
if (result.IsCompleted || result.IsCanceled)
break;
}
}

reader.Complete();

0 comments on commit d72e8f7

Please sign in to comment.
You can’t perform that action at this time.