You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@thrift.apache.org by je...@apache.org on 2021/03/08 08:05:18 UTC
[thrift] branch master updated: misc. netstd improvements Client:
netstd Patch: Jens Geyer
This is an automated email from the ASF dual-hosted git repository.
jensg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/thrift.git
The following commit(s) were added to refs/heads/master by this push:
new 0d12832 misc. netstd improvements Client: netstd Patch: Jens Geyer
0d12832 is described below
commit 0d12832d92351f2cbb2c59edaeb114f789c0ec9b
Author: Jens Geyer <je...@apache.org>
AuthorDate: Thu Feb 25 09:42:52 2021 +0100
misc. netstd improvements
Client: netstd
Patch: Jens Geyer
This closes #2344
---
.gitignore | 2 +
.../Thrift.Benchmarks/Thrift.Benchmarks.csproj | 5 +-
lib/netstd/Thrift/Protocol/TBinaryProtocol.cs | 7 +-
lib/netstd/Thrift/Protocol/TCompactProtocol.cs | 6 +-
lib/netstd/Thrift/Protocol/TJSONProtocol.cs | 32 ++-
.../Thrift/Transport/Client/THttpTransport.cs | 19 +-
.../Thrift/Transport/Client/TNamedPipeTransport.cs | 10 +-
.../Thrift/Transport/Client/TStreamTransport.cs | 10 +-
.../Transport/Server/TNamedPipeServerTransport.cs | 8 +-
tutorial/netstd/Client/Program.cs | 221 ++++++++++-----------
.../netstd/Client/Properties/launchSettings.json | 8 -
tutorial/netstd/Server/Program.cs | 154 +++++++-------
.../netstd/Server/Properties/launchSettings.json | 8 -
13 files changed, 266 insertions(+), 224 deletions(-)
diff --git a/.gitignore b/.gitignore
index 2a4307a..80111b3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -416,5 +416,7 @@ project.lock.json
/tutorial/rs/Cargo.lock
/tutorial/netstd/Interfaces/shared
/tutorial/netstd/Interfaces/tutorial
+/tutorial/netstd/Server/Properties/launchSettings.json
+/tutorial/netstd/Client/Properties/launchSettings.json
/ylwrap
diff --git a/lib/netstd/Benchmarks/Thrift.Benchmarks/Thrift.Benchmarks.csproj b/lib/netstd/Benchmarks/Thrift.Benchmarks/Thrift.Benchmarks.csproj
index baef759..d55074f 100644
--- a/lib/netstd/Benchmarks/Thrift.Benchmarks/Thrift.Benchmarks.csproj
+++ b/lib/netstd/Benchmarks/Thrift.Benchmarks/Thrift.Benchmarks.csproj
@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
<!--
Licensed to the Apache Software Foundation(ASF) under one
or more contributor license agreements.See the NOTICE file
@@ -20,8 +20,9 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
- <TargetFrameworks>net5.0;net48</TargetFrameworks>
+ <TargetFramework>net5.0</TargetFramework>
<TieredCompilation>false</TieredCompilation>
+ <AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects>
</PropertyGroup>
<ItemGroup>
diff --git a/lib/netstd/Thrift/Protocol/TBinaryProtocol.cs b/lib/netstd/Thrift/Protocol/TBinaryProtocol.cs
index 28b7d29..9c23469 100644
--- a/lib/netstd/Thrift/Protocol/TBinaryProtocol.cs
+++ b/lib/netstd/Thrift/Protocol/TBinaryProtocol.cs
@@ -23,6 +23,9 @@ using System.Threading.Tasks;
using Thrift.Protocol.Entities;
using Thrift.Transport;
+#pragma warning disable IDE0079 // unnecessary suppression
+#pragma warning disable IDE0066 // use switch expression
+
namespace Thrift.Protocol
{
// ReSharper disable once InconsistentNaming
@@ -36,7 +39,7 @@ namespace Thrift.Protocol
// minimize memory allocations by means of an preallocated bytes buffer
// The value of 128 is arbitrarily chosen, the required minimum size must be sizeof(long)
- private byte[] PreAllocatedBuffer = new byte[128];
+ private readonly byte[] PreAllocatedBuffer = new byte[128];
public TBinaryProtocol(TTransport trans)
: this(trans, false, true)
@@ -442,7 +445,7 @@ namespace Thrift.Protocol
case TType.Map: return sizeof(int); // element count
case TType.Set: return sizeof(int); // element count
case TType.List: return sizeof(int); // element count
- default: throw new TTransportException(TTransportException.ExceptionType.Unknown, "unrecognized type code");
+ default: throw new TProtocolException(TProtocolException.NOT_IMPLEMENTED, "unrecognized type code");
}
}
diff --git a/lib/netstd/Thrift/Protocol/TCompactProtocol.cs b/lib/netstd/Thrift/Protocol/TCompactProtocol.cs
index 066b327..3758174 100644
--- a/lib/netstd/Thrift/Protocol/TCompactProtocol.cs
+++ b/lib/netstd/Thrift/Protocol/TCompactProtocol.cs
@@ -26,9 +26,11 @@ using System.Threading.Tasks;
using Thrift.Protocol.Entities;
using Thrift.Transport;
+#pragma warning disable IDE0079 // unnecessary suppression
+#pragma warning disable IDE0066 // use switch expression
+
namespace Thrift.Protocol
{
- //TODO: implementation of TProtocol
// ReSharper disable once InconsistentNaming
public class TCompactProtocol : TProtocol
@@ -805,7 +807,7 @@ namespace Thrift.Protocol
case TType.Map: return sizeof(byte); // element count
case TType.Set: return sizeof(byte); // element count
case TType.List: return sizeof(byte); // element count
- default: throw new TTransportException(TTransportException.ExceptionType.Unknown, "unrecognized type code");
+ default: throw new TProtocolException(TProtocolException.NOT_IMPLEMENTED, "unrecognized type code");
}
}
diff --git a/lib/netstd/Thrift/Protocol/TJSONProtocol.cs b/lib/netstd/Thrift/Protocol/TJSONProtocol.cs
index 2f1ccdb..081f42e 100644
--- a/lib/netstd/Thrift/Protocol/TJSONProtocol.cs
+++ b/lib/netstd/Thrift/Protocol/TJSONProtocol.cs
@@ -27,6 +27,10 @@ using Thrift.Protocol.Entities;
using Thrift.Protocol.Utilities;
using Thrift.Transport;
+#pragma warning disable IDE0079 // unnecessary suppression
+#pragma warning disable IDE0063 // simplify using
+#pragma warning disable IDE0066 // use switch expression
+
namespace Thrift.Protocol
{
/// <summary>
@@ -88,7 +92,7 @@ namespace Thrift.Protocol
/// even in cases where the protocol instance was in an undefined state due to
/// dangling/stale/obsolete contexts
/// </summary>
- private void resetContext()
+ private void ResetContext()
{
ContextStack.Clear();
Context = new JSONBaseContext(this);
@@ -277,7 +281,7 @@ namespace Thrift.Protocol
public override async Task WriteMessageBeginAsync(TMessage message, CancellationToken cancellationToken)
{
- resetContext();
+ ResetContext();
await WriteJsonArrayStartAsync(cancellationToken);
await WriteJsonIntegerAsync(Version, cancellationToken);
@@ -430,7 +434,12 @@ namespace Thrift.Protocol
// escaped?
if (ch != TJSONProtocolConstants.EscSequences[0])
{
+#if NETSTANDARD2_0
await buffer.WriteAsync(new[] {ch}, 0, 1, cancellationToken);
+#else
+ var wbuf = new[] { ch };
+ await buffer.WriteAsync(wbuf.AsMemory(0, 1), cancellationToken);
+#endif
continue;
}
@@ -444,7 +453,12 @@ namespace Thrift.Protocol
throw new TProtocolException(TProtocolException.INVALID_DATA, "Expected control char");
}
ch = TJSONProtocolConstants.EscapeCharValues[off];
+#if NETSTANDARD2_0
await buffer.WriteAsync(new[] {ch}, 0, 1, cancellationToken);
+#else
+ var wbuf = new[] { ch };
+ await buffer.WriteAsync( wbuf.AsMemory(0, 1), cancellationToken);
+#endif
continue;
}
@@ -473,13 +487,21 @@ namespace Thrift.Protocol
codeunits.Add((char) wch);
var tmp = Utf8Encoding.GetBytes(codeunits.ToArray());
+#if NETSTANDARD2_0
await buffer.WriteAsync(tmp, 0, tmp.Length, cancellationToken);
+#else
+ await buffer.WriteAsync(tmp.AsMemory(0, tmp.Length), cancellationToken);
+#endif
codeunits.Clear();
}
else
{
- var tmp = Utf8Encoding.GetBytes(new[] {(char) wch});
+ var tmp = Utf8Encoding.GetBytes(new[] { (char)wch });
+#if NETSTANDARD2_0
await buffer.WriteAsync(tmp, 0, tmp.Length, cancellationToken);
+#else
+ await buffer.WriteAsync(tmp.AsMemory( 0, tmp.Length), cancellationToken);
+#endif
}
}
@@ -655,7 +677,7 @@ namespace Thrift.Protocol
{
var message = new TMessage();
- resetContext();
+ ResetContext();
await ReadJsonArrayStartAsync(cancellationToken);
if (await ReadJsonIntegerAsync(cancellationToken) != Version)
{
@@ -816,7 +838,7 @@ namespace Thrift.Protocol
case TType.Map: return 2; // empty map
case TType.Set: return 2; // empty set
case TType.List: return 2; // empty list
- default: throw new TTransportException(TTransportException.ExceptionType.Unknown, "unrecognized type code");
+ default: throw new TProtocolException(TProtocolException.NOT_IMPLEMENTED, "unrecognized type code");
}
}
diff --git a/lib/netstd/Thrift/Transport/Client/THttpTransport.cs b/lib/netstd/Thrift/Transport/Client/THttpTransport.cs
index dcd028c..51f91ce 100644
--- a/lib/netstd/Thrift/Transport/Client/THttpTransport.cs
+++ b/lib/netstd/Thrift/Transport/Client/THttpTransport.cs
@@ -25,6 +25,9 @@ using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
+#pragma warning disable IDE0079 // unnecessary suppression
+#pragma warning disable IDE0063 // simplify using
+
namespace Thrift.Transport.Client
{
// ReSharper disable once InconsistentNaming
@@ -33,7 +36,7 @@ namespace Thrift.Transport.Client
private readonly X509Certificate[] _certificates;
private readonly Uri _uri;
- private int _connectTimeout = 30000; // Timeouts in milliseconds
+ private readonly int _connectTimeout = 30000; // Timeouts in milliseconds
private HttpClient _httpClient;
private Stream _inputStream;
private MemoryStream _outputStream = new MemoryStream();
@@ -133,10 +136,10 @@ namespace Thrift.Transport.Client
try
{
-#if NETSTANDARD2_1
- var ret = await _inputStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
-#else
+#if NETSTANDARD2_0
var ret = await _inputStream.ReadAsync(buffer, offset, length, cancellationToken);
+#else
+ var ret = await _inputStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
#endif
if (ret == -1)
{
@@ -156,7 +159,11 @@ namespace Thrift.Transport.Client
{
cancellationToken.ThrowIfCancellationRequested();
+#if NETSTANDARD2_0
await _outputStream.WriteAsync(buffer, offset, length, cancellationToken);
+#else
+ await _outputStream.WriteAsync(buffer.AsMemory(offset, length), cancellationToken);
+#endif
}
/// <summary>
@@ -224,7 +231,11 @@ namespace Thrift.Transport.Client
var response = (await _httpClient.PostAsync(_uri, contentStream, cancellationToken)).EnsureSuccessStatusCode();
_inputStream?.Dispose();
+#if NETSTANDARD2_0 || NETSTANDARD2_1
_inputStream = await response.Content.ReadAsStreamAsync();
+#else
+ _inputStream = await response.Content.ReadAsStreamAsync(cancellationToken);
+#endif
if (_inputStream.CanSeek)
{
_inputStream.Seek(0, SeekOrigin.Begin);
diff --git a/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs b/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs
index 815983e..47d1ccb 100644
--- a/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs
+++ b/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs
@@ -72,10 +72,10 @@ namespace Thrift.Transport.Client
}
CheckReadBytesAvailable(length);
-#if NETSTANDARD2_1
- var numRead = await PipeStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
-#else
+#if NETSTANDARD2_0
var numRead = await PipeStream.ReadAsync(buffer, offset, length, cancellationToken);
+#else
+ var numRead = await PipeStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
#endif
CountConsumedMessageBytes(numRead);
return numRead;
@@ -94,7 +94,11 @@ namespace Thrift.Transport.Client
var nBytes = Math.Min(15 * 4096, length); // 16 would exceed the limit
while (nBytes > 0)
{
+#if NETSTANDARD2_0
await PipeStream.WriteAsync(buffer, offset, nBytes, cancellationToken);
+#else
+ await PipeStream.WriteAsync(buffer.AsMemory(offset, nBytes), cancellationToken);
+#endif
offset += nBytes;
length -= nBytes;
nBytes = Math.Min(nBytes, length);
diff --git a/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs b/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs
index b397460..90794c6 100644
--- a/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs
+++ b/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs
@@ -80,10 +80,10 @@ namespace Thrift.Transport.Client
"Cannot read from null inputstream");
}
-#if NETSTANDARD2_1
- return await InputStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
-#else
+#if NETSTANDARD2_0
return await InputStream.ReadAsync(buffer, offset, length, cancellationToken);
+#else
+ return await InputStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
#endif
}
@@ -95,7 +95,11 @@ namespace Thrift.Transport.Client
"Cannot write to null outputstream");
}
+#if NETSTANDARD2_0
await OutputStream.WriteAsync(buffer, offset, length, cancellationToken);
+#else
+ await OutputStream.WriteAsync(buffer.AsMemory(offset, length), cancellationToken);
+#endif
}
public override async Task FlushAsync(CancellationToken cancellationToken)
diff --git a/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs b/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs
index 5698776..67ba29f 100644
--- a/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs
+++ b/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs
@@ -279,12 +279,10 @@ namespace Thrift.Transport.Server
}
CheckReadBytesAvailable(length);
-#if NET5_0
- var numBytes = await PipeStream.ReadAsync(buffer.AsMemory(offset, length), cancellationToken);
-#elif NETSTANDARD2_1
- var numBytes = await PipeStream.ReadAsync(new Memory<byte>(buffer, offset, length), cancellationToken);
-#else
+#if NETSTANDARD2_0
var numBytes = await PipeStream.ReadAsync(buffer, offset, length, cancellationToken);
+#else
+ var numBytes = await PipeStream.ReadAsync(buffer.AsMemory(offset, length), cancellationToken);
#endif
CountConsumedMessageBytes(numBytes);
return numBytes;
diff --git a/tutorial/netstd/Client/Program.cs b/tutorial/netstd/Client/Program.cs
index abbba70..f3e93ce 100644
--- a/tutorial/netstd/Client/Program.cs
+++ b/tutorial/netstd/Client/Program.cs
@@ -34,11 +34,14 @@ using Thrift.Transport.Client;
using tutorial;
using shared;
+#pragma warning disable IDE0063 // using
+#pragma warning disable IDE0057 // substr
+
namespace Client
{
public class Program
{
- private static ServiceCollection ServiceCollection = new ServiceCollection();
+ private static readonly ServiceCollection ServiceCollection = new ServiceCollection();
private static ILogger Logger;
private static readonly TConfiguration Configuration = null; // new TConfiguration() if needed
@@ -49,26 +52,27 @@ Usage:
Client -help
will diplay help information
- Client -tr:<transport> -bf:<buffering> -pr:<protocol> -mc:<numClients>
+ Client -tr:<transport> -bf:<buffering> -pr:<protocol> [-mc:<numClients>] [-multiplex]
will run client with specified arguments (tcp transport and binary protocol by default) and with 1 client
Options:
-tr (transport):
- tcp - (default) tcp transport will be used (host - ""localhost"", port - 9090)
- namedpipe - namedpipe transport will be used (pipe address - "".test"")
- http - http transport will be used (address - ""http://localhost:9090"")
- tcptls - tcp tls transport will be used (host - ""localhost"", port - 9090)
+ tcp - (default) tcp transport (localhost:9090)
+ tcptls - tcp tls transport (localhost:9090)
+ namedpipe - namedpipe transport (pipe "".test"")
+ http - http transport (http://localhost:9090)
-bf (buffering):
- none - (default) no buffering will be used
- buffered - buffered transport will be used
- framed - framed transport will be used
+ none - (default) no buffering
+ buffered - buffered transport
+ framed - framed transport
-pr (protocol):
- binary - (default) binary protocol will be used
- compact - compact protocol will be used
- json - json protocol will be used
- multiplexed - multiplexed protocol will be used
+ binary - (default) binary protocol
+ compact - compact protocol
+ json - json protocol
+
+ -multiplex - adds multiplexed protocol
-mc (multiple clients):
<numClients> - number of multiple clients to connect to server (max 100, default 1)
@@ -80,7 +84,7 @@ Sample:
public static void Main(string[] args)
{
- args = args ?? new string[0];
+ args ??= Array.Empty<string>();
ServiceCollection.AddLogging(logging => ConfigureLogging(logging));
using (var serviceProvider = ServiceCollection.BuildServiceProvider())
@@ -115,43 +119,77 @@ Sample:
Logger.LogInformation($"Selected # of clients: {numClients}");
- var transports = new TTransport[numClients];
- for (int i = 0; i < numClients; i++)
- {
- var t = GetTransport(args);
- transports[i] = t;
- }
-
- Logger.LogInformation($"Selected client transport: {transports[0]}");
+ var transport = GetTransport(args);
+ Logger.LogInformation($"Selected client transport: {transport}");
- var protocols = new Tuple<Protocol, TProtocol>[numClients];
- for (int i = 0; i < numClients; i++)
- {
- var p = GetProtocol(args, transports[i]);
- protocols[i] = p;
- }
+ var protocol = MakeProtocol( args, MakeTransport(args));
+ Logger.LogInformation($"Selected client protocol: {GetProtocol(args)}");
- Logger.LogInformation($"Selected client protocol: {protocols[0].Item1}");
+ var mplex = GetMultiplex(args);
+ Logger.LogInformation("Multiplex " + (mplex ? "yes" : "no"));
var tasks = new Task[numClients];
for (int i = 0; i < numClients; i++)
{
- var task = RunClientAsync(protocols[i], cancellationToken);
+ var task = RunClientAsync(protocol, mplex, cancellationToken);
tasks[i] = task;
}
- Task.WaitAll(tasks);
-
+ Task.WaitAll(tasks,cancellationToken);
await Task.CompletedTask;
}
- private static TTransport GetTransport(string[] args)
+ private static bool GetMultiplex(string[] args)
{
- TTransport transport = new TSocketTransport(IPAddress.Loopback, 9090, Configuration);
+ var mplex = args.FirstOrDefault(x => x.StartsWith("-multiplex"));
+ return !string.IsNullOrEmpty(mplex);
+ }
+ private static Protocol GetProtocol(string[] args)
+ {
+ var protocol = args.FirstOrDefault(x => x.StartsWith("-pr"))?.Split(':')?[1];
+ if (string.IsNullOrEmpty(protocol))
+ return Protocol.Binary;
+
+ protocol = protocol.Substring(0, 1).ToUpperInvariant() + protocol.Substring(1).ToLowerInvariant();
+ if (Enum.TryParse(protocol, true, out Protocol selectedProtocol))
+ return selectedProtocol;
+ else
+ return Protocol.Binary;
+ }
+
+ private static Buffering GetBuffering(string[] args)
+ {
+ var buffering = args.FirstOrDefault(x => x.StartsWith("-bf"))?.Split(":")?[1];
+ if (string.IsNullOrEmpty(buffering))
+ return Buffering.None;
+
+ buffering = buffering.Substring(0, 1).ToUpperInvariant() + buffering.Substring(1).ToLowerInvariant();
+ if (Enum.TryParse<Buffering>(buffering, out var selectedBuffering))
+ return selectedBuffering;
+ else
+ return Buffering.None;
+ }
+
+ private static Transport GetTransport(string[] args)
+ {
+ var transport = args.FirstOrDefault(x => x.StartsWith("-tr"))?.Split(':')?[1];
+ if (string.IsNullOrEmpty(transport))
+ return Transport.Tcp;
+
+ transport = transport.Substring(0, 1).ToUpperInvariant() + transport.Substring(1).ToLowerInvariant();
+ if (Enum.TryParse(transport, true, out Transport selectedTransport))
+ return selectedTransport;
+ else
+ return Transport.Tcp;
+ }
+
+
+ private static TTransport MakeTransport(string[] args)
+ {
// construct endpoint transport
- var transportArg = args.FirstOrDefault(x => x.StartsWith("-tr"))?.Split(':')?[1];
- if (Enum.TryParse(transportArg, true, out Transport selectedTransport))
+ TTransport transport = null;
+ Transport selectedTransport = GetTransport(args);
{
switch (selectedTransport)
{
@@ -179,23 +217,20 @@ Sample:
}
// optionally add layered transport(s)
- var bufferingArg = args.FirstOrDefault(x => x.StartsWith("-bf"))?.Split(':')?[1];
- if (Enum.TryParse<Buffering>(bufferingArg, out var selectedBuffering))
+ Buffering selectedBuffering = GetBuffering(args);
+ switch (selectedBuffering)
{
- switch (selectedBuffering)
- {
- case Buffering.Buffered:
- transport = new TBufferedTransport(transport);
- break;
+ case Buffering.Buffered:
+ transport = new TBufferedTransport(transport);
+ break;
- case Buffering.Framed:
- transport = new TFramedTransport(transport);
- break;
+ case Buffering.Framed:
+ transport = new TFramedTransport(transport);
+ break;
- default: // layered transport(s) are optional
- Debug.Assert(selectedBuffering == Buffering.None, "unhandled case");
- break;
- }
+ default: // layered transport(s) are optional
+ Debug.Assert(selectedBuffering == Buffering.None, "unhandled case");
+ break;
}
return transport;
@@ -207,11 +242,10 @@ Sample:
Logger.LogInformation($"Selected # of clients: {numClients}");
- int c;
- if( int.TryParse(numClients, out c) && (0 < c) && (c <= 100))
- return c;
- else
- return 1;
+ if (int.TryParse(numClients, out int c) && (0 < c) && (c <= 100))
+ return c;
+ else
+ return 1;
}
private static X509Certificate2 GetCertificate()
@@ -250,65 +284,37 @@ Sample:
return true;
}
- private static Tuple<Protocol, TProtocol> GetProtocol(string[] args, TTransport transport)
+ private static TProtocol MakeProtocol(string[] args, TTransport transport)
{
- var protocol = args.FirstOrDefault(x => x.StartsWith("-pr"))?.Split(':')?[1];
-
- Protocol selectedProtocol;
- if (Enum.TryParse(protocol, true, out selectedProtocol))
+ Protocol selectedProtocol = GetProtocol(args);
+ switch (selectedProtocol)
{
- switch (selectedProtocol)
- {
- case Protocol.Binary:
- return new Tuple<Protocol, TProtocol>(selectedProtocol, new TBinaryProtocol(transport));
- case Protocol.Compact:
- return new Tuple<Protocol, TProtocol>(selectedProtocol, new TCompactProtocol(transport));
- case Protocol.Json:
- return new Tuple<Protocol, TProtocol>(selectedProtocol, new TJsonProtocol(transport));
- case Protocol.Multiplexed:
- // it returns BinaryProtocol to avoid making wrapped protocol as public in TProtocolDecorator (in RunClientAsync it will be wrapped into Multiplexed protocol)
- return new Tuple<Protocol, TProtocol>(selectedProtocol, new TBinaryProtocol(transport));
- default:
- Debug.Assert(false, "unhandled case");
- break;
- }
+ case Protocol.Binary:
+ return new TBinaryProtocol(transport);
+ case Protocol.Compact:
+ return new TCompactProtocol(transport);
+ case Protocol.Json:
+ return new TJsonProtocol(transport);
+ default:
+ throw new Exception("unhandled protocol");
}
-
- return new Tuple<Protocol, TProtocol>(selectedProtocol, new TBinaryProtocol(transport));
}
- private static async Task RunClientAsync(Tuple<Protocol, TProtocol> protocolTuple, CancellationToken cancellationToken)
+ private static async Task RunClientAsync(TProtocol protocol, bool multiplex, CancellationToken cancellationToken)
{
try
{
- var protocol = protocolTuple.Item2;
- var protocolType = protocolTuple.Item1;
-
- TBaseClient client = null;
-
try
{
- if (protocolType != Protocol.Multiplexed)
- {
-
- client = new Calculator.Client(protocol);
- await ExecuteCalculatorClientOperations(cancellationToken, (Calculator.Client)client);
- }
- else
- {
- // it uses binary protocol there to create Multiplexed protocols
- var multiplex = new TMultiplexedProtocol(protocol, nameof(Calculator));
- client = new Calculator.Client(multiplex);
- await ExecuteCalculatorClientOperations(cancellationToken, (Calculator.Client)client);
-
- multiplex = new TMultiplexedProtocol(protocol, nameof(SharedService));
- client = new SharedService.Client(multiplex);
- await ExecuteSharedServiceClientOperations(cancellationToken, (SharedService.Client)client);
- }
+ if( multiplex)
+ protocol = new TMultiplexedProtocol(protocol, nameof(Calculator));
+
+ var client = new Calculator.Client(protocol);
+ await ExecuteCalculatorClientOperations(client, cancellationToken);
}
catch (Exception ex)
{
- Logger.LogError($"{client?.ClientId} " + ex);
+ Logger.LogError(ex.ToString());
}
finally
{
@@ -321,7 +327,7 @@ Sample:
}
}
- private static async Task ExecuteCalculatorClientOperations(CancellationToken cancellationToken, Calculator.Client client)
+ private static async Task ExecuteCalculatorClientOperations( Calculator.Client client, CancellationToken cancellationToken)
{
await client.OpenTransportAsync(cancellationToken);
@@ -374,16 +380,6 @@ Sample:
Logger.LogInformation($"{client.ClientId} ZipAsync() with delay 100mc on server side");
await client.zipAsync(cancellationToken);
}
- private static async Task ExecuteSharedServiceClientOperations(CancellationToken cancellationToken, SharedService.Client client)
- {
- await client.OpenTransportAsync(cancellationToken);
-
- // Async version
-
- Logger.LogInformation($"{client.ClientId} SharedService GetStructAsync(1)");
- var log = await client.getStructAsync(1, cancellationToken);
- Logger.LogInformation($"{client.ClientId} SharedService Value: {log.Value}");
- }
private enum Transport
@@ -401,7 +397,6 @@ Sample:
Binary,
Compact,
Json,
- Multiplexed
}
private enum Buffering
diff --git a/tutorial/netstd/Client/Properties/launchSettings.json b/tutorial/netstd/Client/Properties/launchSettings.json
deleted file mode 100644
index 6b7b60d..0000000
--- a/tutorial/netstd/Client/Properties/launchSettings.json
+++ /dev/null
@@ -1,8 +0,0 @@
-{
- "profiles": {
- "Client": {
- "commandName": "Project",
- "commandLineArgs": "-p:multiplexed"
- }
- }
-}
\ No newline at end of file
diff --git a/tutorial/netstd/Server/Program.cs b/tutorial/netstd/Server/Program.cs
index 3181e8e..df8f390 100644
--- a/tutorial/netstd/Server/Program.cs
+++ b/tutorial/netstd/Server/Program.cs
@@ -38,17 +38,20 @@ using shared;
using Thrift.Processor;
using System.Diagnostics;
+#pragma warning disable IDE0063 // using
+#pragma warning disable IDE0057 // substr
+
namespace Server
{
public class Program
{
- private static ServiceCollection ServiceCollection = new ServiceCollection();
+ private static readonly ServiceCollection ServiceCollection = new ServiceCollection();
private static ILogger Logger;
private static readonly TConfiguration Configuration = null; // new TConfiguration() if needed
public static void Main(string[] args)
{
- args = args ?? new string[0];
+ args ??= Array.Empty<string>();
ServiceCollection.AddLogging(logging => ConfigureLogging(logging));
using (var serviceProvider = ServiceCollection.BuildServiceProvider())
@@ -89,26 +92,27 @@ Usage:
Server -help
will diplay help information
- Server -tr:<transport> -bf:<buffering> -pr:<protocol>
+ Server -tr:<transport> -bf:<buffering> -pr:<protocol> [-multiplex]
will run server with specified arguments (tcp transport, no buffering, and binary protocol by default)
Options:
-tr (transport):
- tcp - (default) tcp transport will be used (host - ""localhost"", port - 9090)
- namedpipe - namedpipe transport will be used (pipe address - "".test"")
- http - http transport will be used (http address - ""localhost:9090"")
- tcptls - tcp transport with tls will be used (host - ""localhost"", port - 9090)
+ tcp - (default) tcp transport (localhost:9090)
+ tcptls - tcp transport with tls (localhost:9090)
+ namedpipe - namedpipe transport (pipe "".test"")
+ http - http transport (localhost:9090)
-bf (buffering):
- none - (default) no buffering will be used
- buffered - buffered transport will be used
- framed - framed transport will be used
+ none - (default) no buffering
+ buffered - buffered transport
+ framed - framed transport
-pr (protocol):
- binary - (default) binary protocol will be used
- compact - compact protocol will be used
- json - json protocol will be used
- multiplexed - multiplexed protocol will be used
+ binary - (default) binary protocol
+ compact - compact protocol
+ json - json protocol
+
+ -multiplex - adds multiplexed protocol
Sample:
Server -tr:tcp
@@ -120,48 +124,68 @@ Sample:
var selectedTransport = GetTransport(args);
var selectedBuffering = GetBuffering(args);
var selectedProtocol = GetProtocol(args);
+ var multiplex = GetMultiplex(args);
if (selectedTransport == Transport.Http)
{
+ if (multiplex)
+ throw new Exception("This tutorial semple code does not yet allow multiplex over http (although Thrift itself of course does)");
new HttpServerSample().Run(cancellationToken);
}
else
{
- await RunSelectedConfigurationAsync(selectedTransport, selectedBuffering, selectedProtocol, cancellationToken);
+ await RunSelectedConfigurationAsync(selectedTransport, selectedBuffering, selectedProtocol, multiplex, cancellationToken);
}
}
- private static Protocol GetProtocol(string[] args)
+
+ private static bool GetMultiplex(string[] args)
{
- var transport = args.FirstOrDefault(x => x.StartsWith("-pr"))?.Split(':')?[1];
+ var mplex = args.FirstOrDefault(x => x.StartsWith("-multiplex"));
+ return !string.IsNullOrEmpty(mplex);
+ }
- Enum.TryParse(transport, true, out Protocol selectedProtocol);
+ private static Protocol GetProtocol(string[] args)
+ {
+ var protocol = args.FirstOrDefault(x => x.StartsWith("-pr"))?.Split(':')?[1];
+ if (string.IsNullOrEmpty(protocol))
+ return Protocol.Binary;
- return selectedProtocol;
+ protocol = protocol.Substring(0, 1).ToUpperInvariant() + protocol.Substring(1).ToLowerInvariant();
+ if (Enum.TryParse(protocol, true, out Protocol selectedProtocol))
+ return selectedProtocol;
+ else
+ return Protocol.Binary;
}
private static Buffering GetBuffering(string[] args)
{
var buffering = args.FirstOrDefault(x => x.StartsWith("-bf"))?.Split(":")?[1];
+ if (string.IsNullOrEmpty(buffering))
+ return Buffering.None;
- Enum.TryParse<Buffering>(buffering, out var selectedBuffering);
-
- return selectedBuffering;
+ buffering = buffering.Substring(0, 1).ToUpperInvariant() + buffering.Substring(1).ToLowerInvariant();
+ if( Enum.TryParse<Buffering>(buffering, out var selectedBuffering))
+ return selectedBuffering;
+ else
+ return Buffering.None;
}
private static Transport GetTransport(string[] args)
{
var transport = args.FirstOrDefault(x => x.StartsWith("-tr"))?.Split(':')?[1];
+ if (string.IsNullOrEmpty(transport))
+ return Transport.Tcp;
- Enum.TryParse(transport, true, out Transport selectedTransport);
-
- return selectedTransport;
+ transport = transport.Substring(0, 1).ToUpperInvariant() + transport.Substring(1).ToLowerInvariant();
+ if( Enum.TryParse(transport, true, out Transport selectedTransport))
+ return selectedTransport;
+ else
+ return Transport.Tcp;
}
- private static async Task RunSelectedConfigurationAsync(Transport transport, Buffering buffering, Protocol protocol, CancellationToken cancellationToken)
+ private static async Task RunSelectedConfigurationAsync(Transport transport, Buffering buffering, Protocol protocol, bool multiplex, CancellationToken cancellationToken)
{
- var handler = new CalculatorAsyncHandler();
-
TServerTransport serverTransport = null;
switch (transport)
{
@@ -177,18 +201,15 @@ Sample:
break;
}
- TTransportFactory inputTransportFactory = null;
- TTransportFactory outputTransportFactory = null;
+ TTransportFactory transportFactory = null;
switch (buffering)
{
case Buffering.Buffered:
- inputTransportFactory = new TBufferedTransport.Factory();
- outputTransportFactory = new TBufferedTransport.Factory();
+ transportFactory = new TBufferedTransport.Factory();
break;
case Buffering.Framed:
- inputTransportFactory = new TFramedTransport.Factory();
- outputTransportFactory = new TFramedTransport.Factory();
+ transportFactory = new TFramedTransport.Factory();
break;
default: // layered transport(s) are optional
@@ -196,65 +217,57 @@ Sample:
break;
}
- TProtocolFactory inputProtocolFactory = null;
- TProtocolFactory outputProtocolFactory = null;
- ITAsyncProcessor processor = null;
+ TProtocolFactory protocolFactory = null;
switch (protocol)
{
case Protocol.Binary:
- inputProtocolFactory = new TBinaryProtocol.Factory();
- outputProtocolFactory = new TBinaryProtocol.Factory();
- processor = new Calculator.AsyncProcessor(handler);
+ protocolFactory = new TBinaryProtocol.Factory();
break;
case Protocol.Compact:
- inputProtocolFactory = new TCompactProtocol.Factory();
- outputProtocolFactory = new TCompactProtocol.Factory();
- processor = new Calculator.AsyncProcessor(handler);
+ protocolFactory = new TCompactProtocol.Factory();
break;
case Protocol.Json:
- inputProtocolFactory = new TJsonProtocol.Factory();
- outputProtocolFactory = new TJsonProtocol.Factory();
- processor = new Calculator.AsyncProcessor(handler);
+ protocolFactory = new TJsonProtocol.Factory();
break;
- case Protocol.Multiplexed:
- inputProtocolFactory = new TBinaryProtocol.Factory();
- outputProtocolFactory = new TBinaryProtocol.Factory();
-
- var calcHandler = new CalculatorAsyncHandler();
- var calcProcessor = new Calculator.AsyncProcessor(calcHandler);
-
- var sharedServiceHandler = new SharedServiceAsyncHandler();
- var sharedServiceProcessor = new SharedService.AsyncProcessor(sharedServiceHandler);
+ default:
+ throw new ArgumentOutOfRangeException(nameof(protocol), protocol, null);
+ }
- var multiplexedProcessor = new TMultiplexedProcessor();
- multiplexedProcessor.RegisterProcessor(nameof(Calculator), calcProcessor);
- multiplexedProcessor.RegisterProcessor(nameof(SharedService), sharedServiceProcessor);
+ var handler = new CalculatorAsyncHandler();
+ ITAsyncProcessor processor = new Calculator.AsyncProcessor(handler);
- processor = multiplexedProcessor;
- break;
+ if (multiplex)
+ {
+ var multiplexedProcessor = new TMultiplexedProcessor();
+ multiplexedProcessor.RegisterProcessor(nameof(Calculator), processor);
- default:
- throw new ArgumentOutOfRangeException(nameof(protocol), protocol, null);
+ processor = multiplexedProcessor;
}
try
{
Logger.LogInformation(
- $"Selected TAsyncServer with {serverTransport} transport, {processor} processor and {inputProtocolFactory} protocol factories");
+ string.Format(
+ "TSimpleAsyncServer with \n{0} transport\n{1} buffering\nmultiplex = {2}\n{3} protocol",
+ transport,
+ buffering,
+ multiplex ? "yes" : "no",
+ protocol
+ ));
var loggerFactory = ServiceCollection.BuildServiceProvider().GetService<ILoggerFactory>();
var server = new TSimpleAsyncServer(
itProcessorFactory: new TSingletonProcessorFactory(processor),
serverTransport: serverTransport,
- inputTransportFactory: inputTransportFactory,
- outputTransportFactory: outputTransportFactory,
- inputProtocolFactory: inputProtocolFactory,
- outputProtocolFactory: outputProtocolFactory,
+ inputTransportFactory: transportFactory,
+ outputTransportFactory: transportFactory,
+ inputProtocolFactory: protocolFactory,
+ outputProtocolFactory: protocolFactory,
logger: loggerFactory.CreateLogger<TSimpleAsyncServer>());
Logger.LogInformation("Starting the server...");
@@ -323,7 +336,6 @@ Sample:
Binary,
Compact,
Json,
- Multiplexed
}
public class HttpServerSample
@@ -364,6 +376,8 @@ Sample:
// This method gets called by the runtime. Use this method to add services to the container.
public void ConfigureServices(IServiceCollection services)
{
+ // NOTE: this is not really the recommended way to do it
+ // because the HTTP server cannot be configured properly to e.g. accept framed or multiplex
services.AddTransient<Calculator.IAsync, CalculatorAsyncHandler>();
services.AddTransient<ITAsyncProcessor, Calculator.AsyncProcessor>();
services.AddTransient<THttpServerTransport, THttpServerTransport>();
@@ -372,6 +386,8 @@ Sample:
// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
public void Configure(IApplicationBuilder app, IWebHostEnvironment env, ILoggerFactory loggerFactory)
{
+ _ = env;
+ _ = loggerFactory;
app.UseMiddleware<THttpServerTransport>();
}
}
@@ -408,7 +424,7 @@ Sample:
{
Logger.LogInformation($"CalculateAsync({logid}, [{w.Op},{w.Num1},{w.Num2}])");
- var val = 0;
+ int val;
switch (w.Op)
{
case Operation.ADD:
diff --git a/tutorial/netstd/Server/Properties/launchSettings.json b/tutorial/netstd/Server/Properties/launchSettings.json
deleted file mode 100644
index 78076ff..0000000
--- a/tutorial/netstd/Server/Properties/launchSettings.json
+++ /dev/null
@@ -1,8 +0,0 @@
-{
- "profiles": {
- "Server": {
- "commandName": "Project",
- "commandLineArgs": "-p:multiplexed"
- }
- }
-}
\ No newline at end of file