diff options
Diffstat (limited to 'src/jaegertracing/thrift/lib/netstd/Thrift/Transport')
17 files changed, 2549 insertions, 0 deletions
diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/THttpTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/THttpTransport.cs new file mode 100644 index 000000000..c84df83ae --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/THttpTransport.cs @@ -0,0 +1,222 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Net.Http; +using System.Net.Http.Headers; +using System.Security.Cryptography.X509Certificates; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + // ReSharper disable once InconsistentNaming + public class THttpTransport : TTransport + { + private readonly X509Certificate[] _certificates; + private readonly Uri _uri; + + private int _connectTimeout = 30000; // Timeouts in milliseconds + private HttpClient _httpClient; + private Stream _inputStream; + private MemoryStream _outputStream = new MemoryStream(); + private bool _isDisposed; + + public THttpTransport(Uri uri, IDictionary<string, string> customRequestHeaders = null, string userAgent = null) + : this(uri, Enumerable.Empty<X509Certificate>(), customRequestHeaders, userAgent) + { + } + + public THttpTransport(Uri uri, IEnumerable<X509Certificate> certificates, + IDictionary<string, string> customRequestHeaders, string userAgent = null) + { + _uri = uri; + _certificates = (certificates ?? Enumerable.Empty<X509Certificate>()).ToArray(); + + if (!string.IsNullOrEmpty(userAgent)) + UserAgent = userAgent; + + // due to current bug with performance of Dispose in netcore https://github.com/dotnet/corefx/issues/8809 + // this can be switched to default way (create client->use->dispose per flush) later + _httpClient = CreateClient(customRequestHeaders); + } + + // According to RFC 2616 section 3.8, the "User-Agent" header may not carry a version number + public readonly string UserAgent = "Thrift netstd THttpClient"; + + public override bool IsOpen => true; + + public HttpRequestHeaders RequestHeaders => _httpClient.DefaultRequestHeaders; + + public MediaTypeHeaderValue ContentType { get; set; } + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override void Close() + { + if (_inputStream != null) + { + _inputStream.Dispose(); + _inputStream = null; + } + + if (_outputStream != null) + { + _outputStream.Dispose(); + _outputStream = null; + } + + if (_httpClient != null) + { + _httpClient.Dispose(); + _httpClient = null; + } + } + + public override async ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<int>(cancellationToken); + } + + if (_inputStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "No request has been sent"); + } + + try + { + var ret = await _inputStream.ReadAsync(buffer, offset, length, cancellationToken); + + if (ret == -1) + { + throw new TTransportException(TTransportException.ExceptionType.EndOfFile, "No more data available"); + } + + return ret; + } + catch (IOException iox) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString()); + } + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + await _outputStream.WriteAsync(buffer, offset, length, cancellationToken); + } + + private HttpClient CreateClient(IDictionary<string, string> customRequestHeaders) + { + var handler = new HttpClientHandler(); + handler.ClientCertificates.AddRange(_certificates); + handler.AutomaticDecompression = System.Net.DecompressionMethods.Deflate | System.Net.DecompressionMethods.GZip; + + var httpClient = new HttpClient(handler); + + if (_connectTimeout > 0) + { + httpClient.Timeout = TimeSpan.FromMilliseconds(_connectTimeout); + } + + httpClient.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/x-thrift")); + httpClient.DefaultRequestHeaders.UserAgent.TryParseAdd(UserAgent); + + httpClient.DefaultRequestHeaders.AcceptEncoding.Add(new StringWithQualityHeaderValue("deflate")); + httpClient.DefaultRequestHeaders.AcceptEncoding.Add(new StringWithQualityHeaderValue("gzip")); + + if (customRequestHeaders != null) + { + foreach (var item in customRequestHeaders) + { + httpClient.DefaultRequestHeaders.Add(item.Key, item.Value); + } + } + + return httpClient; + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + try + { + _outputStream.Seek(0, SeekOrigin.Begin); + + using (var contentStream = new StreamContent(_outputStream)) + { + contentStream.Headers.ContentType = ContentType ?? new MediaTypeHeaderValue(@"application/x-thrift"); + + var response = (await _httpClient.PostAsync(_uri, contentStream, cancellationToken)).EnsureSuccessStatusCode(); + + _inputStream?.Dispose(); + _inputStream = await response.Content.ReadAsStreamAsync(); + if (_inputStream.CanSeek) + { + _inputStream.Seek(0, SeekOrigin.Begin); + } + } + } + catch (IOException iox) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, iox.ToString()); + } + catch (HttpRequestException wx) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, + "Couldn't connect to server: " + wx); + } + catch (Exception ex) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, ex.Message); + } + finally + { + _outputStream = new MemoryStream(); + } + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + _inputStream?.Dispose(); + _outputStream?.Dispose(); + _httpClient?.Dispose(); + } + } + _isDisposed = true; + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TMemoryBufferTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TMemoryBufferTransport.cs new file mode 100644 index 000000000..25895c2b7 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TMemoryBufferTransport.cs @@ -0,0 +1,179 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Diagnostics; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + // ReSharper disable once InconsistentNaming + public class TMemoryBufferTransport : TTransport + { + private bool IsDisposed; + private byte[] Bytes; + private int _bytesUsed; + + public TMemoryBufferTransport() + { + Bytes = new byte[2048]; // default size + } + + public TMemoryBufferTransport(int initialCapacity) + { + Bytes = new byte[initialCapacity]; // default size + } + + public TMemoryBufferTransport(byte[] buf) + { + Bytes = (byte[])buf.Clone(); + _bytesUsed = Bytes.Length; + } + + public int Position { get; set; } + + public int Capacity + { + get + { + Debug.Assert(_bytesUsed <= Bytes.Length); + return Bytes.Length; + } + set + { + Array.Resize(ref Bytes, value); + _bytesUsed = value; + } + } + + public int Length + { + get { + Debug.Assert(_bytesUsed <= Bytes.Length); + return _bytesUsed; + } + set { + if ((Bytes.Length < value) || (Bytes.Length > (10 * value))) + Array.Resize(ref Bytes, Math.Max(2048, (int)(value * 1.25))); + _bytesUsed = value; + } + } + + public void SetLength(int value) + { + Length = value; + Position = Math.Min(Position, value); + } + + public override bool IsOpen => true; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override void Close() + { + /** do nothing **/ + } + + public void Seek(int delta, SeekOrigin origin) + { + int newPos; + switch (origin) + { + case SeekOrigin.Begin: + newPos = delta; + break; + case SeekOrigin.Current: + newPos = Position + delta; + break; + case SeekOrigin.End: + newPos = _bytesUsed + delta; + break; + default: + throw new ArgumentException(nameof(origin)); + } + + if ((0 > newPos) || (newPos > _bytesUsed)) + throw new ArgumentException(nameof(origin)); + Position = newPos; + } + + public override ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + var count = Math.Min(Length - Position, length); + Buffer.BlockCopy(Bytes, Position, buffer, offset, count); + Position += count; + return new ValueTask<int>(count); + } + + public override Task WriteAsync(byte[] buffer, CancellationToken cancellationToken) + { + return WriteAsync(buffer, 0, buffer.Length, cancellationToken); + } + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + var free = Length - Position; + Length = Length + count - free; + Buffer.BlockCopy(buffer, offset, Bytes, Position, count); + Position += count; + return Task.CompletedTask; + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public byte[] GetBuffer() + { + var retval = new byte[Length]; + Buffer.BlockCopy(Bytes, 0, retval, 0, Length); + return retval; + } + + internal bool TryGetBuffer(out ArraySegment<byte> bufSegment) + { + bufSegment = new ArraySegment<byte>(Bytes, 0, _bytesUsed); + return true; + } + + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!IsDisposed) + { + if (disposing) + { + // nothing to do + } + } + IsDisposed = true; + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs new file mode 100644 index 000000000..7dfe0131e --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TNamedPipeTransport.cs @@ -0,0 +1,108 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.IO.Pipes; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + // ReSharper disable once InconsistentNaming + public class TNamedPipeTransport : TTransport + { + private NamedPipeClientStream PipeStream; + private int ConnectTimeout; + + public TNamedPipeTransport(string pipe, int timeout = Timeout.Infinite) + : this(".", pipe, timeout) + { + } + + public TNamedPipeTransport(string server, string pipe, int timeout = Timeout.Infinite) + { + var serverName = string.IsNullOrWhiteSpace(server) ? server : "."; + ConnectTimeout = (timeout > 0) ? timeout : Timeout.Infinite; + + PipeStream = new NamedPipeClientStream(serverName, pipe, PipeDirection.InOut, PipeOptions.None); + } + + public override bool IsOpen => PipeStream != null && PipeStream.IsConnected; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.AlreadyOpen); + } + + await PipeStream.ConnectAsync( ConnectTimeout, cancellationToken); + } + + public override void Close() + { + if (PipeStream != null) + { + PipeStream.Dispose(); + PipeStream = null; + } + } + + public override async ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (PipeStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + return await PipeStream.ReadAsync(buffer, offset, length, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (PipeStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + // if necessary, send the data in chunks + // there's a system limit around 0x10000 bytes that we hit otherwise + // MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section." + var nBytes = Math.Min(15 * 4096, length); // 16 would exceed the limit + while (nBytes > 0) + { + await PipeStream.WriteAsync(buffer, offset, nBytes, cancellationToken); + offset += nBytes; + length -= nBytes; + nBytes = Math.Min(nBytes, length); + } + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + protected override void Dispose(bool disposing) + { + PipeStream.Dispose(); + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TSocketTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TSocketTransport.cs new file mode 100644 index 000000000..00da04581 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TSocketTransport.cs @@ -0,0 +1,162 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + // ReSharper disable once InconsistentNaming + public class TSocketTransport : TStreamTransport + { + private bool _isDisposed; + + + public TSocketTransport(TcpClient client) + { + TcpClient = client ?? throw new ArgumentNullException(nameof(client)); + SetInputOutputStream(); + } + + public TSocketTransport(IPAddress host, int port) + : this(host, port, 0) + { + } + + public TSocketTransport(IPAddress host, int port, int timeout) + { + Host = host; + Port = port; + + TcpClient = new TcpClient(); + TcpClient.ReceiveTimeout = TcpClient.SendTimeout = timeout; + TcpClient.Client.NoDelay = true; + SetInputOutputStream(); + } + + public TSocketTransport(string host, int port, int timeout = 0) + { + try + { + var entry = Dns.GetHostEntry(host); + if (entry.AddressList.Length == 0) + throw new TTransportException(TTransportException.ExceptionType.Unknown, "unable to resolve host name"); + + var addr = entry.AddressList[0]; + Host = new IPAddress(addr.GetAddressBytes(), addr.ScopeId); + Port = port; + + TcpClient = new TcpClient(host, port); + TcpClient.ReceiveTimeout = TcpClient.SendTimeout = timeout; + TcpClient.Client.NoDelay = true; + SetInputOutputStream(); + } + catch (SocketException e) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, e.Message, e); + } + } + + private void SetInputOutputStream() + { + if (IsOpen) + { + InputStream = TcpClient.GetStream(); + OutputStream = TcpClient.GetStream(); + } + } + + public TcpClient TcpClient { get; private set; } + public IPAddress Host { get; } + public int Port { get; } + + public int Timeout + { + set + { + if (TcpClient != null) + { + TcpClient.ReceiveTimeout = TcpClient.SendTimeout = value; + } + } + } + + public override bool IsOpen + { + get + { + return (TcpClient != null) && TcpClient.Connected; + } + } + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + + if (IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.AlreadyOpen, "Socket already connected"); + } + + if (Port <= 0) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "Cannot open without port"); + } + + if (TcpClient == null) + { + throw new InvalidOperationException("Invalid or not initialized tcp client"); + } + + await TcpClient.ConnectAsync(Host, Port); + SetInputOutputStream(); + } + + public override void Close() + { + base.Close(); + + if (TcpClient != null) + { + TcpClient.Dispose(); + TcpClient = null; + } + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + TcpClient?.Dispose(); + + base.Dispose(disposing); + } + } + _isDisposed = true; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs new file mode 100644 index 000000000..d8574d610 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TStreamTransport.cs @@ -0,0 +1,109 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + // ReSharper disable once InconsistentNaming + public class TStreamTransport : TTransport + { + private bool _isDisposed; + + protected TStreamTransport() + { + } + + public TStreamTransport(Stream inputStream, Stream outputStream) + { + InputStream = inputStream; + OutputStream = outputStream; + } + + protected Stream OutputStream { get; set; } + + protected Stream InputStream { get; set; } + + public override bool IsOpen => true; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override void Close() + { + if (InputStream != null) + { + InputStream.Dispose(); + InputStream = null; + } + + if (OutputStream != null) + { + OutputStream.Dispose(); + OutputStream = null; + } + } + + public override async ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (InputStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "Cannot read from null inputstream"); + } + + return await InputStream.ReadAsync(buffer, offset, length, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (OutputStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, + "Cannot write to null outputstream"); + } + + await OutputStream.WriteAsync(buffer, offset, length, cancellationToken); + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + await OutputStream.FlushAsync(cancellationToken); + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!_isDisposed) + { + if (disposing) + { + InputStream?.Dispose(); + OutputStream?.Dispose(); + } + } + _isDisposed = true; + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TTlsSocketTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TTlsSocketTransport.cs new file mode 100644 index 000000000..9295bb01b --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Client/TTlsSocketTransport.cs @@ -0,0 +1,267 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Net; +using System.Net.Security; +using System.Net.Sockets; +using System.Security.Authentication; +using System.Security.Cryptography.X509Certificates; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport.Client +{ + //TODO: check for correct work + + // ReSharper disable once InconsistentNaming + public class TTlsSocketTransport : TStreamTransport + { + private readonly X509Certificate2 _certificate; + private readonly RemoteCertificateValidationCallback _certValidator; + private readonly IPAddress _host; + private readonly bool _isServer; + private readonly LocalCertificateSelectionCallback _localCertificateSelectionCallback; + private readonly int _port; + private readonly SslProtocols _sslProtocols; + private TcpClient _client; + private SslStream _secureStream; + private int _timeout; + + public TTlsSocketTransport(TcpClient client, X509Certificate2 certificate, bool isServer = false, + RemoteCertificateValidationCallback certValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + { + _client = client; + _certificate = certificate; + _certValidator = certValidator; + _localCertificateSelectionCallback = localCertificateSelectionCallback; + _sslProtocols = sslProtocols; + _isServer = isServer; + + if (isServer && certificate == null) + { + throw new ArgumentException("TTlsSocketTransport needs certificate to be used for server", + nameof(certificate)); + } + + if (IsOpen) + { + InputStream = client.GetStream(); + OutputStream = client.GetStream(); + } + } + + public TTlsSocketTransport(IPAddress host, int port, string certificatePath, + RemoteCertificateValidationCallback certValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + : this(host, port, 0, + new X509Certificate2(certificatePath), + certValidator, + localCertificateSelectionCallback, + sslProtocols) + { + } + + public TTlsSocketTransport(IPAddress host, int port, + X509Certificate2 certificate = null, + RemoteCertificateValidationCallback certValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + : this(host, port, 0, + certificate, + certValidator, + localCertificateSelectionCallback, + sslProtocols) + { + } + + public TTlsSocketTransport(IPAddress host, int port, int timeout, + X509Certificate2 certificate, + RemoteCertificateValidationCallback certValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + { + _host = host; + _port = port; + _timeout = timeout; + _certificate = certificate; + _certValidator = certValidator; + _localCertificateSelectionCallback = localCertificateSelectionCallback; + _sslProtocols = sslProtocols; + + InitSocket(); + } + + public TTlsSocketTransport(string host, int port, int timeout, + X509Certificate2 certificate, + RemoteCertificateValidationCallback certValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + { + try + { + var entry = Dns.GetHostEntry(host); + if (entry.AddressList.Length == 0) + throw new TTransportException(TTransportException.ExceptionType.Unknown, "unable to resolve host name"); + + var addr = entry.AddressList[0]; + + _host = new IPAddress(addr.GetAddressBytes(), addr.ScopeId); + _port = port; + _timeout = timeout; + _certificate = certificate; + _certValidator = certValidator; + _localCertificateSelectionCallback = localCertificateSelectionCallback; + _sslProtocols = sslProtocols; + + InitSocket(); + } + catch (SocketException e) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, e.Message, e); + } + } + + public int Timeout + { + set { _client.ReceiveTimeout = _client.SendTimeout = _timeout = value; } + } + + public TcpClient TcpClient => _client; + + public IPAddress Host => _host; + + public int Port => _port; + + public override bool IsOpen + { + get + { + if (_client == null) + { + return false; + } + + return _client.Connected; + } + } + + private void InitSocket() + { + _client = new TcpClient(); + _client.ReceiveTimeout = _client.SendTimeout = _timeout; + _client.Client.NoDelay = true; + } + + private bool DefaultCertificateValidator(object sender, X509Certificate certificate, X509Chain chain, + SslPolicyErrors sslValidationErrors) + { + return sslValidationErrors == SslPolicyErrors.None; + } + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.AlreadyOpen, "Socket already connected"); + } + + if (_host == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "Cannot open null host"); + } + + if (_port <= 0) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "Cannot open without port"); + } + + if (_client == null) + { + InitSocket(); + } + + if (_client != null) + { + await _client.ConnectAsync(_host, _port); + await SetupTlsAsync(); + } + } + + public async Task SetupTlsAsync() + { + var validator = _certValidator ?? DefaultCertificateValidator; + + if (_localCertificateSelectionCallback != null) + { + _secureStream = new SslStream(_client.GetStream(), false, validator, _localCertificateSelectionCallback); + } + else + { + _secureStream = new SslStream(_client.GetStream(), false, validator); + } + + try + { + if (_isServer) + { + // Server authentication + await + _secureStream.AuthenticateAsServerAsync(_certificate, _certValidator != null, _sslProtocols, + true); + } + else + { + // Client authentication + var certs = _certificate != null + ? new X509CertificateCollection {_certificate} + : new X509CertificateCollection(); + + var targetHost = _host.ToString(); + await _secureStream.AuthenticateAsClientAsync(targetHost, certs, _sslProtocols, true); + } + } + catch (Exception) + { + Close(); + throw; + } + + InputStream = _secureStream; + OutputStream = _secureStream; + } + + public override void Close() + { + base.Close(); + if (_client != null) + { + _client.Dispose(); + _client = null; + } + + if (_secureStream != null) + { + _secureStream.Dispose(); + _secureStream = null; + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/NullLogger.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/NullLogger.cs new file mode 100644 index 000000000..1f1f542d5 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/NullLogger.cs @@ -0,0 +1,56 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using Microsoft.Extensions.Logging; +using System; + + +namespace Thrift.Transport.Server +{ + // sometimes we just don't want to log anything + internal class NullLogger<T> : IDisposable, ILogger, ILogger<T> + { + internal class NullScope : IDisposable + { + public void Dispose() + { + // nothing to do + } + } + + public IDisposable BeginScope<TState>(TState state) + { + return new NullScope(); + } + + public void Dispose() + { + // nothing to do + } + + public bool IsEnabled(LogLevel logLevel) + { + return false; // no + } + + public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func<TState, Exception, string> formatter) + { + // do nothing + } + } +} + diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/THttpServerTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/THttpServerTransport.cs new file mode 100644 index 000000000..056300cfe --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/THttpServerTransport.cs @@ -0,0 +1,118 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.Logging; +using Thrift.Processor; +using Thrift.Protocol; +using Thrift.Transport.Client; + +namespace Thrift.Transport.Server +{ + // ReSharper disable once InconsistentNaming + public class THttpServerTransport + { + protected const string ContentType = "application/x-thrift"; + private readonly ILogger _logger; + private readonly RequestDelegate _next; + protected Encoding Encoding = Encoding.UTF8; + + protected TProtocolFactory InputProtocolFactory; + protected TProtocolFactory OutputProtocolFactory; + + protected TTransportFactory InputTransportFactory; + protected TTransportFactory OutputTransportFactory; + + protected ITAsyncProcessor Processor; + + public THttpServerTransport(ITAsyncProcessor processor, RequestDelegate next = null, ILoggerFactory loggerFactory = null) + : this(processor, new TBinaryProtocol.Factory(), null, next, loggerFactory) + { + } + + public THttpServerTransport( + ITAsyncProcessor processor, + TProtocolFactory protocolFactory, + TTransportFactory transFactory = null, + RequestDelegate next = null, + ILoggerFactory loggerFactory = null) + : this(processor, protocolFactory, protocolFactory, transFactory, transFactory, next, loggerFactory) + { + } + + public THttpServerTransport( + ITAsyncProcessor processor, + TProtocolFactory inputProtocolFactory, + TProtocolFactory outputProtocolFactory, + TTransportFactory inputTransFactory = null, + TTransportFactory outputTransFactory = null, + RequestDelegate next = null, + ILoggerFactory loggerFactory = null) + { + // loggerFactory == null is not illegal anymore + + Processor = processor ?? throw new ArgumentNullException(nameof(processor)); + InputProtocolFactory = inputProtocolFactory ?? throw new ArgumentNullException(nameof(inputProtocolFactory)); + OutputProtocolFactory = outputProtocolFactory ?? throw new ArgumentNullException(nameof(outputProtocolFactory)); + + InputTransportFactory = inputTransFactory; + OutputTransportFactory = outputTransFactory; + + _next = next; + _logger = (loggerFactory != null) ? loggerFactory.CreateLogger<THttpServerTransport>() : new NullLogger<THttpServerTransport>(); + } + + public async Task Invoke(HttpContext context) + { + context.Response.ContentType = ContentType; + await ProcessRequestAsync(context, context.RequestAborted); //TODO: check for correct logic + } + + public async Task ProcessRequestAsync(HttpContext context, CancellationToken cancellationToken) + { + var transport = new TStreamTransport(context.Request.Body, context.Response.Body); + + try + { + var intrans = (InputTransportFactory != null) ? InputTransportFactory.GetTransport(transport) : transport; + var outtrans = (OutputTransportFactory != null) ? OutputTransportFactory.GetTransport(transport) : transport; + + var input = InputProtocolFactory.GetProtocol(intrans); + var output = OutputProtocolFactory.GetProtocol(outtrans); + + while (await Processor.ProcessAsync(input, output, cancellationToken)) + { + if (!context.Response.HasStarted) // oneway method called + await context.Response.Body.FlushAsync(cancellationToken); + } + } + catch (TTransportException) + { + if (!context.Response.HasStarted) // if something goes bust, let the client know + context.Response.StatusCode = 500; + } + finally + { + transport.Close(); + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs new file mode 100644 index 000000000..77b825143 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TNamedPipeServerTransport.cs @@ -0,0 +1,308 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using Microsoft.Win32.SafeHandles; +using System; +using System.IO.Pipes; +using System.Runtime.InteropServices; +using System.Threading; +using System.Threading.Tasks; +using System.ComponentModel; +using System.Security.AccessControl; +using System.Security.Principal; + +namespace Thrift.Transport.Server +{ + // ReSharper disable once InconsistentNaming + public class TNamedPipeServerTransport : TServerTransport + { + /// <summary> + /// This is the address of the Pipe on the localhost. + /// </summary> + private readonly string _pipeAddress; + private bool _asyncMode = true; + private volatile bool _isPending = true; + private NamedPipeServerStream _stream = null; + + public TNamedPipeServerTransport(string pipeAddress) + { + _pipeAddress = pipeAddress; + } + + public override void Listen() + { + // nothing to do here + } + + public override void Close() + { + if (_stream != null) + { + try + { + if (_stream.IsConnected) + _stream.Disconnect(); + _stream.Dispose(); + } + finally + { + _stream = null; + _isPending = false; + } + } + } + + public override bool IsClientPending() + { + return _isPending; + } + + private void EnsurePipeInstance() + { + if (_stream == null) + { + const PipeDirection direction = PipeDirection.InOut; + const int maxconn = NamedPipeServerStream.MaxAllowedServerInstances; + const PipeTransmissionMode mode = PipeTransmissionMode.Byte; + const int inbuf = 4096; + const int outbuf = 4096; + var options = _asyncMode ? PipeOptions.Asynchronous : PipeOptions.None; + + + // TODO: "CreatePipeNative" ist only a workaround, and there are have basically two possible outcomes: + // - once NamedPipeServerStream() gets a CTOR that supports pipesec, remove CreatePipeNative() + // - if 31190 gets resolved before, use _stream.SetAccessControl(pipesec) instead of CreatePipeNative() + // EITHER WAY, + // - if CreatePipeNative() finally gets removed, also remove "allow unsafe code" from the project settings + + try + { + var handle = CreatePipeNative(_pipeAddress, inbuf, outbuf); + if( (handle != null) && (!handle.IsInvalid)) + _stream = new NamedPipeServerStream(PipeDirection.InOut, _asyncMode, false, handle); + else + _stream = new NamedPipeServerStream(_pipeAddress, direction, maxconn, mode, options, inbuf, outbuf/*, pipesec*/); + } + catch (NotImplementedException) // Mono still does not support async, fallback to sync + { + if (_asyncMode) + { + options &= (~PipeOptions.Asynchronous); + _stream = new NamedPipeServerStream(_pipeAddress, direction, maxconn, mode, options, inbuf, outbuf); + _asyncMode = false; + } + else + { + throw; + } + } + } + } + + + #region CreatePipeNative workaround + + + [StructLayout(LayoutKind.Sequential)] + internal class SECURITY_ATTRIBUTES + { + internal int nLength = 0; + internal IntPtr lpSecurityDescriptor = IntPtr.Zero; + internal int bInheritHandle = 0; + } + + + private const string Kernel32 = "kernel32.dll"; + + [DllImport(Kernel32, SetLastError = true)] + internal static extern IntPtr CreateNamedPipe( + string lpName, uint dwOpenMode, uint dwPipeMode, + uint nMaxInstances, uint nOutBufferSize, uint nInBufferSize, uint nDefaultTimeOut, + SECURITY_ATTRIBUTES pipeSecurityDescriptor + ); + + + + // Workaround: create the pipe via API call + // we have to do it this way, since NamedPipeServerStream() for netstd still lacks a few CTORs + // and _stream.SetAccessControl(pipesec); only keeps throwing ACCESS_DENIED errors at us + // References: + // - https://github.com/dotnet/corefx/issues/30170 (closed, continued in 31190) + // - https://github.com/dotnet/corefx/issues/31190 System.IO.Pipes.AccessControl package does not work + // - https://github.com/dotnet/corefx/issues/24040 NamedPipeServerStream: Provide support for WRITE_DAC + // - https://github.com/dotnet/corefx/issues/34400 Have a mechanism for lower privileged user to connect to a privileged user's pipe + private SafePipeHandle CreatePipeNative(string name, int inbuf, int outbuf) + { + if (Environment.OSVersion.Platform != PlatformID.Win32NT) + return null; // Windows only + + var pinningHandle = new GCHandle(); + try + { + // owner gets full access, everyone else read/write + var pipesec = new PipeSecurity(); + using (var currentIdentity = WindowsIdentity.GetCurrent()) + { + var sidOwner = currentIdentity.Owner; + var sidWorld = new SecurityIdentifier(WellKnownSidType.WorldSid, null); + + pipesec.SetOwner(sidOwner); + pipesec.AddAccessRule(new PipeAccessRule(sidOwner, PipeAccessRights.FullControl, AccessControlType.Allow)); + pipesec.AddAccessRule(new PipeAccessRule(sidWorld, PipeAccessRights.ReadWrite, AccessControlType.Allow)); + } + + // create a security descriptor and assign it to the security attribs + var secAttrs = new SECURITY_ATTRIBUTES(); + byte[] sdBytes = pipesec.GetSecurityDescriptorBinaryForm(); + pinningHandle = GCHandle.Alloc(sdBytes, GCHandleType.Pinned); + unsafe { + fixed (byte* pSD = sdBytes) { + secAttrs.lpSecurityDescriptor = (IntPtr)pSD; + } + } + + // a bunch of constants we will need shortly + const int PIPE_ACCESS_DUPLEX = 0x00000003; + const int FILE_FLAG_OVERLAPPED = 0x40000000; + const int WRITE_DAC = 0x00040000; + const int PIPE_TYPE_BYTE = 0x00000000; + const int PIPE_READMODE_BYTE = 0x00000000; + const int PIPE_UNLIMITED_INSTANCES = 255; + + // create the pipe via API call + var rawHandle = CreateNamedPipe( + @"\\.\pipe\" + name, + PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED | WRITE_DAC, + PIPE_TYPE_BYTE | PIPE_READMODE_BYTE, + PIPE_UNLIMITED_INSTANCES, (uint)inbuf, (uint)outbuf, + 5 * 1000, + secAttrs + ); + + // make a SafePipeHandle() from it + var handle = new SafePipeHandle(rawHandle, true); + if (handle.IsInvalid) + throw new Win32Exception(Marshal.GetLastWin32Error()); + + // return it (to be packaged) + return handle; + } + finally + { + if (pinningHandle.IsAllocated) + pinningHandle.Free(); + } + } + + #endregion + + protected override async ValueTask<TTransport> AcceptImplementationAsync(CancellationToken cancellationToken) + { + try + { + EnsurePipeInstance(); + + await _stream.WaitForConnectionAsync(cancellationToken); + + var trans = new ServerTransport(_stream); + _stream = null; // pass ownership to ServerTransport + + //_isPending = false; + + return trans; + } + catch (TTransportException) + { + Close(); + throw; + } + catch (Exception e) + { + Close(); + throw new TTransportException(TTransportException.ExceptionType.NotOpen, e.Message); + } + } + + private class ServerTransport : TTransport + { + private readonly NamedPipeServerStream PipeStream; + + public ServerTransport(NamedPipeServerStream stream) + { + PipeStream = stream; + } + + public override bool IsOpen => PipeStream != null && PipeStream.IsConnected; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + public override void Close() + { + PipeStream?.Dispose(); + } + + public override async ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (PipeStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + return await PipeStream.ReadAsync(buffer, offset, length, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + if (PipeStream == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + // if necessary, send the data in chunks + // there's a system limit around 0x10000 bytes that we hit otherwise + // MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section." + var nBytes = Math.Min(15 * 4096, length); // 16 would exceed the limit + while (nBytes > 0) + { + await PipeStream.WriteAsync(buffer, offset, nBytes, cancellationToken); + offset += nBytes; + length -= nBytes; + nBytes = Math.Min(nBytes, length); + } + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + await Task.FromCanceled(cancellationToken); + } + } + + protected override void Dispose(bool disposing) + { + PipeStream?.Dispose(); + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TServerSocketTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TServerSocketTransport.cs new file mode 100644 index 000000000..86d82e3fc --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TServerSocketTransport.cs @@ -0,0 +1,139 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using Thrift.Transport.Client; + +namespace Thrift.Transport.Server +{ + + // ReSharper disable once InconsistentNaming + public class TServerSocketTransport : TServerTransport + { + private readonly int _clientTimeout; + private TcpListener _server; + + public TServerSocketTransport(TcpListener listener, int clientTimeout = 0) + { + _server = listener; + _clientTimeout = clientTimeout; + } + + public TServerSocketTransport(int port, int clientTimeout = 0) + : this(null, clientTimeout) + { + try + { + // Make server socket + _server = new TcpListener(IPAddress.Any, port); + _server.Server.NoDelay = true; + } + catch (Exception) + { + _server = null; + throw new TTransportException("Could not create ServerSocket on port " + port + "."); + } + } + + public override void Listen() + { + // Make sure not to block on accept + if (_server != null) + { + try + { + _server.Start(); + } + catch (SocketException sx) + { + throw new TTransportException("Could not accept on listening socket: " + sx.Message); + } + } + } + + public override bool IsClientPending() + { + return _server.Pending(); + } + + protected override async ValueTask<TTransport> AcceptImplementationAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TTransport>(cancellationToken); + } + + if (_server == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "No underlying server socket."); + } + + try + { + TTransport tSocketTransport = null; + var tcpClient = await _server.AcceptTcpClientAsync(); + + try + { + tSocketTransport = new TSocketTransport(tcpClient) + { + Timeout = _clientTimeout + }; + + return tSocketTransport; + } + catch (Exception) + { + if (tSocketTransport != null) + { + tSocketTransport.Dispose(); + } + else // Otherwise, clean it up ourselves. + { + ((IDisposable) tcpClient).Dispose(); + } + + throw; + } + } + catch (Exception ex) + { + throw new TTransportException(ex.ToString()); + } + } + + public override void Close() + { + if (_server != null) + { + try + { + _server.Stop(); + } + catch (Exception ex) + { + throw new TTransportException("WARNING: Could not close server socket: " + ex); + } + _server = null; + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TTlsServerSocketTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TTlsServerSocketTransport.cs new file mode 100644 index 000000000..128680599 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/Server/TTlsServerSocketTransport.cs @@ -0,0 +1,150 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Net; +using System.Net.Security; +using System.Net.Sockets; +using System.Security.Authentication; +using System.Security.Cryptography.X509Certificates; +using System.Threading; +using System.Threading.Tasks; +using Thrift.Transport.Client; + +namespace Thrift.Transport.Server +{ + // ReSharper disable once InconsistentNaming + public class TTlsServerSocketTransport : TServerTransport + { + private readonly RemoteCertificateValidationCallback _clientCertValidator; + private readonly int _clientTimeout = 0; + private readonly LocalCertificateSelectionCallback _localCertificateSelectionCallback; + private readonly X509Certificate2 _serverCertificate; + private readonly SslProtocols _sslProtocols; + private TcpListener _server; + + public TTlsServerSocketTransport( + TcpListener listener, + X509Certificate2 certificate, + RemoteCertificateValidationCallback clientCertValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + { + if (!certificate.HasPrivateKey) + { + throw new TTransportException(TTransportException.ExceptionType.Unknown, + "Your server-certificate needs to have a private key"); + } + + _serverCertificate = certificate; + _clientCertValidator = clientCertValidator; + _localCertificateSelectionCallback = localCertificateSelectionCallback; + _sslProtocols = sslProtocols; + _server = listener; + } + + public TTlsServerSocketTransport( + int port, + X509Certificate2 certificate, + RemoteCertificateValidationCallback clientCertValidator = null, + LocalCertificateSelectionCallback localCertificateSelectionCallback = null, + SslProtocols sslProtocols = SslProtocols.Tls12) + : this(null, certificate, clientCertValidator, localCertificateSelectionCallback) + { + try + { + // Create server socket + _server = new TcpListener(IPAddress.Any, port); + _server.Server.NoDelay = true; + } + catch (Exception) + { + _server = null; + throw new TTransportException($"Could not create ServerSocket on port {port}."); + } + } + + public override void Listen() + { + // Make sure accept is not blocking + if (_server != null) + { + try + { + _server.Start(); + } + catch (SocketException sx) + { + throw new TTransportException($"Could not accept on listening socket: {sx.Message}"); + } + } + } + + public override bool IsClientPending() + { + return _server.Pending(); + } + + protected override async ValueTask<TTransport> AcceptImplementationAsync(CancellationToken cancellationToken) + { + if (cancellationToken.IsCancellationRequested) + { + return await Task.FromCanceled<TTransport>(cancellationToken); + } + + if (_server == null) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen, "No underlying server socket."); + } + + try + { + var client = await _server.AcceptTcpClientAsync(); + client.SendTimeout = client.ReceiveTimeout = _clientTimeout; + + //wrap the client in an SSL Socket passing in the SSL cert + var tTlsSocket = new TTlsSocketTransport(client, _serverCertificate, true, _clientCertValidator, + _localCertificateSelectionCallback, _sslProtocols); + + await tTlsSocket.SetupTlsAsync(); + + return tTlsSocket; + } + catch (Exception ex) + { + throw new TTransportException(ex.ToString()); + } + } + + public override void Close() + { + if (_server != null) + { + try + { + _server.Stop(); + } + catch (Exception ex) + { + throw new TTransportException($"WARNING: Could not close server socket: {ex}"); + } + + _server = null; + } + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TBufferedTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TBufferedTransport.cs new file mode 100644 index 000000000..e4fdd3a8d --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TBufferedTransport.cs @@ -0,0 +1,198 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Diagnostics; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport +{ + // ReSharper disable once InconsistentNaming + public class TBufferedTransport : TTransport + { + private readonly int DesiredBufferSize; + private readonly Client.TMemoryBufferTransport ReadBuffer = new Client.TMemoryBufferTransport(1024); + private readonly Client.TMemoryBufferTransport WriteBuffer = new Client.TMemoryBufferTransport(1024); + private readonly TTransport InnerTransport; + private bool IsDisposed; + + public class Factory : TTransportFactory + { + public override TTransport GetTransport(TTransport trans) + { + return new TBufferedTransport(trans); + } + } + + //TODO: should support only specified input transport? + public TBufferedTransport(TTransport transport, int bufSize = 1024) + { + if (bufSize <= 0) + { + throw new ArgumentOutOfRangeException(nameof(bufSize), "Buffer size must be a positive number."); + } + + InnerTransport = transport ?? throw new ArgumentNullException(nameof(transport)); + DesiredBufferSize = bufSize; + + if (DesiredBufferSize != ReadBuffer.Capacity) + ReadBuffer.Capacity = DesiredBufferSize; + if (DesiredBufferSize != WriteBuffer.Capacity) + WriteBuffer.Capacity = DesiredBufferSize; + } + + public TTransport UnderlyingTransport + { + get + { + CheckNotDisposed(); + + return InnerTransport; + } + } + + public override bool IsOpen => !IsDisposed && InnerTransport.IsOpen; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + CheckNotDisposed(); + + await InnerTransport.OpenAsync(cancellationToken); + } + + public override void Close() + { + CheckNotDisposed(); + + InnerTransport.Close(); + } + + public override async ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + CheckNotDisposed(); + ValidateBufferArgs(buffer, offset, length); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + + // do we have something buffered? + var count = ReadBuffer.Length - ReadBuffer.Position; + if (count > 0) + { + return await ReadBuffer.ReadAsync(buffer, offset, length, cancellationToken); + } + + // does the request even fit into the buffer? + // Note we test for >= instead of > to avoid nonsense buffering + if (length >= ReadBuffer.Capacity) + { + return await InnerTransport.ReadAsync(buffer, offset, length, cancellationToken); + } + + // buffer a new chunk of bytes from the underlying transport + ReadBuffer.Length = ReadBuffer.Capacity; + ArraySegment<byte> bufSegment; + ReadBuffer.TryGetBuffer(out bufSegment); + ReadBuffer.Length = await InnerTransport.ReadAsync(bufSegment.Array, 0, bufSegment.Count, cancellationToken); + ReadBuffer.Position = 0; + + // deliver the bytes + return await ReadBuffer.ReadAsync(buffer, offset, length, cancellationToken); + } + + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + CheckNotDisposed(); + ValidateBufferArgs(buffer, offset, length); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + // enough space left in buffer? + var free = WriteBuffer.Capacity - WriteBuffer.Length; + if (length > free) + { + ArraySegment<byte> bufSegment; + WriteBuffer.TryGetBuffer(out bufSegment); + await InnerTransport.WriteAsync(bufSegment.Array, 0, bufSegment.Count, cancellationToken); + WriteBuffer.SetLength(0); + } + + // do the data even fit into the buffer? + // Note we test for < instead of <= to avoid nonsense buffering + if (length < WriteBuffer.Capacity) + { + await WriteBuffer.WriteAsync(buffer, offset, length, cancellationToken); + return; + } + + // write thru + await InnerTransport.WriteAsync(buffer, offset, length, cancellationToken); + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + CheckNotDisposed(); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + if (WriteBuffer.Length > 0) + { + ArraySegment<byte> bufSegment; + WriteBuffer.TryGetBuffer(out bufSegment); + await InnerTransport.WriteAsync(bufSegment.Array, 0, bufSegment.Count, cancellationToken); + WriteBuffer.SetLength(0); + } + + await InnerTransport.FlushAsync(cancellationToken); + } + + private void CheckNotDisposed() + { + if (IsDisposed) + { + throw new ObjectDisposedException(nameof(InnerTransport)); + } + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!IsDisposed) + { + if (disposing) + { + ReadBuffer?.Dispose(); + WriteBuffer?.Dispose(); + InnerTransport?.Dispose(); + } + } + IsDisposed = true; + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TFramedTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TFramedTransport.cs new file mode 100644 index 000000000..de6df7238 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TFramedTransport.cs @@ -0,0 +1,194 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport +{ + // ReSharper disable once InconsistentNaming + public class TFramedTransport : TTransport + { + private const int HeaderSize = 4; + private readonly byte[] HeaderBuf = new byte[HeaderSize]; + private readonly Client.TMemoryBufferTransport ReadBuffer = new Client.TMemoryBufferTransport(); + private readonly Client.TMemoryBufferTransport WriteBuffer = new Client.TMemoryBufferTransport(); + private readonly TTransport InnerTransport; + + private bool IsDisposed; + + public class Factory : TTransportFactory + { + public override TTransport GetTransport(TTransport trans) + { + return new TFramedTransport(trans); + } + } + + public TFramedTransport(TTransport transport) + { + InnerTransport = transport ?? throw new ArgumentNullException(nameof(transport)); + + InitWriteBuffer(); + } + + public override bool IsOpen => !IsDisposed && InnerTransport.IsOpen; + + public override async Task OpenAsync(CancellationToken cancellationToken) + { + CheckNotDisposed(); + + await InnerTransport.OpenAsync(cancellationToken); + } + + public override void Close() + { + CheckNotDisposed(); + + InnerTransport.Close(); + } + + public override async ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + CheckNotDisposed(); + ValidateBufferArgs(buffer, offset, length); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + // Read another frame of data if we run out of bytes + if (ReadBuffer.Position >= ReadBuffer.Length) + { + await ReadFrameAsync(cancellationToken); + } + + return await ReadBuffer.ReadAsync(buffer, offset, length, cancellationToken); + } + + private async ValueTask ReadFrameAsync(CancellationToken cancellationToken) + { + await InnerTransport.ReadAllAsync(HeaderBuf, 0, HeaderSize, cancellationToken); + var size = DecodeFrameSize(HeaderBuf); + + ReadBuffer.SetLength(size); + ReadBuffer.Seek(0, SeekOrigin.Begin); + + ArraySegment<byte> bufSegment; + ReadBuffer.TryGetBuffer(out bufSegment); + await InnerTransport.ReadAllAsync(bufSegment.Array, 0, size, cancellationToken); + } + + public override async Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + CheckNotDisposed(); + ValidateBufferArgs(buffer, offset, length); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + if (WriteBuffer.Length > (int.MaxValue - length)) + { + await FlushAsync(cancellationToken); + } + + await WriteBuffer.WriteAsync(buffer, offset, length, cancellationToken); + } + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + CheckNotDisposed(); + + if (!IsOpen) + { + throw new TTransportException(TTransportException.ExceptionType.NotOpen); + } + + ArraySegment<byte> bufSegment; + WriteBuffer.TryGetBuffer(out bufSegment); + + int dataLen = bufSegment.Count - HeaderSize; + if (dataLen < 0) + { + throw new InvalidOperationException(); // logic error actually + } + + // Inject message header into the reserved buffer space + EncodeFrameSize(dataLen, bufSegment.Array); + + // Send the entire message at once + await InnerTransport.WriteAsync(bufSegment.Array, 0, bufSegment.Count, cancellationToken); + + InitWriteBuffer(); + + await InnerTransport.FlushAsync(cancellationToken); + } + + private void InitWriteBuffer() + { + // Reserve space for message header to be put right before sending it out + WriteBuffer.SetLength(HeaderSize); + WriteBuffer.Seek(0, SeekOrigin.End); + } + + private static void EncodeFrameSize(int frameSize, byte[] buf) + { + buf[0] = (byte) (0xff & (frameSize >> 24)); + buf[1] = (byte) (0xff & (frameSize >> 16)); + buf[2] = (byte) (0xff & (frameSize >> 8)); + buf[3] = (byte) (0xff & (frameSize)); + } + + private static int DecodeFrameSize(byte[] buf) + { + return + ((buf[0] & 0xff) << 24) | + ((buf[1] & 0xff) << 16) | + ((buf[2] & 0xff) << 8) | + (buf[3] & 0xff); + } + + + private void CheckNotDisposed() + { + if (IsDisposed) + { + throw new ObjectDisposedException(this.GetType().Name); + } + } + + // IDisposable + protected override void Dispose(bool disposing) + { + if (!IsDisposed) + { + if (disposing) + { + ReadBuffer?.Dispose(); + WriteBuffer?.Dispose(); + InnerTransport?.Dispose(); + } + } + IsDisposed = true; + } + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TServerTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TServerTransport.cs new file mode 100644 index 000000000..74c54cd0e --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TServerTransport.cs @@ -0,0 +1,54 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport +{ + // ReSharper disable once InconsistentNaming + public abstract class TServerTransport + { + public abstract void Listen(); + public abstract void Close(); + public abstract bool IsClientPending(); + + protected virtual async ValueTask<TTransport> AcceptImplementationAsync() + { + return await AcceptImplementationAsync(CancellationToken.None); + } + + protected abstract ValueTask<TTransport> AcceptImplementationAsync(CancellationToken cancellationToken); + + public async ValueTask<TTransport> AcceptAsync() + { + return await AcceptAsync(CancellationToken.None); + } + + public async ValueTask<TTransport> AcceptAsync(CancellationToken cancellationToken) + { + var transport = await AcceptImplementationAsync(cancellationToken); + + if (transport == null) + { + throw new TTransportException($"{nameof(AcceptImplementationAsync)} should not return null"); + } + + return transport; + } + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransport.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransport.cs new file mode 100644 index 000000000..799801202 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransport.cs @@ -0,0 +1,190 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; +using System.Diagnostics; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Thrift.Transport +{ + //TODO: think about client info + // ReSharper disable once InconsistentNaming + public abstract class TTransport : IDisposable + { + //TODO: think how to avoid peek byte + private readonly byte[] _peekBuffer = new byte[1]; + private bool _hasPeekByte; + public abstract bool IsOpen { get; } + + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + public async ValueTask<bool> PeekAsync(CancellationToken cancellationToken) + { + //If we already have a byte read but not consumed, do nothing. + if (_hasPeekByte) + { + return true; + } + + //If transport closed we can't peek. + if (!IsOpen) + { + return false; + } + + //Try to read one byte. If succeeds we will need to store it for the next read. + try + { + var bytes = await ReadAsync(_peekBuffer, 0, 1, cancellationToken); + if (bytes == 0) + { + return false; + } + } + catch (IOException) + { + return false; + } + + _hasPeekByte = true; + return true; + } + + public virtual async Task OpenAsync() + { + await OpenAsync(CancellationToken.None); + } + + public abstract Task OpenAsync(CancellationToken cancellationToken); + + public abstract void Close(); + + protected static void ValidateBufferArgs(byte[] buffer, int offset, int length) + { + if (buffer == null) + { + throw new ArgumentNullException(nameof(buffer)); + } + +#if DEBUG // let it fail with OutOfRange in RELEASE mode + if (offset < 0) + { + throw new ArgumentOutOfRangeException(nameof(offset), "Buffer offset must be >= 0"); + } + + if (length < 0) + { + throw new ArgumentOutOfRangeException(nameof(length), "Buffer length must be >= 0"); + } + + if (offset + length > buffer.Length) + { + throw new ArgumentOutOfRangeException(nameof(buffer), "Not enough data"); + } +#endif + } + + public virtual async ValueTask<int> ReadAsync(byte[] buffer, int offset, int length) + { + return await ReadAsync(buffer, offset, length, CancellationToken.None); + } + + public abstract ValueTask<int> ReadAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken); + + public virtual async ValueTask<int> ReadAllAsync(byte[] buffer, int offset, int length) + { + return await ReadAllAsync(buffer, offset, length, CancellationToken.None); + } + + public virtual async ValueTask<int> ReadAllAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken) + { + ValidateBufferArgs(buffer, offset, length); + + if (cancellationToken.IsCancellationRequested) + return await Task.FromCanceled<int>(cancellationToken); + + if (length <= 0) + return 0; + + // If we previously peeked a byte, we need to use that first. + var totalBytes = 0; + if (_hasPeekByte) + { + buffer[offset++] = _peekBuffer[0]; + _hasPeekByte = false; + if (1 == length) + { + return 1; // we're done + } + ++totalBytes; + } + + var remaining = length - totalBytes; + Debug.Assert(remaining > 0); // any other possible cases should have been handled already + while (true) + { + var numBytes = await ReadAsync(buffer, offset, remaining, cancellationToken); + totalBytes += numBytes; + if (totalBytes >= length) + { + return totalBytes; // we're done + } + + if (numBytes <= 0) + { + throw new TTransportException(TTransportException.ExceptionType.EndOfFile, + "Cannot read, Remote side has closed"); + } + + remaining -= numBytes; + offset += numBytes; + } + } + + public virtual async Task WriteAsync(byte[] buffer) + { + await WriteAsync(buffer, CancellationToken.None); + } + + public virtual async Task WriteAsync(byte[] buffer, CancellationToken cancellationToken) + { + await WriteAsync(buffer, 0, buffer.Length, CancellationToken.None); + } + + public virtual async Task WriteAsync(byte[] buffer, int offset, int length) + { + await WriteAsync(buffer, offset, length, CancellationToken.None); + } + + public abstract Task WriteAsync(byte[] buffer, int offset, int length, CancellationToken cancellationToken); + + public virtual async Task FlushAsync() + { + await FlushAsync(CancellationToken.None); + } + + public abstract Task FlushAsync(CancellationToken cancellationToken); + + protected abstract void Dispose(bool disposing); + } +} diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransportException.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransportException.cs new file mode 100644 index 000000000..760a178e6 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransportException.cs @@ -0,0 +1,60 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System; + +namespace Thrift.Transport +{ + // ReSharper disable once InconsistentNaming + public class TTransportException : TException + { + public enum ExceptionType + { + Unknown, + NotOpen, + AlreadyOpen, + TimedOut, + EndOfFile, + Interrupted + } + + public ExceptionType ExType { get; private set; } + + public TTransportException() + { + } + + public TTransportException(ExceptionType exType, Exception inner = null) + : base(string.Empty, inner) + { + ExType = exType; + } + + public TTransportException(ExceptionType exType, string message, Exception inner = null) + : base(message, inner) + { + ExType = exType; + } + + public TTransportException(string message, Exception inner = null) + : base(message, inner) + { + } + + public ExceptionType Type => ExType; + } +}
\ No newline at end of file diff --git a/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransportFactory.cs b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransportFactory.cs new file mode 100644 index 000000000..16e27ac82 --- /dev/null +++ b/src/jaegertracing/thrift/lib/netstd/Thrift/Transport/TTransportFactory.cs @@ -0,0 +1,35 @@ +// Licensed to the Apache Software Foundation(ASF) under one +// or more contributor license agreements.See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership.The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +namespace Thrift.Transport +{ + /// <summary> + /// From Mark Slee & Aditya Agarwal of Facebook: + /// Factory class used to create wrapped instance of Transports. + /// This is used primarily in servers, which get Transports from + /// a ServerTransport and then may want to mutate them (i.e. create + /// a BufferedTransport from the underlying base transport) + /// </summary> + // ReSharper disable once InconsistentNaming + public class TTransportFactory + { + public virtual TTransport GetTransport(TTransport trans) + { + return trans; + } + } +} |